Adjust unique constraint names based on dialect truncation rules

Adjusted the unique constraint comparison logic in a similar manner as that
of :ticket:`421` did for indexes in order to take into account SQLAlchemy's
own truncation of long constraint names when a naming convention is in use.
Without this step, a name that is truncated by SQLAlchemy based on a unique
constraint naming convention or hardcoded name will not compare properly.

Also remove unused MySQL _legacy_correct_for_dupe_uq_uix which is no
longer used.

Change-Id: I6f709736b845727223c88951a11a899d7dc9b356
Fixes: #647

6 files changed, 168 insertions, 110 deletions

diff --git a/alembic/autogenerate/compare.py b/alembic/autogenerate/compare.py
index 71f947e..92df3fc 100644
--- a/alembic/autogenerate/compare.py
+++ b/alembic/autogenerate/compare.py
@@ -376,7 +376,9 @@ class _ix_constraint_sig(_constraint_sig):
         self.is_unique = bool(const.unique)
 
     def md_name_to_sql_name(self, context):
-        return sqla_compat._get_index_final_name(context.dialect, self.const)
+        return sqla_compat._get_constraint_final_name(
+            self.const, context.dialect
+        )
 
     @property
     def column_names(self):
@@ -499,6 +501,7 @@ def _compare_indexes_and_uniques(
             conn_indexes,
             metadata_unique_constraints,
             metadata_indexes,
+            autogen_context.dialect,
         )
 
     # 3. give the dialect a chance to omit indexes and constraints that
