- Narrowed the assumption made when reflecting

a foreign-key referenced table with schema in
the current search path; an explicit schema will
be applied to the referenced table only if
it actually matches that of the referencing table,
which also has an explicit schema.   Previously
it was assumed that "current" schema was synonymous
with the full search_path.  [ticket:2249]

1 files changed, 236 insertions, 184 deletions

diff --git a/test/dialect/test_postgresql.py b/test/dialect/test_postgresql.py
index 97221449e..9171ff819 100644
--- a/test/dialect/test_postgresql.py
+++ b/test/dialect/test_postgresql.py
@@ -1255,6 +1255,242 @@ class DistinctOnTest(fixtures.TestBase, AssertsCompiledSQL):
             "t.b AS b FROM t) AS sq WHERE t.id = sq.id"
             )
 
+class ReflectionTest(fixtures.TestBase):
+    @testing.provide_metadata
+    def test_pg_weirdchar_reflection(self):
+        meta1 = self.metadata
+        subject = Table('subject', meta1, Column('id$', Integer,
+                        primary_key=True))
+        referer = Table('referer', meta1, Column('id', Integer,
+                        primary_key=True), Column('ref', Integer,
+                        ForeignKey('subject.id$')))
+        meta1.create_all()
+        meta2 = MetaData(testing.db)
+        subject = Table('subject', meta2, autoload=True)
+        referer = Table('referer', meta2, autoload=True)
+        self.assert_((subject.c['id$']
+                     == referer.c.ref).compare(
+                        subject.join(referer).onclause))
+
+    @testing.provide_metadata
+    def test_renamed_sequence_reflection(self):
+        metadata = self.metadata
+        t = Table('t', metadata, Column('id', Integer, primary_key=True))
+        metadata.create_all()
+        m2 = MetaData(testing.db)
+        t2 = Table('t', m2, autoload=True, implicit_returning=False)
+        eq_(t2.c.id.server_default.arg.text,
+            "nextval('t_id_seq'::regclass)")
+        r = t2.insert().execute()
+        eq_(r.inserted_primary_key, [1])
+        testing.db.connect().execution_options(autocommit=True).\
+                execute('alter table t_id_seq rename to foobar_id_seq'
+                )
+        m3 = MetaData(testing.db)
+        t3 = Table('t', m3, autoload=True, implicit_returning=False)
+        eq_(t3.c.id.server_default.arg.text,
+            "nextval('foobar_id_seq'::regclass)")
+        r = t3.insert().execute()
+        eq_(r.inserted_primary_key, [2])
+
+    @testing.provide_metadata
+    def test_schema_reflection(self):
+        """note: this test requires that the 'test_schema' schema be
+        separate and accessible by the test user"""
+
+        meta1 = self.metadata
+
+        users = Table('users', meta1, Column('user_id', Integer,
+                      primary_key=True), Column('user_name',
+                      String(30), nullable=False), schema='test_schema')
+        addresses = Table(
+            'email_addresses',
+            meta1,
+            Column('address_id', Integer, primary_key=True),
+            Column('remote_user_id', Integer,
+                   ForeignKey(users.c.user_id)),
+            Column('email_address', String(20)),
+            schema='test_schema',
+            )
+        meta1.create_all()
+        meta2 = MetaData(testing.db)
+        addresses = Table('email_addresses', meta2, autoload=True,
+                          schema='test_schema')
+        users = Table('users', meta2, mustexist=True,
+                      schema='test_schema')
+        j = join(users, addresses)
+        self.assert_((users.c.user_id
+                     == addresses.c.remote_user_id).compare(j.onclause))
+
+    @testing.provide_metadata
+    def test_schema_reflection_2(self):
+        meta1 = self.metadata
+        subject = Table('subject', meta1, Column('id', Integer,
+                        primary_key=True))
+        referer = Table('referer', meta1, Column('id', Integer,
+                        primary_key=True), Column('ref', Integer,
+                        ForeignKey('subject.id')), schema='test_schema')
+        meta1.create_all()
+        meta2 = MetaData(testing.db)
+        subject = Table('subject', meta2, autoload=True)
+        referer = Table('referer', meta2, schema='test_schema',
+                        autoload=True)
+        self.assert_((subject.c.id
+                     == referer.c.ref).compare(
+                        subject.join(referer).onclause))
+
+    @testing.provide_metadata
+    def test_schema_reflection_3(self):
+        meta1 = self.metadata
+        subject = Table('subject', meta1, Column('id', Integer,
+                        primary_key=True), schema='test_schema_2')
+        referer = Table('referer', meta1, Column('id', Integer,
+                        primary_key=True), Column('ref', Integer,
+                        ForeignKey('test_schema_2.subject.id')),
+                        schema='test_schema')
+        meta1.create_all()
+        meta2 = MetaData(testing.db)
+        subject = Table('subject', meta2, autoload=True,
+                        schema='test_schema_2')
+        referer = Table('referer', meta2, schema='test_schema',
+                        autoload=True)
+        self.assert_((subject.c.id
+                     == referer.c.ref).compare(
+                        subject.join(referer).onclause))
+
+    def test_schema_reflection_multi_search_path(self):
+        """test the 'set the same schema' rule when
+        multiple schemas/search paths are in effect."""
+
+        db = engines.testing_engine()
+        conn = db.connect()
+        trans = conn.begin()
+        try:
+            conn.execute("set search_path to test_schema_2, "
+                                "test_schema, public")
+            conn.dialect.default_schema_name = "test_schema_2"
+
+            conn.execute("""
+            CREATE TABLE test_schema.some_table (
+                id SERIAL not null primary key
+            )
+            """)
+
+            conn.execute("""
+            CREATE TABLE test_schema_2.some_other_table (
+                id SERIAL not null primary key,
+                sid INTEGER REFERENCES test_schema.some_table(id)
+            )
+            """)
+
+            m1 = MetaData()
+
+            t2_schema = Table('some_other_table', 
+                                m1, 
+                                schema="test_schema_2", 
+                                autoload=True, 
+                                autoload_with=conn)
+            t1_schema = Table('some_table', 
+                                m1, 
+                                schema="test_schema", 
+                                autoload=True,
+                                autoload_with=conn)
+
+            t2_no_schema = Table('some_other_table', 
+                                m1, 
+                                autoload=True, 
+                                autoload_with=conn)
+
+            t1_no_schema = Table('some_table', 
+                                m1, 
+                                autoload=True, 
+                                autoload_with=conn)
+
+            # OK, this because, "test_schema" is 
+            # in the search path, and might as well be
+            # the default too.  why would we assign
+            # a "schema" to the Table ?
+            assert t2_schema.c.sid.references(
+                                t1_no_schema.c.id)
+
+            assert t2_no_schema.c.sid.references(
+                                t1_no_schema.c.id)
+
+        finally:
+            trans.rollback()
+            conn.close()
+            db.dispose()
+
+    @testing.provide_metadata
+    def test_index_reflection(self):
+        """ Reflecting partial & expression-based indexes should warn
+        """
+
+        metadata = self.metadata
+
+        t1 = Table('party', metadata, Column('id', String(10),
+                   nullable=False), Column('name', String(20),
+                   index=True), Column('aname', String(20)))
+        metadata.create_all()
+        testing.db.execute("""
+          create index idx1 on party ((id || name))
+        """)
+        testing.db.execute("""
+          create unique index idx2 on party (id) where name = 'test'
+        """)
+        testing.db.execute("""
+            create index idx3 on party using btree
+                (lower(name::text), lower(aname::text))
+        """)
+
+        def go():
+            m2 = MetaData(testing.db)
+            t2 = Table('party', m2, autoload=True)
+            assert len(t2.indexes) == 2
+
+            # Make sure indexes are in the order we expect them in
+
+            tmp = [(idx.name, idx) for idx in t2.indexes]
+            tmp.sort()
+            r1, r2 = [idx[1] for idx in tmp]
+            assert r1.name == 'idx2'
+            assert r1.unique == True
+            assert r2.unique == False
+            assert [t2.c.id] == r1.columns
+            assert [t2.c.name] == r2.columns
+
+        testing.assert_warnings(go,
+            [
+                'Skipped unsupported reflection of '
+                'expression-based index idx1',
+                'Predicate of partial index idx2 ignored during '
+                'reflection',
+                'Skipped unsupported reflection of '
+                'expression-based index idx3'
+            ])
+
+    @testing.provide_metadata
+    def test_index_reflection_modified(self):
+        """reflect indexes when a column name has changed - PG 9 
+        does not update the name of the column in the index def.
+        [ticket:2141]
+
+        """
+
+        metadata = self.metadata
+
+        t1 = Table('t', metadata,
+            Column('id', Integer, primary_key=True),
+            Column('x', Integer)
+        )
+        metadata.create_all()
+        conn = testing.db.connect().execution_options(autocommit=True)
+        conn.execute("CREATE INDEX idx1 ON t (x)")
+        conn.execute("ALTER TABLE t RENAME COLUMN x to y")
+
+        ind = testing.db.dialect.get_indexes(conn, "t", None)
+        eq_(ind, [{'unique': False, 'column_names': [u'y'], 'name': u'idx1'}])
+        conn.close()
 
 class MiscTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL):
 
