Fixed #20581 -- Added support for deferrable unique constraints.

4 files changed, 320 insertions, 1 deletions

diff --git a/tests/constraints/models.py b/tests/constraints/models.py
index 98955498d4..3d091f6ccf 100644
--- a/tests/constraints/models.py
+++ b/tests/constraints/models.py
@@ -59,6 +59,28 @@ class UniqueConstraintConditionProduct(models.Model):
         ]
 
 
+class UniqueConstraintDeferrable(models.Model):
+    name = models.CharField(max_length=255)
+    shelf = models.CharField(max_length=31)
+
+    class Meta:
+        required_db_features = {
+            'supports_deferrable_unique_constraints',
+        }
+        constraints = [
+            models.UniqueConstraint(
+                fields=['name'],
+                name='name_init_deferred_uniq',
+                deferrable=models.Deferrable.DEFERRED,
+            ),
+            models.UniqueConstraint(
+                fields=['shelf'],
+                name='sheld_init_immediate_uniq',
+                deferrable=models.Deferrable.IMMEDIATE,
+            ),
+        ]
+
+
 class AbstractModel(models.Model):
     age = models.IntegerField()
 
diff --git a/tests/constraints/tests.py b/tests/constraints/tests.py
index 85edb51aa7..8eb62a940d 100644
--- a/tests/constraints/tests.py
+++ b/tests/constraints/tests.py
@@ -3,11 +3,12 @@ from unittest import mock
 from django.core.exceptions import ValidationError
 from django.db import IntegrityError, connection, models
 from django.db.models.constraints import BaseConstraint
+from django.db.transaction import atomic
 from django.test import SimpleTestCase, TestCase, skipUnlessDBFeature
 
 from .models import (
     ChildModel, Product, UniqueConstraintConditionProduct,
-    UniqueConstraintProduct,
+    UniqueConstraintDeferrable, UniqueConstraintProduct,
 )
 
 
@@ -166,6 +167,20 @@ class UniqueConstraintTests(TestCase):
             ),
         )
 
+    def test_eq_with_deferrable(self):
+        constraint_1 = models.UniqueConstraint(
+            fields=['foo', 'bar'],
+            name='unique',
+            deferrable=models.Deferrable.DEFERRED,
+        )
+        constraint_2 = models.UniqueConstraint(
+            fields=['foo', 'bar'],
+            name='unique',
+            deferrable=models.Deferrable.IMMEDIATE,
+        )
+        self.assertEqual(constraint_1, constraint_1)
+        self.assertNotEqual(constraint_1, constraint_2)
+
     def test_repr(self):
         fields = ['foo', 'bar']
         name = 'unique_fields'
@@ -187,6 +202,18 @@ class UniqueConstraintTests(TestCase):
             "condition=(AND: ('foo', F(bar)))>",
         )
 
+    def test_repr_with_deferrable(self):
+        constraint = models.UniqueConstraint(
+            fields=['foo', 'bar'],
+            name='unique_fields',
+            deferrable=models.Deferrable.IMMEDIATE,
+        )
+        self.assertEqual(
+            repr(constraint),
+            "<UniqueConstraint: fields=('foo', 'bar') name='unique_fields' "
+            "deferrable=Deferrable.IMMEDIATE>",
+        )
+
     def test_deconstruction(self):
         fields = ['foo', 'bar']
         name = 'unique_fields'
@@ -206,6 +233,23 @@ class UniqueConstraintTests(TestCase):
         self.assertEqual(args, ())
         self.assertEqual(kwargs, {'fields': tuple(fields), 'name': name, 'condition': condition})
 
+    def test_deconstruction_with_deferrable(self):
+        fields = ['foo']
+        name = 'unique_fields'
+        constraint = models.UniqueConstraint(
+            fields=fields,
+            name=name,
+            deferrable=models.Deferrable.DEFERRED,
+        )
+        path, args, kwargs = constraint.deconstruct()
+        self.assertEqual(path, 'django.db.models.UniqueConstraint')
+        self.assertEqual(args, ())
+        self.assertEqual(kwargs, {
+            'fields': tuple(fields),
+            'name': name,
+            'deferrable': models.Deferrable.DEFERRED,
+        })
+
     def test_database_constraint(self):
         with self.assertRaises(IntegrityError):
             UniqueConstraintProduct.objects.create(name=self.p1.name, color=self.p1.color)
