From 0326f3cf014ffb4928b4c6051d2fca13cb6945d7 Mon Sep 17 00:00:00 2001 From: Mike Bayer Date: Sun, 2 Feb 2014 16:33:54 -0500 Subject: - 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] --- test/dialect/postgresql/test_reflection.py | 263 +++++++++++++++++++---------- 1 file changed, 175 insertions(+), 88 deletions(-) (limited to 'test/dialect/postgresql/test_reflection.py') 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): -- cgit v1.2.1