Fix #137: handle legacy unique together constraint (#138)

sparrowt · web-flow · commit c25bed288b0e · 2022-05-27T13:16:29.000-07:00
* TDD: regression test for #137 * Fix #137: drop legacy unique_together CONSTRAINT before AlterField This means that a database previously migrated by an older backend (e.g. django-mssql-backend < v2.6.0, or django-pyodbc-azure) will not explode later when an AlterField type/nullability change acts on a field which is in an old unique_together added by the old backend as a table CONSTRAINT (rather than a filtered unique INDEX like the current backend code now does) * Comment improvements Some related to this ticket, others accrued locally while investigating other things, but I figured they would be helpful * Fix argument order in another test Not necessary but also add kwarg names for clarity while we're here
diff --git a/mssql/functions.py b/mssql/functions.py
@@ -288,6 +288,7 @@ def bulk_update_with_default(self, objs, fields, batch_size=None, default=0):
     return rows_updated
 
 
+# `as_microsoft` called by django.db.models.sql.compiler based on connection.vendor
 ATan2.as_microsoft = sqlserver_atan2
 # Need copy of old In.split_parameter_list_as_sql for other backends to call
 in_split_parameter_list_as_sql = In.split_parameter_list_as_sql
diff --git a/mssql/schema.py b/mssql/schema.py
@@ -234,9 +234,12 @@ def _db_table_constraint_names(self, db_table, column_names=None, column_match_a
                                    unique=None, primary_key=None, index=None, foreign_key=None,
                                    check=None, type_=None, exclude=None, unique_constraint=None):
         """
-        Return all constraint names matching the columns and conditions. Modified from base `_constraint_names`
-        `any_column_matches`=False: (default) only return constraints covering exactly `column_names`
-        `any_column_matches`=True : return any constraints which include at least 1 of `column_names`
+        Return all constraint names matching the columns and conditions.
+        Modified from base `_constraint_names` but with the following new arguments:
+         - `unique_constraint` which explicitly finds unique implemented by CONSTRAINT not by an INDEX
+         - `column_match_any`:
+                False: (default) only return constraints covering exactly `column_names`
+                True : return any constraints which include at least 1 of `column_names`
         """
         if column_names is not None:
             column_names = [
@@ -533,7 +536,7 @@ def _alter_field(self, model, old_field, new_field, old_type, new_type,
 
         # Restore indexes & unique constraints deleted above, SQL Server requires explicit restoration
         if (old_type != new_type or (old_field.null != new_field.null)) and (
-            old_field.column == new_field.column
+            old_field.column == new_field.column  # column rename is handled separately above
         ):
             # Restore unique constraints
             # Note: if nullable they are implemented via an explicit filtered UNIQUE INDEX (not CONSTRAINT)
@@ -688,8 +691,20 @@ def _delete_indexes(self, model, old_field, new_field):
 
     def _delete_unique_constraints(self, model, old_field, new_field, strict=False):
         unique_columns = []
+        # Considering just this column, we only need to drop unique constraints in advance of altering the field
+        # *if* it remains unique - if it wasn't unique before there's nothing to drop; if it won't remain unique
+        # afterwards then that is handled separately in _alter_field
         if old_field.unique and new_field.unique:
             unique_columns.append([old_field.column])
+
+        # Also consider unique_together because, although this is implemented with a filtered unique INDEX now, we
+        # need to handle the possibility that we're acting on a database previously created by an older version of
+        # this backend, where unique_together used to be implemented with a CONSTRAINT
+        for fields in model._meta.unique_together:
+            columns = [model._meta.get_field(field).column for field in fields]
+            if old_field.column in columns:
+                unique_columns.append(columns)
+
         if unique_columns:
             for columns in unique_columns:
                 self._delete_unique_constraint_for_columns(model, columns, strict=strict)
@@ -709,10 +724,12 @@ def _delete_unique_constraint_for_columns(self, model, columns, strict=False, **
                 len(constraint_names),
                 repr(columns),
             ))
+        # Delete constraints which are implemented as a table CONSTRAINT (this may include some created by an
+        # older version of this backend, even if the current version would implement it with an INDEX instead)
         for constraint_name in constraint_names_normal:
             self.execute(self._delete_constraint_sql(self.sql_delete_unique, model, constraint_name))
-        # Unique indexes which are not table constraints must be deleted using the appropriate SQL.
-        # These may exist for example to enforce ANSI-compliant unique constraints on nullable columns.
+        # Delete constraints which are implemented with an explicit index instead (not a table CONSTRAINT)
+        # These are used for example to enforce ANSI-compliant unique constraints on nullable columns.
         for index_name in constraint_names_index:
             self.execute(self._delete_constraint_sql(self.sql_delete_index, model, index_name))
 
diff --git a/testapp/tests/__init__.py b/testapp/tests/__init__.py
@@ -0,0 +1,17 @@
+import django.db
+
+
+def get_constraints(table_name):
+    connection = django.db.connections[django.db.DEFAULT_DB_ALIAS]
+    return connection.introspection.get_constraints(
+        connection.cursor(),
+        table_name=table_name,
+    )
+
+
+def get_constraint_names_where(table_name, **kwargs):
+    return [
+        name
+        for name, details in get_constraints(table_name=table_name).items()
+        if all(details[k] == v for k, v in kwargs.items())
+    ]
diff --git a/testapp/tests/test_constraints.py b/testapp/tests/test_constraints.py
@@ -1,12 +1,15 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the BSD license.
+import logging
 
+import django.db.utils
 from django.db import connections, migrations, models
 from django.db.migrations.state import ProjectState
 from django.db.utils import IntegrityError
 from django.test import TestCase, TransactionTestCase, skipUnlessDBFeature
 
 from mssql.base import DatabaseWrapper
+from . import get_constraint_names_where
 from ..models import (
     Author,
     Editor,
@@ -17,6 +20,8 @@
     TestRenameManyToManyFieldModel,
 )
 
+logger = logging.getLogger('mssql.tests')
+
 
 @skipUnlessDBFeature('supports_nullable_unique_constraints')
 class TestNullableUniqueColumn(TestCase):
@@ -83,6 +88,82 @@ def test_after_type_change(self):
             TestNullableUniqueTogetherModel.objects.create(a='aaa', b='bbb', c='ccc')
 
 
+class TestHandleOldStyleUniqueTogether(TransactionTestCase):
+    """
+    Regression test for https://github.com/microsoft/mssql-django/issues/137
+
+    Start with a unique_together which was created by an older version of this backend code, which implemented
+    it with a table CONSTRAINT instead of a filtered UNIQUE INDEX like the current code does.
+    e.g. django-mssql-backend < v2.6.0 or (before that) all versions of django-pyodbc-azure
+
+    Then alter the type of a column (e.g. max_length of CharField) which is part of that unique_together and
+    check that the (old-style) CONSTRAINT is dropped before (& a new-style UNIQUE INDEX created afterwards).
+    """
+    def test_drop_old_unique_together_constraint(self):
+        class TestMigrationA(migrations.Migration):
+            initial = True
+
+            operations = [
+                migrations.CreateModel(
+                    name='TestHandleOldStyleUniqueTogether',
+                    fields=[
+                        ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                        ('foo', models.CharField(max_length=50)),
+                        ('bar', models.CharField(max_length=50)),
+                    ],
+                ),
+                # Create the unique_together so that Django knows it exists, however we will deliberately drop
+                # it (filtered unique INDEX) below & manually replace with the old implementation (CONSTRAINT)
+                migrations.AlterUniqueTogether(
+                    name='testhandleoldstyleuniquetogether',
+                    unique_together={('foo', 'bar')}
+                ),
+            ]
+
+        class TestMigrationB(migrations.Migration):
+            operations = [
+                # Alter the type of the field to trigger the _alter_field code which drops/recreats indexes/constraints
+                migrations.AlterField(
+                    model_name='testhandleoldstyleuniquetogether',
+                    name='foo',
+                    field=models.CharField(max_length=99),
+                )
+            ]
+
+        migration_a = TestMigrationA(name='test_drop_old_unique_together_constraint_a', app_label='testapp')
+        migration_b = TestMigrationB(name='test_drop_old_unique_together_constraint_b', app_label='testapp')
+
+        connection = connections['default']
+
+        # Setup
+        with connection.schema_editor(atomic=True) as editor:
+            project_state = migration_a.apply(ProjectState(), editor)
+
+        # Manually replace the unique_together-enforcing INDEX with the old implementation using a CONSTRAINT instead
+        # to simulate the state of a database which had been migrated using an older version of this backend
+        table_name = 'testapp_testhandleoldstyleuniquetogether'
+        unique_index_names = get_constraint_names_where(table_name=table_name, index=True, unique=True)
+        assert len(unique_index_names) == 1
+        unique_together_name = unique_index_names[0]
+        logger.debug('Replacing UNIQUE INDEX %s with a CONSTRAINT of the same name', unique_together_name)
+        with connection.schema_editor(atomic=True) as editor:
+            # Out with the new
+            editor.execute('DROP INDEX [%s] ON [%s]' % (unique_together_name, table_name))
+            # In with the old, so that we end up in the state that an old database might be in
+            editor.execute('ALTER TABLE [%s] ADD CONSTRAINT [%s] UNIQUE ([foo], [bar])' % (table_name, unique_together_name))
+
+        # Test by running AlterField
+        with connection.schema_editor(atomic=True) as editor:
+            # If this doesn't explode then all is well. Without the bugfix, the CONSTRAINT wasn't dropped before,
+            # so then re-instating the unique_together using an INDEX of the same name (after altering the field)
+            # would fail due to the presence of a CONSTRAINT (really still an index under the hood) with that name.
+            try:
+                migration_b.apply(project_state, editor)
+            except django.db.utils.DatabaseError as e:
+                logger.exception('Failed to AlterField:')
+                self.fail('Check for regression of issue #137, AlterField failed with exception: %s' % e)
+
+
 class TestRenameManyToManyField(TestCase):
     def test_uniqueness_still_enforced_afterwards(self):
         # Issue https://github.com/microsoft/mssql-django/issues/86
@@ -139,7 +220,7 @@ class TestMigration(migrations.Migration):
                     ),
                 ]
 
