diff options
Diffstat (limited to 'test/dialect/test_postgresql.py')
| -rw-r--r-- | test/dialect/test_postgresql.py | 420 |
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', |