@@ -1325,24 +1561,6 @@ class MiscTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL):
         assert 'will create implicit sequence' in msgs
         assert 'will create implicit index' in msgs
 
-    def test_pg_weirdchar_reflection(self):
-        meta1 = MetaData(testing.db)
-        subject = Table('subject', meta1, Column('id$', Integer,
-                        primary_key=True))
-        referer = Table('referer', meta1, Column('id', Integer,
-                        primary_key=True), Column('ref', Integer,
-                        ForeignKey('subject.id$')))
-        meta1.create_all()
-        try:
-            meta2 = MetaData(testing.db)
-            subject = Table('subject', meta2, autoload=True)
-            referer = Table('referer', meta2, autoload=True)
-            print str(subject.join(referer).onclause)
-            self.assert_((subject.c['id$']
-                         == referer.c.ref).compare(
-                            subject.join(referer).onclause))
-        finally:
-            meta1.drop_all()
 
     @testing.fails_on('+zxjdbc',
                       "Can't infer the SQL type to use for an instance "
@@ -1369,102 +1587,6 @@ class MiscTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL):
         finally:
             t.drop(checkfirst=True)
 
-    @testing.provide_metadata
-    def test_renamed_sequence_reflection(self):
-        metadata = self.metadata
-        t = Table('t', metadata, Column('id', Integer, primary_key=True))
-        metadata.create_all()
-        m2 = MetaData(testing.db)
-        t2 = Table('t', m2, autoload=True, implicit_returning=False)
-        eq_(t2.c.id.server_default.arg.text,
-            "nextval('t_id_seq'::regclass)")
-        r = t2.insert().execute()
-        eq_(r.inserted_primary_key, [1])
-        testing.db.connect().execution_options(autocommit=True).\
-                execute('alter table t_id_seq rename to foobar_id_seq'
-                )
-        m3 = MetaData(testing.db)
-        t3 = Table('t', m3, autoload=True, implicit_returning=False)
-        eq_(t3.c.id.server_default.arg.text,
-            "nextval('foobar_id_seq'::regclass)")
-        r = t3.insert().execute()
-        eq_(r.inserted_primary_key, [2])
-
-    def test_schema_reflection(self):
-        """note: this test requires that the 'test_schema' schema be
-        separate and accessible by the test user"""
-
-        meta1 = MetaData(testing.db)
-        users = Table('users', meta1, Column('user_id', Integer,
-                      primary_key=True), Column('user_name',
-                      String(30), nullable=False), schema='test_schema')
-        addresses = Table(
-            'email_addresses',
-            meta1,
-            Column('address_id', Integer, primary_key=True),
-            Column('remote_user_id', Integer,
-                   ForeignKey(users.c.user_id)),
-            Column('email_address', String(20)),
-            schema='test_schema',
-            )
-        meta1.create_all()
-        try:
-            meta2 = MetaData(testing.db)
-            addresses = Table('email_addresses', meta2, autoload=True,
-                              schema='test_schema')
-            users = Table('users', meta2, mustexist=True,
-                          schema='test_schema')
-            print users
-            print addresses
-            j = join(users, addresses)
-            print str(j.onclause)
-            self.assert_((users.c.user_id
-                         == addresses.c.remote_user_id).compare(j.onclause))
-        finally:
-            meta1.drop_all()
-
-    def test_schema_reflection_2(self):
-        meta1 = MetaData(testing.db)
-        subject = Table('subject', meta1, Column('id', Integer,
-                        primary_key=True))
-        referer = Table('referer', meta1, Column('id', Integer,
-                        primary_key=True), Column('ref', Integer,
-                        ForeignKey('subject.id')), schema='test_schema')
-        meta1.create_all()
-        try:
-            meta2 = MetaData(testing.db)
-            subject = Table('subject', meta2, autoload=True)
-            referer = Table('referer', meta2, schema='test_schema',
-                            autoload=True)
-            print str(subject.join(referer).onclause)
-            self.assert_((subject.c.id
-                         == referer.c.ref).compare(
-                            subject.join(referer).onclause))
-        finally:
-            meta1.drop_all()
-
-    def test_schema_reflection_3(self):
-        meta1 = MetaData(testing.db)
-        subject = Table('subject', meta1, Column('id', Integer,
-                        primary_key=True), schema='test_schema_2')
-        referer = Table('referer', meta1, Column('id', Integer,
-                        primary_key=True), Column('ref', Integer,
-                        ForeignKey('test_schema_2.subject.id')),
-                        schema='test_schema')
-        meta1.create_all()
-        try:
-            meta2 = MetaData(testing.db)
-            subject = Table('subject', meta2, autoload=True,
-                            schema='test_schema_2')
-            referer = Table('referer', meta2, schema='test_schema',
-                            autoload=True)
-            print str(subject.join(referer).onclause)
-            self.assert_((subject.c.id
-                         == referer.c.ref).compare(
-                            subject.join(referer).onclause))
-        finally:
-            meta1.drop_all()
-
     def test_schema_roundtrips(self):
         meta = MetaData(testing.db)
         users = Table('users', meta, Column('id', Integer,
@@ -1515,76 +1637,6 @@ class MiscTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL):
         finally:
             testing.db.execute('drop table speedy_users')
 
-    @testing.provide_metadata
-    def test_index_reflection(self):
-        """ Reflecting partial & expression-based indexes should warn
-        """
-
-        metadata = self.metadata
-
-        t1 = Table('party', metadata, Column('id', String(10),
-                   nullable=False), Column('name', String(20),
-                   index=True), Column('aname', String(20)))
-        metadata.create_all()
-        testing.db.execute("""
-          create index idx1 on party ((id || name))
-        """)
-        testing.db.execute("""
-          create unique index idx2 on party (id) where name = 'test'
-        """)
-        testing.db.execute("""
-            create index idx3 on party using btree
-                (lower(name::text), lower(aname::text))
-        """)
-
-        def go():
-            m2 = MetaData(testing.db)
-            t2 = Table('party', m2, autoload=True)
-            assert len(t2.indexes) == 2
-
-            # Make sure indexes are in the order we expect them in
-
-            tmp = [(idx.name, idx) for idx in t2.indexes]
-            tmp.sort()
-            r1, r2 = [idx[1] for idx in tmp]
-            assert r1.name == 'idx2'
-            assert r1.unique == True
-            assert r2.unique == False
-            assert [t2.c.id] == r1.columns
-            assert [t2.c.name] == r2.columns
-
-        testing.assert_warnings(go,
-            [
-                'Skipped unsupported reflection of '
-                'expression-based index idx1',
-                'Predicate of partial index idx2 ignored during '
-                'reflection',
-                'Skipped unsupported reflection of '
-                'expression-based index idx3'
-            ])
-
-    @testing.provide_metadata
-    def test_index_reflection_modified(self):
-        """reflect indexes when a column name has changed - PG 9 
-        does not update the name of the column in the index def.
-        [ticket:2141]
-
-        """
-
-        metadata = self.metadata
-
-        t1 = Table('t', metadata,
-            Column('id', Integer, primary_key=True),
-            Column('x', Integer)
-        )
-        metadata.create_all()
-        conn = testing.db.connect().execution_options(autocommit=True)
-        conn.execute("CREATE INDEX idx1 ON t (x)")
-        conn.execute("ALTER TABLE t RENAME COLUMN x to y")
-
-        ind = testing.db.dialect.get_indexes(conn, "t", None)
-        eq_(ind, [{'unique': False, 'column_names': [u'y'], 'name': u'idx1'}])
-        conn.close()
 
     @testing.fails_on('+zxjdbc', 'psycopg2/pg8000 specific assertion')
     @testing.fails_on('pypostgresql',