]> git.ipfire.org Git - thirdparty/sqlalchemy/sqlalchemy.git/commitdiff
Add missing tests for postgresql
authorRobotScribe <quentinso@theodo.fr>
Wed, 29 Apr 2020 10:35:46 +0000 (12:35 +0200)
committerRobotScribe <quentinso@theodo.fr>
Wed, 29 Apr 2020 12:45:54 +0000 (14:45 +0200)
test/dialect/postgresql/test_compiler.py

index 4cc9c837d6eacf1960c0dc1d51135b5d19c025ca..c707137a819c163785ddcfe4c07d8c061d954032 100644 (file)
@@ -950,6 +950,24 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
             "FROM mytable WHERE mytable.myid = %(myid_1)s FOR SHARE NOWAIT",
         )
 
+        self.assert_compile(
+            table1.select(table1.c.myid == 7).with_for_update(
+                key_share=True, nowait=True
+            ),
+            "SELECT mytable.myid, mytable.name, mytable.description "
+            "FROM mytable WHERE mytable.myid = %(myid_1)s "
+            "FOR NO KEY UPDATE NOWAIT",
+        )
+
+        self.assert_compile(
+            table1.select(table1.c.myid == 7).with_for_update(
+                key_share=True, read=True, nowait=True
+            ),
+            "SELECT mytable.myid, mytable.name, mytable.description "
+            "FROM mytable WHERE mytable.myid = %(myid_1)s "
+            "FOR KEY SHARE NOWAIT",
+        )
+
         self.assert_compile(
             table1.select(table1.c.myid == 7).with_for_update(
                 read=True, skip_locked=True
@@ -977,6 +995,15 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
             "FOR SHARE OF mytable NOWAIT",
         )
 
+        self.assert_compile(
+            table1.select(table1.c.myid == 7).with_for_update(
+                key_share=True, read=True, nowait=True, of=table1
+            ),
+            "SELECT mytable.myid, mytable.name, mytable.description "
+            "FROM mytable WHERE mytable.myid = %(myid_1)s "
+            "FOR KEY SHARE OF mytable NOWAIT",
+        )
+
         self.assert_compile(
             table1.select(table1.c.myid == 7).with_for_update(
                 read=True, nowait=True, of=table1.c.myid
@@ -995,6 +1022,27 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
             "FOR SHARE OF mytable NOWAIT",
         )
 
+        self.assert_compile(
+            table1.select(table1.c.myid == 7).with_for_update(
+                read=True,
+                skip_locked=True,
+                of=[table1.c.myid, table1.c.name],
+                key_share=True,
+            ),
+            "SELECT mytable.myid, mytable.name, mytable.description "
+            "FROM mytable WHERE mytable.myid = %(myid_1)s "
+            "FOR KEY SHARE OF mytable SKIP LOCKED",
+        )
+
+        self.assert_compile(
+            table1.select(table1.c.myid == 7).with_for_update(
+                skip_locked=True, of=[table1.c.myid, table1.c.name]
+            ),
+            "SELECT mytable.myid, mytable.name, mytable.description "
+            "FROM mytable WHERE mytable.myid = %(myid_1)s "
+            "FOR UPDATE OF mytable SKIP LOCKED",
+        )
+
         self.assert_compile(
             table1.select(table1.c.myid == 7).with_for_update(
                 read=True, skip_locked=True, of=[table1.c.myid, table1.c.name]
@@ -1058,6 +1106,15 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
             "FOR KEY SHARE OF mytable",
         )
 
+        self.assert_compile(
+            table1.select(table1.c.myid == 7).with_for_update(
+                read=True, of=table1
+            ),
+            "SELECT mytable.myid, mytable.name, mytable.description "
+            "FROM mytable WHERE mytable.myid = %(myid_1)s "
+            "FOR SHARE OF mytable",
+        )
+
         self.assert_compile(
             table1.select(table1.c.myid == 7).with_for_update(
                 read=True, key_share=True, skip_locked=True
@@ -1067,6 +1124,15 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
             "FOR KEY SHARE SKIP LOCKED",
         )
 
+        self.assert_compile(
+            table1.select(table1.c.myid == 7).with_for_update(
+                key_share=True, skip_locked=True
+            ),
+            "SELECT mytable.myid, mytable.name, mytable.description "
+            "FROM mytable WHERE mytable.myid = %(myid_1)s "
+            "FOR NO KEY UPDATE SKIP LOCKED",
+        )
+
         ta = table1.alias()
         self.assert_compile(
             ta.select(ta.c.myid == 7).with_for_update(