- Added :paramref:`.MetaData.reflect.**dialect_kwargs`

to support dialect-level reflection options for all :class:`.Table`
objects reflected.

- Added a new dialect-level argument ``postgresql_ignore_search_path``;
this argument is accepted by both the :class:`.Table` constructor
as well as by the :meth:`.MetaData.reflect` method.  When in use
against Postgresql, a foreign-key referenced table which specifies
a remote schema name will retain that schema name even if the name
is present in the ``search_path``; the default behavior since 0.7.3
has been that schemas present in ``search_path`` would not be copied
to reflected :class:`.ForeignKey` objects.  The documentation has been
updated to describe in detail the behavior of the ``pg_get_constraintdef()``
function and how the ``postgresql_ignore_search_path`` feature essentially
determines if we will honor the schema qualification reported by
this function or not. [ticket:2922]

1 files changed, 175 insertions, 88 deletions

diff --git a/test/dialect/postgresql/test_reflection.py b/test/dialect/postgresql/test_reflection.py
index 58f34d5d0..705a64c8e 100644
--- a/test/dialect/postgresql/test_reflection.py
+++ b/test/dialect/postgresql/test_reflection.py
@@ -203,24 +203,19 @@ class ReflectionTest(fixtures.TestBase):
         eq_([c.name for c in t2.primary_key], ['t_id'])
 
     @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"""
+    def test_cross_schema_reflection_one(self):
 
         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,
+        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('remote_user_id', Integer, ForeignKey(users.c.user_id)),
             Column('email_address', String(20)),
-            schema='test_schema',
-            )
+            schema='test_schema')
         meta1.create_all()
         meta2 = MetaData(testing.db)
         addresses = Table('email_addresses', meta2, autoload=True,
@@ -232,13 +227,14 @@ class ReflectionTest(fixtures.TestBase):
                      == addresses.c.remote_user_id).compare(j.onclause))
 
     @testing.provide_metadata
-    def test_schema_reflection_2(self):
+    def test_cross_schema_reflection_two(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')
+        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)
@@ -249,113 +245,204 @@ class ReflectionTest(fixtures.TestBase):
                         subject.join(referer).onclause))
 
     @testing.provide_metadata
-    def test_schema_reflection_3(self):
+    def test_cross_schema_reflection_three(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')),
+        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)
+        referer = Table('referer', meta2, autoload=True,
+                        schema='test_schema')
         self.assert_((subject.c.id
                      == referer.c.ref).compare(
                         subject.join(referer).onclause))
 
     @testing.provide_metadata
-    def test_uppercase_lowercase_table(self):
-        metadata = self.metadata
-
-        a_table = Table('a', metadata, Column('x', Integer))
-        A_table = Table('A', metadata, Column('x', Integer))
-
-        a_table.create()
-        assert testing.db.has_table("a")
-        assert not testing.db.has_table("A")
-        A_table.create(checkfirst=True)
-        assert testing.db.has_table("A")
-
-    def test_uppercase_lowercase_sequence(self):
+    def test_cross_schema_reflection_four(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()
 
-        a_seq = Sequence('a')
-        A_seq = Sequence('A')
+        conn = testing.db.connect()
+        conn.detach()
+        conn.execute("SET search_path TO test_schema, test_schema_2")
+        meta2 = MetaData(bind=conn)
+        subject = Table('subject', meta2, autoload=True,
+                        schema='test_schema_2',
+                        postgresql_ignore_search_path=True)
+        referer = Table('referer', meta2, autoload=True,
+                        schema='test_schema',
+                        postgresql_ignore_search_path=True)
+        self.assert_((subject.c.id
+                     == referer.c.ref).compare(
+                        subject.join(referer).onclause))
+        conn.close()
 
-        a_seq.create(testing.db)
-        assert testing.db.dialect.has_sequence(testing.db, "a")
-        assert not testing.db.dialect.has_sequence(testing.db, "A")
-        A_seq.create(testing.db, checkfirst=True)
-        assert testing.db.dialect.has_sequence(testing.db, "A")
+    @testing.provide_metadata
+    def test_cross_schema_reflection_five(self):
+        meta1 = self.metadata
 
-        a_seq.drop(testing.db)
-        A_seq.drop(testing.db)
+        # we assume 'public'
+        default_schema = testing.db.dialect.default_schema_name
+        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()
 
-    def test_schema_reflection_multi_search_path(self):
-        """test the 'set the same schema' rule when
-        multiple schemas/search paths are in effect."""
+        meta2 = MetaData(testing.db)
+        subject = Table('subject', meta2, autoload=True,
+                        schema=default_schema,
+                        postgresql_ignore_search_path=True
+                        )
+        referer = Table('referer', meta2, autoload=True,
+                        schema=default_schema,
+                        postgresql_ignore_search_path=True
+                        )
+        assert subject.schema == default_schema
+        self.assert_((subject.c.id
+                     == referer.c.ref).compare(
+                        subject.join(referer).onclause))
 
-        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"
+    @testing.provide_metadata
+    def test_cross_schema_reflection_six(self):
+        # test that the search path *is* taken into account
+        # by default
+        meta1 = self.metadata
 
-            conn.execute("""
-            CREATE TABLE test_schema.some_table (
-                id SERIAL not null primary key
+        Table('some_table', meta1,
+                Column('id', Integer, primary_key=True),
+                schema='test_schema'
             )
-            """)
+        Table('some_other_table', meta1,
+                Column('id', Integer, primary_key=True),
+                Column('sid', Integer, ForeignKey('test_schema.some_table.id')),
+                schema='test_schema_2'
+                )
+        meta1.create_all()
+        with testing.db.connect() as conn:
+            conn.detach()
 
-            conn.execute("""
-            CREATE TABLE test_schema_2.some_other_table (
-                id SERIAL not null primary key,
-                sid INTEGER REFERENCES test_schema.some_table(id)
-            )
-            """)
+            conn.execute("set search_path to test_schema_2, test_schema, public")
 
-            m1 = MetaData()
+            m1 = MetaData(conn)
 
-            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)
+                                autoload=True)
+            t2_schema = Table('some_other_table',
+                                m1,
+                                schema="test_schema_2",
+                                autoload=True)
 
             t2_no_schema = Table('some_other_table',
                                 m1,
-                                autoload=True,
-                                autoload_with=conn)
+                                autoload=True)
 
             t1_no_schema = Table('some_table',
                                 m1,
+                                autoload=True)
+
+            m2 = MetaData(conn)
+            t1_schema_isp = Table('some_table',
+                                m2,
+                                schema="test_schema",
                                 autoload=True,
-                                autoload_with=conn)
+                                postgresql_ignore_search_path=True)
+            t2_schema_isp = Table('some_other_table',
+                                m2,
+                                schema="test_schema_2",
+                                autoload=True,
+                                postgresql_ignore_search_path=True)
+
 
-            # 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 ?
+            # t2_schema refers to t1_schema, but since "test_schema"
+            # is in the search path, we instead link to t2_no_schema
             assert t2_schema.c.sid.references(
                                 t1_no_schema.c.id)
 
+            # the two no_schema tables refer to each other also.
             assert t2_no_schema.c.sid.references(
                                 t1_no_schema.c.id)
 
-        finally:
-            trans.rollback()
-            conn.close()
-            db.dispose()
+            # but if we're ignoring search path, then we maintain
+            # those explicit schemas vs. what the "default" schema is
+            assert t2_schema_isp.c.sid.references(t1_schema_isp.c.id)
+
+    @testing.provide_metadata
+    def test_cross_schema_reflection_seven(self):
+        # test that the search path *is* taken into account
+        # by default
+        meta1 = self.metadata
+
+        Table('some_table', meta1,
+                Column('id', Integer, primary_key=True),
+                schema='test_schema'
+            )
+        Table('some_other_table', meta1,
+                Column('id', Integer, primary_key=True),
+                Column('sid', Integer, ForeignKey('test_schema.some_table.id')),
+                schema='test_schema_2'
+                )
+        meta1.create_all()
+        with testing.db.connect() as conn:
+            conn.detach()
+
+            conn.execute("set search_path to test_schema_2, test_schema, public")
+            meta2 = MetaData(conn)
+            meta2.reflect(schema="test_schema_2")
+
+            eq_(set(meta2.tables), set(['test_schema_2.some_other_table', 'some_table']))
+
+            meta3 = MetaData(conn)
+            meta3.reflect(schema="test_schema_2", postgresql_ignore_search_path=True)
+
+            eq_(set(meta3.tables),
+                    set(['test_schema_2.some_other_table', 'test_schema.some_table']))
+
+
+    @testing.provide_metadata
+    def test_uppercase_lowercase_table(self):
+        metadata = self.metadata
+
+        a_table = Table('a', metadata, Column('x', Integer))
+        A_table = Table('A', metadata, Column('x', Integer))
+
+        a_table.create()
+        assert testing.db.has_table("a")
+        assert not testing.db.has_table("A")
+        A_table.create(checkfirst=True)
+        assert testing.db.has_table("A")
+
+    def test_uppercase_lowercase_sequence(self):
+
+        a_seq = Sequence('a')
+        A_seq = Sequence('A')
+
+        a_seq.create(testing.db)
+        assert testing.db.dialect.has_sequence(testing.db, "a")
+        assert not testing.db.dialect.has_sequence(testing.db, "A")
+        A_seq.create(testing.db, checkfirst=True)
+        assert testing.db.dialect.has_sequence(testing.db, "A")
+
+        a_seq.drop(testing.db)
+        A_seq.drop(testing.db)
+
 
     @testing.provide_metadata
     def test_index_reflection(self):