-            migration = TestMigration('testapp', 'test_unsupportable_unique_constraint')
+            migration = TestMigration(name='test_unsupportable_unique_constraint', app_label='testapp')
 
             with connection.schema_editor(atomic=True) as editor:
                 with self.assertRaisesRegex(
diff --git a/testapp/tests/test_indexes.py b/testapp/tests/test_indexes.py
@@ -8,6 +8,7 @@
 from django.db.utils import DEFAULT_DB_ALIAS, ConnectionHandler, ProgrammingError
 from django.test import TestCase
 
+from . import get_constraints
 from ..models import (
     TestIndexesRetainedRenamed,
     Choice,
@@ -26,14 +27,6 @@
 logger = logging.getLogger('mssql.tests')
 
 
-def _get_constraints(table_name):
-    connection = django.db.connections[django.db.DEFAULT_DB_ALIAS]
-    return connection.introspection.get_constraints(
-        connection.cursor(),
-        table_name=table_name,
-    )
-
-
 class TestIndexesRetained(TestCase):
     """
     Issue https://github.com/microsoft/mssql-django/issues/14
@@ -46,7 +39,7 @@ def setUpClass(cls):
         super().setUpClass()
         # Pre-fetch which indexes exist for the relevant test model
         # now that all the test migrations have run
-        cls.constraints = _get_constraints(table_name=TestIndexesRetainedRenamed._meta.db_table)
+        cls.constraints = get_constraints(table_name=TestIndexesRetainedRenamed._meta.db_table)
         cls.indexes = {k: v for k, v in cls.constraints.items() if v['index'] is True}
 
     def _assert_index_exists(self, columns):
@@ -97,7 +90,7 @@ def test_correct_indexes_exist(self):
             if not model_cls._meta.managed:
                 # Models where the table is not managed by Django migrations are irrelevant
                 continue
-            model_constraints = _get_constraints(table_name=model_cls._meta.db_table)
+            model_constraints = get_constraints(table_name=model_cls._meta.db_table)
             # Check correct indexes are in place for all fields in model
             for field in model_cls._meta.get_fields():
                 if not hasattr(field, 'column'):
