- Fixed an issue with unique constraint autogenerate detection where

a named ``UniqueConstraint`` on both sides with column changes would
render with the "add" operation before the "drop", requiring the
user to reverse the order manually.
- reorganize the index/unique autogenerate test into individual test cases;
ideally the whole test suite would be broken out like this for those big
tests

diff --git a/alembic/autogenerate/compare.py b/alembic/autogenerate/compare.py
index 3b2e1d7698a9fb5894fd920e93396260d86ad982..4fb61ab887f21884b0e1e23aee8a67cb0e9b4d15 100644 (file)
--- a/alembic/autogenerate/compare.py
+++ b/alembic/autogenerate/compare.py
@@ -58,6 +58,7 @@ def _compare_tables(conn_table_names, metadata_table_names,
         name = '%s.%s' % (s, tname) if s else tname
         metadata_table = metadata.tables[name]
         conn_table = existing_metadata.tables[name]
+
         if _run_filters(metadata_table, tname, "table", False, conn_table, object_filters):
             _compare_columns(s, tname, object_filters,
                     conn_table,
@@ -187,6 +188,17 @@ def _compare_uniques(schema, tname, object_filters, conn_table,
     )
     c_keys = set(c_objs)
 
+    c_obj_by_name = dict((uq.name, uq) for uq in c_objs.values())
+    m_obj_by_name = dict((uq.name, uq) for uq in m_objs.values())
+
+    # for constraints that are named the same on both sides,
+    # keep these as a single "drop"/"add" so that the ordering
+    # comes out correctly
+    names_equal = set(c_obj_by_name).intersection(m_obj_by_name)
+    for name_equal in names_equal:
+        m_keys.remove(_uq_constraint_sig(m_obj_by_name[name_equal]))
+        c_keys.remove(_uq_constraint_sig(c_obj_by_name[name_equal]))
+
     for key in m_keys.difference(c_keys):
         meta_constraint = m_objs[key]
         diffs.append(("add_constraint", meta_constraint))
@@ -202,9 +214,11 @@ def _compare_uniques(schema, tname, object_filters, conn_table,
             key, tname
         )
 
-    for key in m_keys.intersection(c_keys):
-        meta_constraint = m_objs[key]
-        conn_constraint = c_objs[key]
+    for meta_constraint, conn_constraint in [
+        (m_objs[key], c_objs[key]) for key in m_keys.intersection(c_keys)
+    ] + [
+        (m_obj_by_name[key], c_obj_by_name[key]) for key in names_equal
+    ]:
         conn_cols = [col.name for col in conn_constraint.columns]
         meta_cols = [col.name for col in meta_constraint.columns]
 

diff --git a/docs/build/changelog.rst b/docs/build/changelog.rst
index f960c59f541263d9464058d677e965ae0818d97b..c51842ba21271ade4e15e236dc09e8f2a781fec4 100644 (file)
--- a/docs/build/changelog.rst
+++ b/docs/build/changelog.rst
@@ -6,6 +6,14 @@ Changelog
 .. changelog::
     :version: 0.6.2
 
+    .. change::
+      :tags: bug
+
+      Fixed an issue with unique constraint autogenerate detection where
+      a named ``UniqueConstraint`` on both sides with column changes would
+      render with the "add" operation before the "drop", requiring the
+      user to reverse the order manually.
+
     .. change::
       :tags: feature, mssql
 

diff --git a/tests/test_autogenerate.py b/tests/test_autogenerate.py
index 47c207dced33f2d0c1e5b6876835b372d9ed8d65..5e2a4881979620bfc1203cec5f6e66f826ddcfbc 100644 (file)
--- a/tests/test_autogenerate.py
+++ b/tests/test_autogenerate.py
@@ -103,7 +103,7 @@ def _default_include_object(obj, name, type_, reflected, compare_to):
     if type_ == "table":
         return name in ("parent", "child",
                                 "user", "order", "item",
-                                "address", "extra")
+                                "address", "extra", "col_change")
     else:
         return True
 
@@ -750,100 +750,169 @@ class AutogenerateDiffTest(AutogenTest, TestCase):
 
 
 
-class AutogenerateUniqueIndexTest(AutogenTest, TestCase):
+class AutogenerateUniqueIndexTest(TestCase):
 
-    @classmethod
-    def _get_db_schema(cls):
-        m = MetaData()
 
-        Table('user', m,
+    def _fixture_one(self):
+        m1 = MetaData()
+        m2 = MetaData()
+
+        Table('user', m1,
             Column('id', Integer, primary_key=True),
             Column('name', String(50), nullable=False, index=True),
             Column('a1', Text, server_default="x")
         )
 
-        Table('address', m,
+        Table('user', m2,
             Column('id', Integer, primary_key=True),
-            Column('email_address', String(100), nullable=False),
-            Column('qpr', String(10), index=True),
-        )
-
-        Table('order', m,
-            Column('order_id', Integer, primary_key=True),
-            Column('amount', Numeric(10, 2), nullable=True),
-            Column('user_id', Integer, ForeignKey('user.id')),
-            CheckConstraint('amount > -1', name='ck_order_amount'),
-            UniqueConstraint('order_id', 'user_id',
-                name='order_order_id_user_id_unique'
-            ),
-            Index('order_user_id_amount_idx', 'user_id', 'amount')
+            Column('name', String(50), nullable=False),
+            Column('a1', Text, server_default="x"),
+            UniqueConstraint("name", name="uq_user_name")
         )
+        return m1, m2
 
-        Table('item', m,
-                Column('x', Integer),
-                UniqueConstraint('x', name="db_generated_name")
-            )
-        return m
+    def test_index_flag_becomes_named_unique_constraint(self):
+        diffs = self._fixture(self._fixture_one)
 
+        eq_(diffs[0][0], "add_constraint")
+        eq_(diffs[0][1].name, "uq_user_name")
 
-    @classmethod
-    def _get_model_schema(cls):
-        m = MetaData()
+        eq_(diffs[1][0], "remove_index")
+        eq_(diffs[1][1].name, "ix_user_name")
 
-        Table('user', m,
+    def _fixture_two(self):
+        m1 = MetaData()
+        m2 = MetaData()
+        Table('address', m1,
             Column('id', Integer, primary_key=True),
-            Column('name', String(50), nullable=False),
-            Column('a1', Text, server_default="x"),
-            UniqueConstraint("name", name="uq_user_name")
+            Column('email_address', String(100), nullable=False),
+            Column('qpr', String(10), index=True),
         )
-
-        Table('address', m,
+        Table('address', m2,
             Column('id', Integer, primary_key=True),
             Column('email_address', String(100), nullable=False),
             Column('qpr', String(10), index=True),
             UniqueConstraint("email_address", name="uq_email_address")
         )
+        return m1, m2
 
-        Table('order', m,
+    def test_add_unique_constraint(self):
+        diffs = self._fixture(self._fixture_two)
+        eq_(diffs[0][0], "add_constraint")
+        eq_(diffs[0][1].name, "uq_email_address")
+
+    def _fixture_three(self):
+        m1 = MetaData()
+        m2 = MetaData()
+        Table('order', m1,
             Column('order_id', Integer, primary_key=True),
             Column('amount', Numeric(10, 2), nullable=True),
-            Column('user_id', Integer, ForeignKey('user.id')),
+            Column('user_id', Integer),
+            UniqueConstraint('order_id', 'user_id',
+                name='order_order_id_user_id_unique'
+            ),
+            Index('order_user_id_amount_idx', 'user_id', 'amount')
+        )
+
+        Table('order', m2,
+            Column('order_id', Integer, primary_key=True),
+            Column('amount', Numeric(10, 2), nullable=True),
+            Column('user_id', Integer),
             UniqueConstraint('order_id', 'user_id',
                 name='order_order_id_user_id_unique'
             ),
             Index('order_user_id_amount_idx', 'user_id', 'amount', unique=True),
-            CheckConstraint('amount >= 0', name='ck_order_amount')
         )
+        return m1, m2
+
+    def test_index_becomes_unique(self):
+        diffs = self._fixture(self._fixture_three)
+        eq_(diffs[0][0], "remove_index")
+        eq_(diffs[0][1].name, "order_user_id_amount_idx")
+        eq_(diffs[0][1].unique, False)
+
+        eq_(diffs[1][0], "add_index")
+        eq_(diffs[1][1].name, "order_user_id_amount_idx")
+        eq_(diffs[1][1].unique, True)
+
+    def _fixture_four(self):
+        m1 = MetaData()
+        m2 = MetaData()
+        Table('item', m1,
+                Column('x', Integer),
+                UniqueConstraint('x', name="db_generated_name")
+            )
 
         # test mismatch between unique=True and
         # named uq constraint
-        Table('item', m,
+        Table('item', m2,
                 Column('x', Integer, unique=True)
             )
+        return m1, m2
+
+    def test_mismatch_db_named_col_flag(self):
+        diffs = self._fixture(self._fixture_four)
 
-        Table('extra', m,
+        eq_(diffs, [])
+
+    def _fixture_five(self):
+        m1 = MetaData()
+        m2 = MetaData()
+        Table('extra', m2,
                 Column('foo', Integer, index=True),
                 Column('bar', Integer),
                 Index('newtable_idx', 'bar')
             )
-        return m
+        return m1, m2
 
-    @classmethod
-    @requires_07
-    def setup_class(cls):
+    def test_new_table_added(self):
+        diffs = self._fixture(self._fixture_five)
+
+        eq_(diffs[0][0], "add_table")
+
+        eq_(diffs[1][0], "add_index")
+        eq_(diffs[1][1].name, "ix_extra_foo")
+
+        eq_(diffs[2][0], "add_index")
+        eq_(diffs[2][1].name, "newtable_idx")
+
+    def _fixture_six(self):
+        m1 = MetaData()
+        m2 = MetaData()
+        Table('col_change', m1,
+                Column('x', Integer),
+                Column('y', Integer),
+                UniqueConstraint('x', name="nochange")
+            )
+        Table('col_change', m2,
+                Column('x', Integer),
+                Column('y', Integer),
+                UniqueConstraint('x', 'y', name="nochange")
+            )
+        return m1, m2
+
+    def test_named_cols_changed(self):
+        diffs = self._fixture(self._fixture_six)
+
+        eq_(diffs[0][0], "remove_constraint")
+        eq_(diffs[0][1].name, "nochange")
+
+        eq_(diffs[1][0], "add_constraint")
+        eq_(diffs[1][1].name, "nochange")
+
+    def _fixture(self, fn):
         staging_env()
-        cls.bind = cls._get_bind()
-        cls.m2 = cls._get_db_schema()
-        cls.m2.create_all(cls.bind)
-        cls.m5 = cls._get_model_schema()
+        self.bind = sqlite_db()
+        self.metadata, model_metadata = fn()
+        self.metadata.create_all(self.bind)
 
-        conn = cls.bind.connect()
-        cls.context = context = MigrationContext.configure(
+        conn = self.bind.connect()
+        self.context = context = MigrationContext.configure(
             connection=conn,
             opts={
                 'compare_type': True,
                 'compare_server_default': True,
-                'target_metadata': cls.m5,
+                'target_metadata': model_metadata,
                 'upgrade_token': "upgrades",
                 'downgrade_token': "downgrades",
                 'alembic_module_prefix': 'op.',
@@ -852,61 +921,25 @@ class AutogenerateUniqueIndexTest(AutogenTest, TestCase):
         )
 
         connection = context.bind
-        cls.autogen_context = {
+        autogen_context = {
             'imports': set(),
             'connection': connection,
             'dialect': connection.dialect,
             'context': context
             }
-
-    @classmethod
-    def teardown_class(cls):
-        cls.m2.drop_all(cls.bind)
-        clear_staging_env()
-
-    def test_diffs(self):
-        """test generation of diff rules"""
-
-        metadata = self.m5
-        connection = self.context.bind
         diffs = []
-        autogenerate._produce_net_changes(connection, metadata, diffs,
-                                          self.autogen_context,
+        autogenerate._produce_net_changes(connection, model_metadata, diffs,
+                                          autogen_context,
                                           object_filters=_default_object_filters,
                                     )
+        return diffs
 
-        eq_(diffs[0][0], "add_table")
-        eq_(diffs[0][1].name, "extra")
-
-        eq_(diffs[1][0], "add_index")
-        eq_(diffs[2][0], "add_index")
-
-        eq_(set([
-                    diffs[1][1].name,
-                    diffs[2][1].name
-                ]),
-                set([
-                    "ix_extra_foo",  # sqlalchemy's naming scheme
-                    "newtable_idx"
-                ])
-        )
-
-        eq_(diffs[3][0], "add_constraint")
-        eq_(diffs[3][1].table.name, "address")
-
-        eq_(diffs[4][0], "remove_index")
-        eq_(diffs[4][1].name, "order_user_id_amount_idx")
-        eq_(diffs[4][1].unique, False)
+    def tearDown(self):
+        self.metadata.drop_all(self.bind)
+        clear_staging_env()
 
-        eq_(diffs[5][0], "add_index")
-        eq_(diffs[5][1].name, "order_user_id_amount_idx")
-        eq_(diffs[5][1].unique, True)
 
-        eq_(diffs[6][0], "add_constraint")
-        eq_(diffs[6][1].table.name, "user")
 
-        eq_(diffs[7][0], "remove_index")
-        eq_(diffs[7][1].name, "ix_user_name")
 
 class PGUniqueIndexTest(AutogenerateUniqueIndexTest):
     @classmethod