1 files changed, 80 insertions, 0 deletions
diff --git a/test/sql/test_selectable.py b/test/sql/test_selectable.py
index df174fb25..8c7bf43b0 100644
--- a/test/sql/test_selectable.py
+++ b/test/sql/test_selectable.py
@@ -10,6 +10,7 @@ from sqlalchemy.sql import util as sql_util, visitors, expression
 from sqlalchemy import exc
 from sqlalchemy.sql import table, column, null
 from sqlalchemy import util
+from sqlalchemy.schema import Column, Table, MetaData
 
 metadata = MetaData()
 table1 = Table('table1', metadata,
@@ -513,6 +514,18 @@ class SelectableTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiled
             "SELECT c FROM (SELECT (SELECT (SELECT table1.col1 AS a FROM table1) AS b) AS c)"
         )
 
+    def test_self_referential_select_raises(self):
+        t = table('t', column('x'))
+
+        s = select([t])
+
+        s.append_whereclause(s.c.x > 5)
+        assert_raises_message(
+            exc.InvalidRequestError,
+            r"select\(\) construct refers to itself as a FROM",
+            s.compile
+        )
+
     def test_unusual_column_elements_text(self):
         """test that .c excludes text()."""
 
@@ -1460,6 +1473,12 @@ class AnnotationsTest(fixtures.TestBase):
         c1.name = 'somename'
         eq_(c1_a.name, 'somename')
 
+    def test_late_table_add(self):
+        c1 = Column("foo", Integer)
+        c1_a = c1._annotate({"foo": "bar"})
+        t = Table('t', MetaData(), c1)
+        is_(c1_a.table, t)
+
     def test_custom_constructions(self):
         from sqlalchemy.schema import Column
         class MyColumn(Column):
@@ -1884,3 +1903,64 @@ 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_legacy_setter(self):
+        t = table('t', column('c'))
+        s = select([t])
+        s.for_update = 'nowait'
+        eq_(s._for_update_arg.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")