diff options
author | Mike Bayer <mike_mp@zzzcomputing.com> | 2013-11-28 22:25:09 -0500 |
---|---|---|
committer | Mike Bayer <mike_mp@zzzcomputing.com> | 2013-11-28 22:25:09 -0500 |
commit | 4aaf3753d75c68050c136e734c29aae5ff9504b4 (patch) | |
tree | 21cf4c877fca83271447a7e1637a7876aa42f423 /test/sql/test_selectable.py | |
parent | bb60a8ad946dd331f546f06a156b7ebb87d1709d (diff) | |
download | sqlalchemy-4aaf3753d75c68050c136e734c29aae5ff9504b4.tar.gz |
- fix up rendering of "of"
- move out tests, dialect specific out of compiler, compiler tests use new API,
legacy API tests in test_selecatble
- add support for adaptation of ForUpdateArg, alias support in compilers
Diffstat (limited to 'test/sql/test_selectable.py')
-rw-r--r-- | test/sql/test_selectable.py | 54 |
1 files changed, 54 insertions, 0 deletions
diff --git a/test/sql/test_selectable.py b/test/sql/test_selectable.py index 0fc7a0ed0..66cdd87c2 100644 --- a/test/sql/test_selectable.py +++ b/test/sql/test_selectable.py @@ -1903,3 +1903,57 @@ class WithLabelsTest(fixtures.TestBase): ['t1_x', 't2_x'] ) self._assert_result_keys(sel, ['t1_a', 't2_b']) + +class ForUpdateTest(fixtures.TestBase, AssertsCompiledSQL): + __dialect__ = "default" + + def _assert_legacy(self, leg, read=False, nowait=False): + t = table('t', column('c')) + s1 = select([t], for_update=leg) + if leg is False: + assert s1._for_update_arg is None + assert s1.for_update is None + else: + eq_( + s1._for_update_arg.read, read + ) + eq_( + s1._for_update_arg.nowait, nowait + ) + eq_(s1.for_update, leg) + + def test_false_legacy(self): + self._assert_legacy(False) + + def test_plain_true_legacy(self): + self._assert_legacy(True) + + def test_read_legacy(self): + self._assert_legacy("read", read=True) + + def test_nowait_legacy(self): + self._assert_legacy("nowait", nowait=True) + + def test_read_nowait_legacy(self): + self._assert_legacy("read_nowait", read=True, nowait=True) + + def test_basic_clone(self): + t = table('t', column('c')) + s = select([t]).with_for_update(read=True, of=t.c.c) + s2 = visitors.ReplacingCloningVisitor().traverse(s) + assert s2._for_update_arg is not s._for_update_arg + eq_(s2._for_update_arg.read, True) + eq_(s2._for_update_arg.of, [t.c.c]) + self.assert_compile(s2, + "SELECT t.c FROM t FOR SHARE OF t", + dialect="postgresql") + + def test_adapt(self): + t = table('t', column('c')) + s = select([t]).with_for_update(read=True, of=t.c.c) + a = t.alias() + s2 = sql_util.ClauseAdapter(a).traverse(s) + eq_(s2._for_update_arg.of, [a.c.c]) + self.assert_compile(s2, + "SELECT t_1.c FROM t AS t_1 FOR SHARE OF t_1", + dialect="postgresql") |