summaryrefslogtreecommitdiff
path: root/test/dialect/postgresql/test_compiler.py
diff options
context:
space:
mode:
Diffstat (limited to 'test/dialect/postgresql/test_compiler.py')
-rw-r--r--test/dialect/postgresql/test_compiler.py54
1 files changed, 53 insertions, 1 deletions
diff --git a/test/dialect/postgresql/test_compiler.py b/test/dialect/postgresql/test_compiler.py
index c061cfaf1..c8dc9582a 100644
--- a/test/dialect/postgresql/test_compiler.py
+++ b/test/dialect/postgresql/test_compiler.py
@@ -1,7 +1,7 @@
# coding: utf-8
from sqlalchemy.testing.assertions import AssertsCompiledSQL, is_, \
- assert_raises
+ assert_raises, assert_raises_message
from sqlalchemy.testing import engines, fixtures
from sqlalchemy import testing
from sqlalchemy import Sequence, Table, Column, Integer, update, String,\
@@ -667,6 +667,58 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
"FROM mytable WHERE mytable.myid = %(myid_1)s "
"FOR SHARE OF mytable SKIP LOCKED")
+ self.assert_compile(
+ table1.select(table1.c.myid == 7).
+ with_for_update(key_share=True, nowait=True,
+ of=[table1.c.myid, table1.c.name]),
+ "SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable WHERE mytable.myid = %(myid_1)s "
+ "FOR NO KEY UPDATE OF mytable NOWAIT")
+
+ self.assert_compile(
+ table1.select(table1.c.myid == 7).
+ with_for_update(key_share=True, 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 NO KEY UPDATE OF mytable SKIP LOCKED")
+
+ self.assert_compile(
+ table1.select(table1.c.myid == 7).
+ with_for_update(key_share=True,
+ of=[table1.c.myid, table1.c.name]),
+ "SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable WHERE mytable.myid = %(myid_1)s "
+ "FOR NO KEY UPDATE OF mytable")
+
+ self.assert_compile(
+ table1.select(table1.c.myid == 7).
+ with_for_update(key_share=True),
+ "SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable WHERE mytable.myid = %(myid_1)s "
+ "FOR NO KEY UPDATE")
+
+ self.assert_compile(
+ table1.select(table1.c.myid == 7).
+ with_for_update(read=True, key_share=True),
+ "SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable WHERE mytable.myid = %(myid_1)s "
+ "FOR KEY SHARE")
+
+ self.assert_compile(
+ table1.select(table1.c.myid == 7).
+ with_for_update(read=True, key_share=True, of=table1),
+ "SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable WHERE mytable.myid = %(myid_1)s "
+ "FOR KEY SHARE OF mytable")
+
+ self.assert_compile(
+ table1.select(table1.c.myid == 7).
+ with_for_update(read=True, key_share=True, skip_locked=True),
+ "SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable WHERE mytable.myid = %(myid_1)s "
+ "FOR KEY SHARE SKIP LOCKED")
+
ta = table1.alias()
self.assert_compile(
ta.select(ta.c.myid == 7).