diff options
Diffstat (limited to 'test')
| -rw-r--r-- | test/dialect/postgresql/test_compiler.py | 37 | ||||
| -rw-r--r-- | test/dialect/test_oracle.py | 6 | ||||
| -rw-r--r-- | test/orm/test_lockmode.py | 37 |
3 files changed, 77 insertions, 3 deletions
diff --git a/test/dialect/postgresql/test_compiler.py b/test/dialect/postgresql/test_compiler.py index 87e48d3f2..6ff1102e6 100644 --- a/test/dialect/postgresql/test_compiler.py +++ b/test/dialect/postgresql/test_compiler.py @@ -602,6 +602,43 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): "FROM mytable WHERE mytable.myid = %(myid_1)s " "FOR SHARE OF mytable NOWAIT") + self.assert_compile( + table1.select(table1.c.myid == 7). + with_for_update(no_key=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(no_key=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(no_key=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(no_key=True), + "SELECT mytable.myid, mytable.name, mytable.description " + "FROM mytable WHERE mytable.myid = %(myid_1)s " + "FOR NO KEY UPDATE") + + assert_raises( + exc.ArgumentError, + table1.select(table1.c.myid == 7).with_for_update, + no_key=True, + read=True + ) + ta = table1.alias() self.assert_compile( ta.select(ta.c.myid == 7). diff --git a/test/dialect/test_oracle.py b/test/dialect/test_oracle.py index 1f3e63040..b715667f9 100644 --- a/test/dialect/test_oracle.py +++ b/test/dialect/test_oracle.py @@ -334,6 +334,12 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): "FROM mytable WHERE mytable.myid = :myid_1 FOR UPDATE OF " "mytable.myid, mytable.name NOWAIT") + self.assert_compile( + table1.select(table1.c.myid == 7). + with_for_update(no_key=True), + "SELECT mytable.myid, mytable.name, mytable.description " + "FROM mytable WHERE mytable.myid = :myid_1 FOR UPDATE") + ta = table1.alias() self.assert_compile( ta.select(ta.c.myid == 7). diff --git a/test/orm/test_lockmode.py b/test/orm/test_lockmode.py index fc473a329..fa9588ac7 100644 --- a/test/orm/test_lockmode.py +++ b/test/orm/test_lockmode.py @@ -16,7 +16,7 @@ class LegacyLockModeTest(_fixtures.FixtureTest): User, users = cls.classes.User, cls.tables.users mapper(User, users) - def _assert_legacy(self, arg, read=False, nowait=False): + def _assert_legacy(self, arg, read=False, nowait=False, no_key=False): User = self.classes.User s = Session() q = s.query(User).with_lockmode(arg) @@ -29,9 +29,11 @@ class LegacyLockModeTest(_fixtures.FixtureTest): assert q._for_update_arg.read is read assert q._for_update_arg.nowait is nowait + assert q._for_update_arg.no_key is no_key assert sel._for_update_arg.read is read assert sel._for_update_arg.nowait is nowait + assert sel._for_update_arg.no_key is no_key def test_false_legacy(self): self._assert_legacy(None) @@ -42,6 +44,12 @@ class LegacyLockModeTest(_fixtures.FixtureTest): def test_nowait_legacy(self): self._assert_legacy("update_nowait", nowait=True) + def test_no_key_nowait_legacy(self): + self._assert_legacy("no_key_nowait", nowait=True, no_key=True) + + def test_no_key_legacy(self): + self._assert_legacy("no_key", no_key=True) + def test_read_legacy(self): self._assert_legacy("read", read=True) @@ -53,17 +61,18 @@ class LegacyLockModeTest(_fixtures.FixtureTest): sess.query(User.id).with_lockmode, 'unknown_mode' ) + class ForUpdateTest(_fixtures.FixtureTest): @classmethod def setup_mappers(cls): User, users = cls.classes.User, cls.tables.users mapper(User, users) - def _assert(self, read=False, nowait=False, of=None, + def _assert(self, read=False, nowait=False, of=None, no_key=None, assert_q_of=None, assert_sel_of=None): User = self.classes.User s = Session() - q = s.query(User).with_for_update(read=read, nowait=nowait, of=of) + q = s.query(User).with_for_update(read=read, nowait=nowait, of=of, no_key=no_key) sel = q._compile_context().statement assert q._for_update_arg.read is read @@ -72,9 +81,15 @@ class ForUpdateTest(_fixtures.FixtureTest): assert q._for_update_arg.nowait is nowait assert sel._for_update_arg.nowait is nowait + assert q._for_update_arg.no_key is no_key + assert sel._for_update_arg.no_key is no_key + eq_(q._for_update_arg.of, assert_q_of) eq_(sel._for_update_arg.of, assert_sel_of) + def test_no_key(self): + self._assert(no_key=True) + def test_read(self): self._assert(read=True) @@ -172,6 +187,22 @@ class CompileTest(_fixtures.FixtureTest, AssertsCompiledSQL): dialect="postgresql" ) + def test_postgres_for_no_key_update(self): + User = self.classes.User + sess = Session() + self.assert_compile(sess.query(User.id).with_for_update(no_key=True), + "SELECT users.id AS users_id FROM users FOR NO KEY UPDATE", + dialect="postgresql" + ) + + def test_postgres_for_no_key_nowait_update(self): + User = self.classes.User + sess = Session() + self.assert_compile(sess.query(User.id).with_for_update(no_key=True, nowait=True), + "SELECT users.id AS users_id FROM users FOR NO KEY UPDATE NOWAIT", + dialect="postgresql" + ) + def test_postgres_update_of_list(self): User = self.classes.User sess = Session() |