@@ -537,7 +540,6 @@ def _compare_indexes_and_uniques(
 
     conn_uniques_by_name = dict((c.name, c) for c in conn_unique_constraints)
     conn_indexes_by_name = dict((c.name, c) for c in conn_indexes)
-
     conn_names = dict(
         (c.name, c)
         for c in conn_unique_constraints.union(conn_indexes)
@@ -726,6 +728,7 @@ def _correct_for_uq_duplicates_uix(
     conn_indexes,
     metadata_unique_constraints,
     metadata_indexes,
+    dialect,
 ):
     # dedupe unique indexes vs. constraints, since MySQL / Oracle
     # doesn't really have unique constraints as a separate construct.
@@ -733,25 +736,38 @@ def _correct_for_uq_duplicates_uix(
     # that already seem to be defined one way or the other
     # on that side.  This logic was formerly local to MySQL dialect,
     # generalized to Oracle and others. See #276
+
+    # resolve final rendered name for unique constraints defined in the
+    # metadata.   this includes truncation of long names.  naming convention
+    # names currently should already be set as cons.name, however leave this
+    # to the sqla_compat to decide.
+    metadata_cons_names = [
+        (sqla_compat._get_constraint_final_name(cons, dialect), cons)
+        for cons in metadata_unique_constraints
+    ]
+
     metadata_uq_names = set(
-        [
-            cons.name
-            for cons in metadata_unique_constraints
-            if cons.name is not None
-        ]
+        name for name, cons in metadata_cons_names if name is not None
     )
 
     unnamed_metadata_uqs = set(
         [
             _uq_constraint_sig(cons).sig
-            for cons in metadata_unique_constraints
-            if cons.name is None
+            for name, cons in metadata_cons_names
+            if name is None
         ]
     )
 
     metadata_ix_names = set(
-        [cons.name for cons in metadata_indexes if cons.unique]
+        [
+            sqla_compat._get_constraint_final_name(cons, dialect)
+            for cons in metadata_indexes
+            if cons.unique
+        ]
     )
+
+    # for reflection side, names are in their final database form
+    # already since they're from the database
     conn_ix_names = dict(
         (cons.name, cons) for cons in conn_indexes if cons.unique
     )
diff --git a/alembic/ddl/mysql.py b/alembic/ddl/mysql.py
index e919599..ceac853 100644
--- a/alembic/ddl/mysql.py
+++ b/alembic/ddl/mysql.py
@@ -214,56 +214,6 @@ class MySQLImpl(DefaultImpl):
             if idx.name in removed:
                 metadata_indexes.remove(idx)
 
-    def _legacy_correct_for_dupe_uq_uix(
-        self,
-        conn_unique_constraints,
-        conn_indexes,
-        metadata_unique_constraints,
-        metadata_indexes,
-    ):
-
-        # then dedupe unique indexes vs. constraints, since MySQL
-        # doesn't really have unique constraints as a separate construct.
-        # but look in the metadata and try to maintain constructs
-        # that already seem to be defined one way or the other
-        # on that side.  See #276
-        metadata_uq_names = set(
-            [
-                cons.name
-                for cons in metadata_unique_constraints
-                if cons.name is not None
-            ]
-        )
-
-        unnamed_metadata_uqs = set(
-            [
-                compare._uq_constraint_sig(cons).sig
-                for cons in metadata_unique_constraints
-                if cons.name is None
-            ]
-        )
-
-        metadata_ix_names = set(
-            [cons.name for cons in metadata_indexes if cons.unique]
-        )
-        conn_uq_names = dict(
-            (cons.name, cons) for cons in conn_unique_constraints
-        )
-        conn_ix_names = dict(
-            (cons.name, cons) for cons in conn_indexes if cons.unique
-        )
-
-        for overlap in set(conn_uq_names).intersection(conn_ix_names):
-            if overlap not in metadata_uq_names:
-                if (
-                    compare._uq_constraint_sig(conn_uq_names[overlap]).sig
-                    not in unnamed_metadata_uqs
-                ):
-
-                    conn_unique_constraints.discard(conn_uq_names[overlap])
-            elif overlap not in metadata_ix_names:
-                conn_indexes.discard(conn_ix_names[overlap])
-
     def correct_for_autogen_foreignkeys(self, conn_fks, metadata_fks):
         conn_fk_by_sig = dict(
             (compare._fk_constraint_sig(fk).sig, fk) for fk in conn_fks
diff --git a/alembic/util/sqla_compat.py b/alembic/util/sqla_compat.py
index 0104164..c7b43d0 100644
--- a/alembic/util/sqla_compat.py
+++ b/alembic/util/sqla_compat.py
@@ -194,26 +194,10 @@ def _column_kwargs(col):
         return {}
 
 
-def _get_index_final_name(dialect, idx):
-    if sqla_14:
-        return _get_constraint_final_name(idx, dialect)
-
-    # prior to SQLAlchemy 1.4, work around quoting logic to get at the
-    # final compiled name without quotes.
-    if hasattr(idx.name, "quote"):
-        # might be quoted_name, might be truncated_name, keep it the
-        # same
-        quoted_name_cls = type(idx.name)
-        new_name = quoted_name_cls(str(idx.name), quote=False)
-        idx = schema.Index(name=new_name)
-    return dialect.ddl_compiler(dialect, None)._prepared_index_name(idx)
-
-
 def _get_constraint_final_name(constraint, dialect):
-    if sqla_14:
-        if constraint.name is None:
-            return None
-
+    if constraint.name is None:
+        return None
+    elif sqla_14:
         # for SQLAlchemy 1.4 we would like to have the option to expand
         # the use of "deferred" names for constraints as well as to have
         # some flexibility with "None" name and similar; make use of new
@@ -223,7 +207,22 @@ def _get_constraint_final_name(constraint, dialect):
             constraint, _alembic_quote=False
         )
     else:
-        return constraint.name
+
+        # prior to SQLAlchemy 1.4, work around quoting logic to get at the
+        # final compiled name without quotes.
+        if hasattr(constraint.name, "quote"):
+            # might be quoted_name, might be truncated_name, keep it the
+            # same
+            quoted_name_cls = type(constraint.name)
+            new_name = quoted_name_cls(str(constraint.name), quote=False)
+            constraint = constraint.__class__(name=new_name)
+
+        if isinstance(constraint, schema.Index):
+            return dialect.ddl_compiler(dialect, None)._prepared_index_name(
+                constraint
+            )
+        else:
+            return dialect.identifier_preparer.format_constraint(constraint)
 
 
 def _constraint_is_named(constraint, dialect):
diff --git a/docs/build/unreleased/647.rst b/docs/build/unreleased/647.rst
new file mode 100644
index 0000000..bdc95d6
--- /dev/null
+++ b/docs/build/unreleased/647.rst
@@ -0,0 +1,10 @@
+.. change::
+    :tags: bug, autogenerate
+    :tickets: 647
+
+    Adjusted the unique constraint comparison logic in a similar manner as that
+    of :ticket:`421` did for indexes in order to take into account SQLAlchemy's
+    own truncation of long constraint names when a naming convention is in use.
+    Without this step, a name that is truncated by SQLAlchemy based on a unique
+    constraint naming convention or hardcoded name will not compare properly.
+
diff --git a/tests/_autogen_fixtures.py b/tests/_autogen_fixtures.py
index a417a0e..0f3a6cf 100644
--- a/tests/_autogen_fixtures.py
+++ b/tests/_autogen_fixtures.py
@@ -266,37 +266,52 @@ class AutogenFixtureTest(_ComparesFKs):
         opts=None,
         object_filters=_default_object_filters,
         return_ops=False,
+        max_identifier_length=None,
     ):
-        self.metadata, model_metadata = m1, m2
-        for m in util.to_list(self.metadata):
-            m.create_all(self.bind)
-
-        with self.bind.connect() as conn:
-            ctx_opts = {
-                "compare_type": True,
-                "compare_server_default": True,
-                "target_metadata": model_metadata,
-                "upgrade_token": "upgrades",
-                "downgrade_token": "downgrades",
-                "alembic_module_prefix": "op.",
-                "sqlalchemy_module_prefix": "sa.",
-                "include_object": object_filters,
-                "include_schemas": include_schemas,
-            }
-            if opts:
-                ctx_opts.update(opts)
-            self.context = context = MigrationContext.configure(
-                connection=conn, opts=ctx_opts
-            )
-
-            autogen_context = api.AutogenContext(context, model_metadata)
-            uo = ops.UpgradeOps(ops=[])
-            autogenerate._produce_net_changes(autogen_context, uo)
-
-            if return_ops:
-                return uo
-            else:
-                return uo.as_diffs()
+
+        if max_identifier_length:
+            dialect = self.bind.dialect
+            existing_length = dialect.max_identifier_length
+            dialect.max_identifier_length = (
+                dialect._user_defined_max_identifier_length
+            ) = max_identifier_length
+        try:
+            self.metadata, model_metadata = m1, m2
+            for m in util.to_list(self.metadata):
+                m.create_all(self.bind)
+
+            with self.bind.connect() as conn:
+                ctx_opts = {
+                    "compare_type": True,
+                    "compare_server_default": True,
+                    "target_metadata": model_metadata,
+                    "upgrade_token": "upgrades",
+                    "downgrade_token": "downgrades",
+                    "alembic_module_prefix": "op.",
+                    "sqlalchemy_module_prefix": "sa.",
+                    "include_object": object_filters,
+                    "include_schemas": include_schemas,
+                }
+                if opts:
+                    ctx_opts.update(opts)
+                self.context = context = MigrationContext.configure(
+                    connection=conn, opts=ctx_opts
+                )
+
+                autogen_context = api.AutogenContext(context, model_metadata)
+                uo = ops.UpgradeOps(ops=[])
+                autogenerate._produce_net_changes(autogen_context, uo)
+
+                if return_ops:
+                    return uo
+                else:
+                    return uo.as_diffs()
+        finally:
+            if max_identifier_length:
+                dialect = self.bind.dialect
+                dialect.max_identifier_length = (
+                    dialect._user_defined_max_identifier_length
+                ) = existing_length
 
     reports_unnamed_constraints = False
 
diff --git a/tests/test_autogen_indexes.py b/tests/test_autogen_indexes.py
index 416444c..6a8800b 100644
--- a/tests/test_autogen_indexes.py
+++ b/tests/test_autogen_indexes.py
@@ -297,6 +297,74 @@ class AutogenerateUniqueIndexTest(AutogenFixtureTest, TestBase):
         diffs = self._fixture(m1, m2)
         eq_(diffs, [])
 
+    def test_nothing_uq_changed_labels_were_truncated(self):
+        m1 = MetaData(
+            naming_convention={
+                "ix": "index_%(table_name)s_%(column_0_label)s",
+                "uq": "unique_%(table_name)s_%(column_0_label)s",
+            }
+        )
+        m2 = MetaData(
+            naming_convention={
+                "ix": "index_%(table_name)s_%(column_0_label)s",
+                "uq": "unique_%(table_name)s_%(column_0_label)s",
+            }
+        )
+
+        Table(
+            "nothing_changed",
+            m1,
+            Column("id1", Integer, primary_key=True),
+            Column("id2", Integer, primary_key=True),
+            Column("a_long_name", String(20), unique=True),
+            mysql_engine="InnoDB",
+        )
+
+        Table(
+            "nothing_changed",
+            m2,
+            Column("id1", Integer, primary_key=True),
+            Column("id2", Integer, primary_key=True),
+            Column("a_long_name", String(20), unique=True),
+            mysql_engine="InnoDB",
+        )
+        diffs = self._fixture(m1, m2, max_identifier_length=30)
+        eq_(diffs, [])
+
+    def test_nothing_ix_changed_labels_were_truncated(self):
+        m1 = MetaData(
+            naming_convention={
+                "ix": "index_%(table_name)s_%(column_0_label)s",
+                "uq": "unique_%(table_name)s_%(column_0_label)s",
+            }
+        )
+        m2 = MetaData(
+            naming_convention={
+                "ix": "index_%(table_name)s_%(column_0_label)s",
+                "uq": "unique_%(table_name)s_%(column_0_label)s",
+            }
+        )
+
+        Table(
+            "nothing_changed",
+            m1,
+            Column("id1", Integer, primary_key=True),
+            Column("id2", Integer, primary_key=True),
+            Column("a_particularly_long_column_name", String(20), index=True),
+            mysql_engine="InnoDB",
+        )
+
+        Table(
+            "nothing_changed",
+            m2,
+            Column("id1", Integer, primary_key=True),
+            Column("id2", Integer, primary_key=True),
+            Column("a_particularly_long_column_name", String(20), index=True),
+            mysql_engine="InnoDB",
+        )
+        diffs = self._fixture(m1, m2, max_identifier_length=30)
+        eq_(diffs, [])
+
     def test_nothing_changed_two(self):
         m1 = MetaData()
         m2 = MetaData()