- expand the check to determine if a selectable column is embedded

in the corresponding selectable to take into account clones
of the target column.  fixes [ticket:2419]
- have _make_proxy() copy out the _is_clone_of attribute on the
new column so that even more corresponding_column() checks
work as expected for cloned elements.
- add a new test fixture so that mapped tests can be specified
using declarative.

1 files changed, 80 insertions, 2 deletions

diff --git a/test/sql/test_generative.py b/test/sql/test_generative.py
index f9333dbf5..98e783ede 100644
--- a/test/sql/test_generative.py
+++ b/test/sql/test_generative.py
@@ -5,7 +5,7 @@ from test.lib import *
 from sqlalchemy.sql.visitors import *
 from sqlalchemy import util, exc
 from sqlalchemy.sql import util as sql_util
-from test.lib.testing import eq_, assert_raises
+from test.lib.testing import eq_, ne_, assert_raises
 
 class TraversalTest(fixtures.TestBase, AssertsExecutionResults):
     """test ClauseVisitor's traversal, particularly its 
@@ -173,7 +173,7 @@ class ClauseTest(fixtures.TestBase, AssertsCompiledSQL):
 
     @classmethod
     def setup_class(cls):
-        global t1, t2
+        global t1, t2, t3
         t1 = table("table1",
             column("col1"),
             column("col2"),
@@ -184,6 +184,10 @@ class ClauseTest(fixtures.TestBase, AssertsCompiledSQL):
             column("col2"),
             column("col3"),
             )
+        t3 = Table('table3', MetaData(), 
+            Column('col1', Integer),
+            Column('col2', Integer)
+        )
 
     def test_binary(self):
         clause = t1.c.col2 == t2.c.col2
@@ -242,6 +246,80 @@ class ClauseTest(fixtures.TestBase, AssertsCompiledSQL):
             str(f)
         )
 
+    def test_aliased_cloned_column_adapt_inner(self):
+        clause = select([t1.c.col1, func.foo(t1.c.col2).label('foo')])
+
+        aliased1 = select([clause.c.col1, clause.c.foo])
+        aliased2 = clause
+        aliased2.c.col1, aliased2.c.foo
+        aliased3 = cloned_traverse(aliased2, {}, {})
+
+        # fixed by [ticket:2419].   the inside columns
+        # on aliased3 have _is_clone_of pointers to those of
+        # aliased2.  corresponding_column checks these 
+        # now.
+        adapter = sql_util.ColumnAdapter(aliased1)
+        f1 = select([
+            adapter.columns[c]
+            for c in aliased2._raw_columns
+        ])
+        f2 = select([
+            adapter.columns[c]
+            for c in aliased3._raw_columns
+        ])
+        eq_(
+            str(f1), str(f2)
+        )
+
+    def test_aliased_cloned_column_adapt_exported(self):
+        clause = select([t1.c.col1, func.foo(t1.c.col2).label('foo')])
+
+        aliased1 = select([clause.c.col1, clause.c.foo])
+        aliased2 = clause
+        aliased2.c.col1, aliased2.c.foo
+        aliased3 = cloned_traverse(aliased2, {}, {})
+
+        # also fixed by [ticket:2419].  When we look at the
+        # *outside* columns of aliased3, they previously did not 
+        # have an _is_clone_of pointer.   But we now modified _make_proxy
+        # to assign this.
+        adapter = sql_util.ColumnAdapter(aliased1)
+        f1 = select([
+            adapter.columns[c]
+            for c in aliased2.c
+        ])
+        f2 = select([
+            adapter.columns[c]
+            for c in aliased3.c
+        ])
+        eq_(
+            str(f1), str(f2)
+        )
+
+    def test_aliased_cloned_schema_column_adapt_exported(self):
+        clause = select([t3.c.col1, func.foo(t3.c.col2).label('foo')])
+
+        aliased1 = select([clause.c.col1, clause.c.foo])
+        aliased2 = clause
+        aliased2.c.col1, aliased2.c.foo
+        aliased3 = cloned_traverse(aliased2, {}, {})
+
+        # also fixed by [ticket:2419].  When we look at the
+        # *outside* columns of aliased3, they previously did not 
+        # have an _is_clone_of pointer.   But we now modified _make_proxy
+        # to assign this.
+        adapter = sql_util.ColumnAdapter(aliased1)
+        f1 = select([
+            adapter.columns[c]
+            for c in aliased2.c
+        ])
+        f2 = select([
+            adapter.columns[c]
+            for c in aliased3.c
+        ])
+        eq_(
+            str(f1), str(f2)
+        )
 
     def test_text(self):
         clause = text(