diff options
Diffstat (limited to 'test/sql/test_compiler.py')
| -rw-r--r-- | test/sql/test_compiler.py | 86 |
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". |
