diff options
Diffstat (limited to 'test/engine/test_reflection.py')
| -rw-r--r-- | test/engine/test_reflection.py | 88 |
1 files changed, 58 insertions, 30 deletions
diff --git a/test/engine/test_reflection.py b/test/engine/test_reflection.py index 087610333..83650609d 100644 --- a/test/engine/test_reflection.py +++ b/test/engine/test_reflection.py @@ -14,6 +14,7 @@ from sqlalchemy.util import ue metadata, users = None, None + class ReflectionTest(fixtures.TestBase, ComparesTables): __backend__ = True @@ -253,7 +254,7 @@ class ReflectionTest(fixtures.TestBase, ComparesTables): """ Table('a', self.metadata, Column('id', Integer, primary_key=True)) Table('b', self.metadata, Column('id', Integer, primary_key=True), - Column('a_id', Integer, sa.ForeignKey('a.id'))) + Column('a_id', Integer, sa.ForeignKey('a.id'))) self.metadata.create_all() m2 = MetaData() @@ -275,7 +276,7 @@ class ReflectionTest(fixtures.TestBase, ComparesTables): """ Table('a', self.metadata, Column('id', Integer, primary_key=True)) Table('b', self.metadata, Column('id', Integer, primary_key=True), - Column('a_id', Integer, sa.ForeignKey('a.id'))) + Column('a_id', Integer, sa.ForeignKey('a.id'))) self.metadata.create_all() m2 = MetaData() @@ -404,7 +405,6 @@ class ReflectionTest(fixtures.TestBase, ComparesTables): eq_(list(table.primary_key), [table.c.col1]) eq_(table.c.col1.primary_key, True) - @testing.provide_metadata def test_override_pkfk(self): """test that you can override columns which contain foreign keys @@ -419,7 +419,6 @@ class ReflectionTest(fixtures.TestBase, ComparesTables): Column('id', sa.Integer, primary_key=True), Column('street', sa.String(30))) - meta.create_all() meta2 = MetaData(testing.db) a2 = Table('addresses', meta2, @@ -541,8 +540,6 @@ class ReflectionTest(fixtures.TestBase, ComparesTables): assert f1 in b1.constraints assert len(b1.constraints) == 2 - - @testing.provide_metadata def test_override_keys(self): """test that columns can be overridden with a 'key', @@ -654,12 +651,13 @@ class ReflectionTest(fixtures.TestBase, ComparesTables): backends with {dialect}.get_foreign_keys() support)""" if testing.against('postgresql'): - test_attrs = ('match', 'onupdate', 'ondelete', 'deferrable', 'initially') + test_attrs = ('match', 'onupdate', 'ondelete', + 'deferrable', 'initially') addresses_user_id_fkey = sa.ForeignKey( # Each option is specifically not a Postgres default, or # it won't be returned by PG's inspection 'users.id', - name = 'addresses_user_id_fkey', + name='addresses_user_id_fkey', match='FULL', onupdate='RESTRICT', ondelete='RESTRICT', @@ -672,7 +670,7 @@ class ReflectionTest(fixtures.TestBase, ComparesTables): # elided by MySQL's inspection addresses_user_id_fkey = sa.ForeignKey( 'users.id', - name = 'addresses_user_id_fkey', + name='addresses_user_id_fkey', onupdate='CASCADE', ondelete='CASCADE' ) @@ -726,11 +724,12 @@ class ReflectionTest(fixtures.TestBase, ComparesTables): Column('slot', sa.String(128)), ) - assert_raises_message(sa.exc.InvalidRequestError, - "Foreign key associated with column 'slots.pkg_id' " - "could not find table 'pkgs' with which to generate " - "a foreign key to target column 'pkg_id'", - metadata.create_all) + assert_raises_message( + sa.exc.InvalidRequestError, + "Foreign key associated with column 'slots.pkg_id' " + "could not find table 'pkgs' with which to generate " + "a foreign key to target column 'pkg_id'", + metadata.create_all) def test_composite_pks(self): """test reflection of a composite primary key""" @@ -797,7 +796,6 @@ class ReflectionTest(fixtures.TestBase, ComparesTables): table.c.multi_hoho == table2.c.lala).compare(j.onclause)) - @testing.crashes('oracle', 'FIXME: unknown, confirm not fails_on') @testing.requires.check_constraints @testing.provide_metadata @@ -869,7 +867,6 @@ class ReflectionTest(fixtures.TestBase, ComparesTables): def test_reflect_uses_bind_engine_reflect(self): self._test_reflect_uses_bind(lambda e: MetaData().reflect(e)) - @testing.provide_metadata def test_reflect_all(self): existing = testing.db.table_names() @@ -1053,6 +1050,7 @@ class ReflectionTest(fixtures.TestBase, ComparesTables): finally: _drop_views(metadata.bind) + class CreateDropTest(fixtures.TestBase): __backend__ = True @@ -1101,7 +1099,6 @@ class CreateDropTest(fixtures.TestBase): eq_(ua, ['users', 'email_addresses']) eq_(oi, ['orders', 'items']) - def test_checkfirst(self): try: assert not users.exists(testing.db) @@ -1141,6 +1138,7 @@ class CreateDropTest(fixtures.TestBase): - set(testing.db.table_names())) metadata.drop_all(bind=testing.db) + class SchemaManipulationTest(fixtures.TestBase): __backend__ = True @@ -1159,6 +1157,7 @@ class SchemaManipulationTest(fixtures.TestBase): assert len(addresses.c.user_id.foreign_keys) == 1 assert addresses.constraints == set([addresses.primary_key, fk]) + class UnicodeReflectionTest(fixtures.TestBase): __backend__ = True @@ -1170,16 +1169,40 @@ class UnicodeReflectionTest(fixtures.TestBase): ('plain', 'col_plain', 'ix_plain') ]) no_has_table = [ - ('no_has_table_1', ue('col_Unit\u00e9ble'), ue('ix_Unit\u00e9ble')), - ('no_has_table_2', ue('col_\u6e2c\u8a66'), ue('ix_\u6e2c\u8a66')), + ( + 'no_has_table_1', + ue('col_Unit\u00e9ble'), + ue('ix_Unit\u00e9ble') + ), + ( + 'no_has_table_2', + ue('col_\u6e2c\u8a66'), + ue('ix_\u6e2c\u8a66') + ), ] no_case_sensitivity = [ - (ue('\u6e2c\u8a66'), ue('col_\u6e2c\u8a66'), ue('ix_\u6e2c\u8a66')), - (ue('unit\u00e9ble'), ue('col_unit\u00e9ble'), ue('ix_unit\u00e9ble')), + ( + ue('\u6e2c\u8a66'), + ue('col_\u6e2c\u8a66'), + ue('ix_\u6e2c\u8a66') + ), + ( + ue('unit\u00e9ble'), + ue('col_unit\u00e9ble'), + ue('ix_unit\u00e9ble') + ), ] full = [ - (ue('Unit\u00e9ble'), ue('col_Unit\u00e9ble'), ue('ix_Unit\u00e9ble')), - (ue('\u6e2c\u8a66'), ue('col_\u6e2c\u8a66'), ue('ix_\u6e2c\u8a66')), + ( + ue('Unit\u00e9ble'), + ue('col_Unit\u00e9ble'), + ue('ix_Unit\u00e9ble') + ), + ( + ue('\u6e2c\u8a66'), + ue('col_\u6e2c\u8a66'), + ue('ix_\u6e2c\u8a66') + ), ] # as you can see, our options for this kind of thing @@ -1268,6 +1291,7 @@ class UnicodeReflectionTest(fixtures.TestBase): [(names[tname][1], names[tname][0])] ) + class SchemaTest(fixtures.TestBase): __backend__ = True @@ -1398,8 +1422,6 @@ class SchemaTest(fixtures.TestBase): ) - - # Tests related to engine.reflection @@ -1432,7 +1454,8 @@ def createTables(meta, schema=None): dingalings = Table("dingalings", meta, Column('dingaling_id', sa.Integer, primary_key=True), Column('address_id', sa.Integer, - sa.ForeignKey('%semail_addresses.address_id' % schema_prefix)), + sa.ForeignKey( + '%semail_addresses.address_id' % schema_prefix)), Column('data', sa.String(30)), schema=schema, test_needs_fk=True, ) @@ -1448,6 +1471,7 @@ def createTables(meta, schema=None): return (users, addresses, dingalings) + def createIndexes(con, schema=None): fullname = 'users' if schema: @@ -1455,6 +1479,7 @@ def createIndexes(con, schema=None): query = "CREATE INDEX users_t_idx ON %s (test1, test2)" % fullname con.execute(sa.sql.text(query)) + @testing.requires.views def _create_views(con, schema=None): for table_name in ('users', 'email_addresses'): @@ -1462,10 +1487,10 @@ def _create_views(con, schema=None): if schema: fullname = "%s.%s" % (schema, table_name) view_name = fullname + '_v' - query = "CREATE VIEW %s AS SELECT * FROM %s" % (view_name, - fullname) + query = "CREATE VIEW %s AS SELECT * FROM %s" % (view_name, fullname) con.execute(sa.sql.text(query)) + @testing.requires.views def _drop_views(con, schema=None): for table_name in ('email_addresses', 'users'): @@ -1504,6 +1529,7 @@ class ReverseCasingReflectTest(fixtures.TestBase, AssertsCompiledSQL): 'weird_casing."Col2", weird_casing."col3" ' 'FROM weird_casing') + class CaseSensitiveTest(fixtures.TablesTest): """Nail down case sensitive behaviors, mostly on MySQL.""" __backend__ = True @@ -1539,7 +1565,8 @@ class CaseSensitiveTest(fixtures.TablesTest): ) def test_reflect_via_fk(self): m = MetaData() - t2 = Table("SomeOtherTable", m, autoload=True, autoload_with=testing.db) + t2 = Table("SomeOtherTable", m, autoload=True, + autoload_with=testing.db) eq_(t2.name, "SomeOtherTable") assert "SomeTable" in m.tables @@ -1551,7 +1578,6 @@ class CaseSensitiveTest(fixtures.TablesTest): eq_(t2.name, "sOmEtAbLe") - class ColumnEventsTest(fixtures.RemovesEvents, fixtures.TestBase): __backend__ = True @@ -1584,6 +1610,7 @@ class ColumnEventsTest(fixtures.RemovesEvents, fixtures.TestBase): from sqlalchemy.schema import Table m = MetaData(testing.db) + def column_reflect(insp, table, column_info): if column_info['name'] == col: column_info.update(update) @@ -1620,6 +1647,7 @@ class ColumnEventsTest(fixtures.RemovesEvents, fixtures.TestBase): def test_override_key_fk(self): m = MetaData(testing.db) + def column_reflect(insp, table, column_info): if column_info['name'] == 'q': |