summaryrefslogtreecommitdiff
path: root/test/sql/test_compiler.py
diff options
context:
space:
mode:
Diffstat (limited to 'test/sql/test_compiler.py')
-rw-r--r--test/sql/test_compiler.py86
1 files changed, 64 insertions, 22 deletions
diff --git a/test/sql/test_compiler.py b/test/sql/test_compiler.py
index 970030d55..c3cf001fa 100644
--- a/test/sql/test_compiler.py
+++ b/test/sql/test_compiler.py
@@ -1139,38 +1139,80 @@ class SelectTest(fixtures.TestBase, AssertsCompiledSQL):
def test_for_update(self):
self.assert_compile(
- table1.select(table1.c.myid==7, for_update=True),
- "SELECT mytable.myid, mytable.name, mytable.description "
- "FROM mytable WHERE mytable.myid = :myid_1 FOR UPDATE")
+ table1.select(table1.c.myid==7, for_update=True),
+ "SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable WHERE mytable.myid = :myid_1 FOR UPDATE")
self.assert_compile(
- table1.select(table1.c.myid==7, for_update="nowait"),
- "SELECT mytable.myid, mytable.name, mytable.description "
- "FROM mytable WHERE mytable.myid = :myid_1 FOR UPDATE")
+ table1.select(table1.c.myid==7, for_update=False),
+ "SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable WHERE mytable.myid = :myid_1")
+ # not supported by dialect, should just use update
self.assert_compile(
- table1.select(table1.c.myid==7, for_update="nowait"),
- "SELECT mytable.myid, mytable.name, mytable.description "
- "FROM mytable WHERE mytable.myid = :myid_1 FOR UPDATE NOWAIT",
- dialect=oracle.dialect())
+ table1.select(table1.c.myid==7, for_update='nowait'),
+ "SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable WHERE mytable.myid = :myid_1 FOR UPDATE")
+ # unknown lock mode
self.assert_compile(
- table1.select(table1.c.myid==7, for_update="read"),
- "SELECT mytable.myid, mytable.name, mytable.description "
- "FROM mytable WHERE mytable.myid = %s LOCK IN SHARE MODE",
- dialect=mysql.dialect())
+ table1.select(table1.c.myid==7, for_update='unknown_mode'),
+ "SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable WHERE mytable.myid = :myid_1 FOR UPDATE")
+
+ # ----- mysql
+
+ self.assert_compile(
+ table1.select(table1.c.myid==7, for_update=True),
+ "SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable WHERE mytable.myid = %s FOR UPDATE",
+ dialect=mysql.dialect())
+
+ self.assert_compile(
+ table1.select(table1.c.myid==7, for_update="read"),
+ "SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable WHERE mytable.myid = %s LOCK IN SHARE MODE",
+ dialect=mysql.dialect())
+
+ # ----- oracle
+
+ self.assert_compile(
+ table1.select(table1.c.myid==7, for_update=True),
+ "SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable WHERE mytable.myid = :myid_1 FOR UPDATE",
+ dialect=oracle.dialect())
+
+ self.assert_compile(
+ table1.select(table1.c.myid==7, for_update="nowait"),
+ "SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable WHERE mytable.myid = :myid_1 FOR UPDATE NOWAIT",
+ dialect=oracle.dialect())
+
+ # ----- postgresql
self.assert_compile(
- table1.select(table1.c.myid==7, for_update=True),
- "SELECT mytable.myid, mytable.name, mytable.description "
- "FROM mytable WHERE mytable.myid = %s FOR UPDATE",
- dialect=mysql.dialect())
+ table1.select(table1.c.myid==7, for_update=True),
+ "SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable WHERE mytable.myid = %(myid_1)s FOR UPDATE",
+ dialect=postgresql.dialect())
self.assert_compile(
- table1.select(table1.c.myid==7, for_update=True),
- "SELECT mytable.myid, mytable.name, mytable.description "
- "FROM mytable WHERE mytable.myid = :myid_1 FOR UPDATE",
- dialect=oracle.dialect())
+ table1.select(table1.c.myid==7, for_update="nowait"),
+ "SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable WHERE mytable.myid = %(myid_1)s FOR UPDATE NOWAIT",
+ dialect=postgresql.dialect())
+
+ self.assert_compile(
+ table1.select(table1.c.myid==7, for_update="read"),
+ "SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable WHERE mytable.myid = %(myid_1)s FOR SHARE",
+ dialect=postgresql.dialect())
+
+ self.assert_compile(
+ table1.select(table1.c.myid==7, for_update="read_nowait"),
+ "SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable WHERE mytable.myid = %(myid_1)s FOR SHARE NOWAIT",
+ dialect=postgresql.dialect())
def test_alias(self):
# test the alias for a table1. column names stay the same, table name "changes" to "foo".