@@ -238,3 +282,54 @@ class UniqueConstraintTests(TestCase):
     def test_condition_must_be_q(self):
         with self.assertRaisesMessage(ValueError, 'UniqueConstraint.condition must be a Q instance.'):
             models.UniqueConstraint(name='uniq', fields=['name'], condition='invalid')
+
+    @skipUnlessDBFeature('supports_deferrable_unique_constraints')
+    def test_initially_deferred_database_constraint(self):
+        obj_1 = UniqueConstraintDeferrable.objects.create(name='p1', shelf='front')
+        obj_2 = UniqueConstraintDeferrable.objects.create(name='p2', shelf='back')
+
+        def swap():
+            obj_1.name, obj_2.name = obj_2.name, obj_1.name
+            obj_1.save()
+            obj_2.save()
+
+        swap()
+        # Behavior can be changed with SET CONSTRAINTS.
+        with self.assertRaises(IntegrityError):
+            with atomic(), connection.cursor() as cursor:
+                constraint_name = connection.ops.quote_name('name_init_deferred_uniq')
+                cursor.execute('SET CONSTRAINTS %s IMMEDIATE' % constraint_name)
+                swap()
+
+    @skipUnlessDBFeature('supports_deferrable_unique_constraints')
+    def test_initially_immediate_database_constraint(self):
+        obj_1 = UniqueConstraintDeferrable.objects.create(name='p1', shelf='front')
+        obj_2 = UniqueConstraintDeferrable.objects.create(name='p2', shelf='back')
+        obj_1.shelf, obj_2.shelf = obj_2.shelf, obj_1.shelf
+        with self.assertRaises(IntegrityError), atomic():
+            obj_1.save()
+        # Behavior can be changed with SET CONSTRAINTS.
+        with connection.cursor() as cursor:
+            constraint_name = connection.ops.quote_name('sheld_init_immediate_uniq')
+            cursor.execute('SET CONSTRAINTS %s DEFERRED' % constraint_name)
+            obj_1.save()
+            obj_2.save()
+
+    def test_deferrable_with_condition(self):
+        message = 'UniqueConstraint with conditions cannot be deferred.'
+        with self.assertRaisesMessage(ValueError, message):
+            models.UniqueConstraint(
+                fields=['name'],
+                name='name_without_color_unique',
+                condition=models.Q(color__isnull=True),
+                deferrable=models.Deferrable.DEFERRED,
+            )
+
+    def test_invalid_defer_argument(self):
+        message = 'UniqueConstraint.deferrable must be a Deferrable instance.'
+        with self.assertRaisesMessage(ValueError, message):
+            models.UniqueConstraint(
+                fields=['name'],
+                name='name_invalid',
+                deferrable='invalid',
+            )
diff --git a/tests/invalid_models_tests/test_models.py b/tests/invalid_models_tests/test_models.py
index f3f0bc8bd5..6bfdf2e736 100644
--- a/tests/invalid_models_tests/test_models.py
+++ b/tests/invalid_models_tests/test_models.py
@@ -1414,3 +1414,47 @@ class ConstraintsTests(TestCase):
                 ]
 
         self.assertEqual(Model.check(databases=self.databases), [])
+
+    def test_deferrable_unique_constraint(self):
+        class Model(models.Model):
+            age = models.IntegerField()
+
+            class Meta:
+                constraints = [
+                    models.UniqueConstraint(
+                        fields=['age'],
+                        name='unique_age_deferrable',
+                        deferrable=models.Deferrable.DEFERRED,
+                    ),
+                ]
+
+        errors = Model.check(databases=self.databases)
+        expected = [] if connection.features.supports_deferrable_unique_constraints else [
+            Warning(
+                '%s does not support deferrable unique constraints.'
+                % connection.display_name,
+                hint=(
+                    "A constraint won't be created. Silence this warning if "
+                    "you don't care about it."
+                ),
+                obj=Model,
+                id='models.W038',
+            ),
+        ]
+        self.assertEqual(errors, expected)
+
+    def test_deferrable_unique_constraint_required_db_features(self):
+        class Model(models.Model):
+            age = models.IntegerField()
+
+            class Meta:
+                required_db_features = {'supports_deferrable_unique_constraints'}
+                constraints = [
+                    models.UniqueConstraint(
+                        fields=['age'],
+                        name='unique_age_deferrable',
+                        deferrable=models.Deferrable.IMMEDIATE,
+                    ),
+                ]
+
+        self.assertEqual(Model.check(databases=self.databases), [])
diff --git a/tests/migrations/test_operations.py b/tests/migrations/test_operations.py
index 401cae6eae..e985535679 100644
--- a/tests/migrations/test_operations.py
+++ b/tests/migrations/test_operations.py
@@ -393,6 +393,60 @@ class OperationTests(OperationTestBase):
         self.assertEqual(definition[1], [])
         self.assertEqual(definition[2]['options']['constraints'], [partial_unique_constraint])
 
