Merge branch 'master' into ticket_2501ticket_2501

Conflicts:
	lib/sqlalchemy/orm/mapper.py

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