diff options
Diffstat (limited to 'test/sql/test_ddlemit.py')
| -rw-r--r-- | test/sql/test_ddlemit.py | 176 |
1 files changed, 176 insertions, 0 deletions
diff --git a/test/sql/test_ddlemit.py b/test/sql/test_ddlemit.py new file mode 100644 index 000000000..be75f63ec --- /dev/null +++ b/test/sql/test_ddlemit.py @@ -0,0 +1,176 @@ +from sqlalchemy.testing import fixtures +from sqlalchemy.sql.ddl import SchemaGenerator, SchemaDropper +from sqlalchemy.engine import default +from sqlalchemy import MetaData, Table, Column, Integer, Sequence +from sqlalchemy import schema +from sqlalchemy.testing.mock import Mock + +class EmitDDLTest(fixtures.TestBase): + def _mock_connection(self, item_exists): + def has_item(connection, name, schema): + return item_exists(name) + + return Mock(dialect=Mock( + supports_sequences=True, + has_table=Mock(side_effect=has_item), + has_sequence=Mock(side_effect=has_item) + ) + ) + + def _mock_create_fixture(self, checkfirst, tables, + item_exists=lambda item: False): + connection = self._mock_connection(item_exists) + + return SchemaGenerator(connection.dialect, connection, + checkfirst=checkfirst, + tables=tables) + + def _mock_drop_fixture(self, checkfirst, tables, + item_exists=lambda item: True): + connection = self._mock_connection(item_exists) + + return SchemaDropper(connection.dialect, connection, + checkfirst=checkfirst, + tables=tables) + + def _table_fixture(self): + m = MetaData() + + return (m, ) + tuple( + Table('t%d' % i, m, Column('x', Integer)) + for i in range(1, 6) + ) + + def _table_seq_fixture(self): + m = MetaData() + + s1 = Sequence('s1') + s2 = Sequence('s2') + t1 = Table('t1', m, Column("x", Integer, s1, primary_key=True)) + t2 = Table('t2', m, Column("x", Integer, s2, primary_key=True)) + + return m, t1, t2, s1, s2 + + + def test_create_seq_checkfirst(self): + m, t1, t2, s1, s2 = self._table_seq_fixture() + generator = self._mock_create_fixture(True, [t1, t2], + item_exists=lambda t: t not in ("t1", "s1") + ) + + self._assert_create([t1, s1], generator, m) + + + def test_drop_seq_checkfirst(self): + m, t1, t2, s1, s2 = self._table_seq_fixture() + generator = self._mock_drop_fixture(True, [t1, t2], + item_exists=lambda t: t in ("t1", "s1") + ) + + self._assert_drop([t1, s1], generator, m) + + def test_create_collection_checkfirst(self): + m, t1, t2, t3, t4, t5 = self._table_fixture() + generator = self._mock_create_fixture(True, [t2, t3, t4], + item_exists=lambda t: t not in ("t2", "t4") + ) + + self._assert_create_tables([t2, t4], generator, m) + + def test_drop_collection_checkfirst(self): + m, t1, t2, t3, t4, t5 = self._table_fixture() + generator = self._mock_drop_fixture(True, [t2, t3, t4], + item_exists=lambda t: t in ("t2", "t4") + ) + + self._assert_drop_tables([t2, t4], generator, m) + + def test_create_collection_nocheck(self): + m, t1, t2, t3, t4, t5 = self._table_fixture() + generator = self._mock_create_fixture(False, [t2, t3, t4], + item_exists=lambda t: t not in ("t2", "t4") + ) + + self._assert_create_tables([t2, t3, t4], generator, m) + + def test_create_empty_collection(self): + m, t1, t2, t3, t4, t5 = self._table_fixture() + generator = self._mock_create_fixture(True, [], + item_exists=lambda t: t not in ("t2", "t4") + ) + + self._assert_create_tables([], generator, m) + + def test_drop_empty_collection(self): + m, t1, t2, t3, t4, t5 = self._table_fixture() + generator = self._mock_drop_fixture(True, [], + item_exists=lambda t: t in ("t2", "t4") + ) + + self._assert_drop_tables([], generator, m) + + def test_drop_collection_nocheck(self): + m, t1, t2, t3, t4, t5 = self._table_fixture() + generator = self._mock_drop_fixture(False, [t2, t3, t4], + item_exists=lambda t: t in ("t2", "t4") + ) + + self._assert_drop_tables([t2, t3, t4], generator, m) + + def test_create_metadata_checkfirst(self): + m, t1, t2, t3, t4, t5 = self._table_fixture() + generator = self._mock_create_fixture(True, None, + item_exists=lambda t: t not in ("t2", "t4") + ) + + self._assert_create_tables([t2, t4], generator, m) + + def test_drop_metadata_checkfirst(self): + m, t1, t2, t3, t4, t5 = self._table_fixture() + generator = self._mock_drop_fixture(True, None, + item_exists=lambda t: t in ("t2", "t4") + ) + + self._assert_drop_tables([t2, t4], generator, m) + + def test_create_metadata_nocheck(self): + m, t1, t2, t3, t4, t5 = self._table_fixture() + generator = self._mock_create_fixture(False, None, + item_exists=lambda t: t not in ("t2", "t4") + ) + + self._assert_create_tables([t1, t2, t3, t4, t5], generator, m) + + def test_drop_metadata_nocheck(self): + m, t1, t2, t3, t4, t5 = self._table_fixture() + generator = self._mock_drop_fixture(False, None, + item_exists=lambda t: t in ("t2", "t4") + ) + + self._assert_drop_tables([t1, t2, t3, t4, t5], generator, m) + + def _assert_create_tables(self, elements, generator, argument): + self._assert_ddl(schema.CreateTable, elements, generator, argument) + + def _assert_drop_tables(self, elements, generator, argument): + self._assert_ddl(schema.DropTable, elements, generator, argument) + + def _assert_create(self, elements, generator, argument): + self._assert_ddl( + (schema.CreateTable, schema.CreateSequence), + elements, generator, argument) + + def _assert_drop(self, elements, generator, argument): + self._assert_ddl( + (schema.DropTable, schema.DropSequence), + elements, generator, argument) + + def _assert_ddl(self, ddl_cls, elements, generator, argument): + generator.traverse_single(argument) + for call_ in generator.connection.execute.mock_calls: + c = call_[1][0] + assert isinstance(c, ddl_cls) + assert c.element in elements, "element %r was not expected"\ + % c.element + elements.remove(c.element) + assert not elements, "elements remain in list: %r" % elements |