+    def test_create_model_with_deferred_unique_constraint(self):
+        deferred_unique_constraint = models.UniqueConstraint(
+            fields=['pink'],
+            name='deferrable_pink_constraint',
+            deferrable=models.Deferrable.DEFERRED,
+        )
+        operation = migrations.CreateModel(
+            'Pony',
+            [
+                ('id', models.AutoField(primary_key=True)),
+                ('pink', models.IntegerField(default=3)),
+            ],
+            options={'constraints': [deferred_unique_constraint]},
+        )
+        project_state = ProjectState()
+        new_state = project_state.clone()
+        operation.state_forwards('test_crmo', new_state)
+        self.assertEqual(len(new_state.models['test_crmo', 'pony'].options['constraints']), 1)
+        self.assertTableNotExists('test_crmo_pony')
+        # Create table.
+        with connection.schema_editor() as editor:
+            operation.database_forwards('test_crmo', editor, project_state, new_state)
+        self.assertTableExists('test_crmo_pony')
+        Pony = new_state.apps.get_model('test_crmo', 'Pony')
+        Pony.objects.create(pink=1)
+        if connection.features.supports_deferrable_unique_constraints:
+            # Unique constraint is deferred.
+            with transaction.atomic():
+                obj = Pony.objects.create(pink=1)
+                obj.pink = 2
+                obj.save()
+            # Constraint behavior can be changed with SET CONSTRAINTS.
+            with self.assertRaises(IntegrityError):
+                with transaction.atomic(), connection.cursor() as cursor:
+                    quoted_name = connection.ops.quote_name(deferred_unique_constraint.name)
+                    cursor.execute('SET CONSTRAINTS %s IMMEDIATE' % quoted_name)
+                    obj = Pony.objects.create(pink=1)
+                    obj.pink = 3
+                    obj.save()
+        else:
+            Pony.objects.create(pink=1)
+        # Reversal.
+        with connection.schema_editor() as editor:
+            operation.database_backwards('test_crmo', editor, new_state, project_state)
+        self.assertTableNotExists('test_crmo_pony')
+        # Deconstruction.
+        definition = operation.deconstruct()
+        self.assertEqual(definition[0], 'CreateModel')
+        self.assertEqual(definition[1], [])
+        self.assertEqual(
+            definition[2]['options']['constraints'],
+            [deferred_unique_constraint],
+        )
+
     def test_create_model_managers(self):
         """
         The managers on a model are set.
@@ -2046,6 +2100,110 @@ class OperationTests(OperationTestBase):
             'name': 'test_constraint_pony_pink_for_weight_gt_5_uniq',
         })
 
+    def test_add_deferred_unique_constraint(self):
+        app_label = 'test_adddeferred_uc'
+        project_state = self.set_up_test_model(app_label)
+        deferred_unique_constraint = models.UniqueConstraint(
+            fields=['pink'],
+            name='deferred_pink_constraint_add',
+            deferrable=models.Deferrable.DEFERRED,
+        )
+        operation = migrations.AddConstraint('Pony', deferred_unique_constraint)
+        self.assertEqual(
+            operation.describe(),
+            'Create constraint deferred_pink_constraint_add on model Pony',
+        )
+        # Add constraint.
+        new_state = project_state.clone()
+        operation.state_forwards(app_label, new_state)
+        self.assertEqual(len(new_state.models[app_label, 'pony'].options['constraints']), 1)
+        Pony = new_state.apps.get_model(app_label, 'Pony')
+        self.assertEqual(len(Pony._meta.constraints), 1)
+        with connection.schema_editor() as editor:
+            operation.database_forwards(app_label, editor, project_state, new_state)
+        Pony.objects.create(pink=1, weight=4.0)
+        if connection.features.supports_deferrable_unique_constraints:
+            # Unique constraint is deferred.
+            with transaction.atomic():
+                obj = Pony.objects.create(pink=1, weight=4.0)
+                obj.pink = 2
+                obj.save()
+            # Constraint behavior can be changed with SET CONSTRAINTS.
+            with self.assertRaises(IntegrityError):
+                with transaction.atomic(), connection.cursor() as cursor:
+                    quoted_name = connection.ops.quote_name(deferred_unique_constraint.name)
+                    cursor.execute('SET CONSTRAINTS %s IMMEDIATE' % quoted_name)
+                    obj = Pony.objects.create(pink=1, weight=4.0)
+                    obj.pink = 3
+                    obj.save()
+        else:
+            Pony.objects.create(pink=1, weight=4.0)
+        # Reversal.
+        with connection.schema_editor() as editor:
+            operation.database_backwards(app_label, editor, new_state, project_state)
+        # Constraint doesn't work.
+        Pony.objects.create(pink=1, weight=4.0)
+        # Deconstruction.
+        definition = operation.deconstruct()
+        self.assertEqual(definition[0], 'AddConstraint')
+        self.assertEqual(definition[1], [])
+        self.assertEqual(
+            definition[2],
+            {'model_name': 'Pony', 'constraint': deferred_unique_constraint},
+        )
+
+    def test_remove_deferred_unique_constraint(self):
+        app_label = 'test_removedeferred_uc'
+        deferred_unique_constraint = models.UniqueConstraint(
+            fields=['pink'],
+            name='deferred_pink_constraint_rm',
+            deferrable=models.Deferrable.DEFERRED,
+        )
+        project_state = self.set_up_test_model(app_label, constraints=[deferred_unique_constraint])
+        operation = migrations.RemoveConstraint('Pony', deferred_unique_constraint.name)
+        self.assertEqual(
+            operation.describe(),
+            'Remove constraint deferred_pink_constraint_rm from model Pony',
+        )
+        # Remove constraint.
+        new_state = project_state.clone()
+        operation.state_forwards(app_label, new_state)
+        self.assertEqual(len(new_state.models[app_label, 'pony'].options['constraints']), 0)
+        Pony = new_state.apps.get_model(app_label, 'Pony')
+        self.assertEqual(len(Pony._meta.constraints), 0)
+        with connection.schema_editor() as editor:
+            operation.database_forwards(app_label, editor, project_state, new_state)
+        # Constraint doesn't work.
+        Pony.objects.create(pink=1, weight=4.0)
+        Pony.objects.create(pink=1, weight=4.0).delete()
+        # Reversal.
+        with connection.schema_editor() as editor:
+            operation.database_backwards(app_label, editor, new_state, project_state)
+        if connection.features.supports_deferrable_unique_constraints:
+            # Unique constraint is deferred.
+            with transaction.atomic():
+                obj = Pony.objects.create(pink=1, weight=4.0)
+                obj.pink = 2
+                obj.save()
+            # Constraint behavior can be changed with SET CONSTRAINTS.
+            with self.assertRaises(IntegrityError):
+                with transaction.atomic(), connection.cursor() as cursor:
+                    quoted_name = connection.ops.quote_name(deferred_unique_constraint.name)
+                    cursor.execute('SET CONSTRAINTS %s IMMEDIATE' % quoted_name)
+                    obj = Pony.objects.create(pink=1, weight=4.0)
+                    obj.pink = 3
+                    obj.save()
+        else:
+            Pony.objects.create(pink=1, weight=4.0)
+        # Deconstruction.
+        definition = operation.deconstruct()
+        self.assertEqual(definition[0], 'RemoveConstraint')
+        self.assertEqual(definition[1], [])
+        self.assertEqual(definition[2], {
+            'model_name': 'Pony',
+            'name': 'deferred_pink_constraint_rm',
+        })
+
     def test_alter_model_options(self):
         """
         Tests the AlterModelOptions operation.