Fixed #11964 -- Added support for database check constraints.

9 files changed, 371 insertions, 9 deletions

diff --git a/tests/constraints/__init__.py b/tests/constraints/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
--- /dev/null
+++ b/tests/constraints/__init__.py
diff --git a/tests/constraints/models.py b/tests/constraints/models.py
new file mode 100644
index 0000000000..de49fa2765
--- /dev/null
+++ b/tests/constraints/models.py
@@ -0,0 +1,15 @@
+from django.db import models
+
+
+class Product(models.Model):
+    name = models.CharField(max_length=255)
+    price = models.IntegerField()
+    discounted_price = models.IntegerField()
+
+    class Meta:
+        constraints = [
+            models.CheckConstraint(
+                models.Q(price__gt=models.F('discounted_price')),
+                'price_gt_discounted_price'
+            )
+        ]
diff --git a/tests/constraints/tests.py b/tests/constraints/tests.py
new file mode 100644
index 0000000000..19573dffa1
--- /dev/null
+++ b/tests/constraints/tests.py
@@ -0,0 +1,30 @@
+from django.db import IntegrityError, models
+from django.test import TestCase, skipUnlessDBFeature
+
+from .models import Product
+
+
+class CheckConstraintTests(TestCase):
+    def test_repr(self):
+        constraint = models.Q(price__gt=models.F('discounted_price'))
+        name = 'price_gt_discounted_price'
+        check = models.CheckConstraint(constraint, name)
+        self.assertEqual(
+            repr(check),
+            "<CheckConstraint: constraint='{}' name='{}'>".format(constraint, name),
+        )
+
+    def test_deconstruction(self):
+        constraint = models.Q(price__gt=models.F('discounted_price'))
+        name = 'price_gt_discounted_price'
+        check = models.CheckConstraint(constraint, name)
+        path, args, kwargs = check.deconstruct()
+        self.assertEqual(path, 'django.db.models.CheckConstraint')
+        self.assertEqual(args, ())
+        self.assertEqual(kwargs, {'constraint': constraint, 'name': name})
+
+    @skipUnlessDBFeature('supports_table_check_constraints')
+    def test_model_constraint(self):
+        Product.objects.create(name='Valid', price=10, discounted_price=5)
+        with self.assertRaises(IntegrityError):
+            Product.objects.create(name='Invalid', price=10, discounted_price=20)
diff --git a/tests/invalid_models_tests/test_models.py b/tests/invalid_models_tests/test_models.py
index 19ec21c9ae..9dd2fd1f06 100644
--- a/tests/invalid_models_tests/test_models.py
+++ b/tests/invalid_models_tests/test_models.py
@@ -1,10 +1,10 @@
 import unittest
 
 from django.conf import settings
-from django.core.checks import Error
+from django.core.checks import Error, Warning
 from django.core.checks.model_checks import _check_lazy_references
 from django.core.exceptions import ImproperlyConfigured
-from django.db import connections, models
+from django.db import connection, connections, models
 from django.db.models.signals import post_init
 from django.test import SimpleTestCase
 from django.test.utils import isolate_apps, override_settings
@@ -972,3 +972,26 @@ class OtherModelTests(SimpleTestCase):
                 id='signals.E001',
             ),
         ])
+
+
+@isolate_apps('invalid_models_tests')
+class ConstraintsTests(SimpleTestCase):
+    def test_check_constraints(self):
+        class Model(models.Model):
+            age = models.IntegerField()
+
+            class Meta:
+                constraints = [models.CheckConstraint(models.Q(age__gte=18), 'is_adult')]
+
+        errors = Model.check()
+        warn = Warning(
+            '%s does not support check 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.W027',
+        )
+        expected = [] if connection.features.supports_table_check_constraints else [warn, warn]
+        self.assertCountEqual(errors, expected)
diff --git a/tests/migrations/test_autodetector.py b/tests/migrations/test_autodetector.py
index fd1bc383b9..1fde1ba466 100644
--- a/tests/migrations/test_autodetector.py
+++ b/tests/migrations/test_autodetector.py
@@ -61,6 +61,12 @@ class AutodetectorTests(TestCase):
         ("id", models.AutoField(primary_key=True)),
         ("name", models.CharField(max_length=200, default='Ada Lovelace')),
     ])
+    author_name_check_constraint = ModelState("testapp", "Author", [
+        ("id", models.AutoField(primary_key=True)),
+        ("name", models.CharField(max_length=200)),
+    ],
+        {'constraints': [models.CheckConstraint(models.Q(name__contains='Bob'), 'name_contains_bob')]},
+    )
     author_dates_of_birth_auto_now = ModelState("testapp", "Author", [
         ("id", models.AutoField(primary_key=True)),
         ("date_of_birth", models.DateField(auto_now=True)),
@@ -1389,6 +1395,40 @@ class AutodetectorTests(TestCase):
         added_index = models.Index(fields=['title', 'author'], name='book_author_title_idx')
         self.assertOperationAttributes(changes, 'otherapp', 0, 1, model_name='book', index=added_index)
 
+    def test_create_model_with_check_constraint(self):
+        """Test creation of new model with constraints already defined."""
+        author = ModelState('otherapp', 'Author', [
+            ('id', models.AutoField(primary_key=True)),
+            ('name', models.CharField(max_length=200)),
+        ], {'constraints': [models.CheckConstraint(models.Q(name__contains='Bob'), 'name_contains_bob')]})
+        changes = self.get_changes([], [author])
+        added_constraint = models.CheckConstraint(models.Q(name__contains='Bob'), 'name_contains_bob')
+        # Right number of migrations?
+        self.assertEqual(len(changes['otherapp']), 1)
+        # Right number of actions?
+        migration = changes['otherapp'][0]
+        self.assertEqual(len(migration.operations), 2)
+        # Right actions order?
+        self.assertOperationTypes(changes, 'otherapp', 0, ['CreateModel', 'AddConstraint'])
+        self.assertOperationAttributes(changes, 'otherapp', 0, 0, name='Author')
+        self.assertOperationAttributes(changes, 'otherapp', 0, 1, model_name='author', constraint=added_constraint)
+
+    def test_add_constraints(self):
+        """Test change detection of new constraints."""
+        changes = self.get_changes([self.author_name], [self.author_name_check_constraint])
+        self.assertNumberMigrations(changes, 'testapp', 1)
+        self.assertOperationTypes(changes, 'testapp', 0, ['AddConstraint'])
+        added_constraint = models.CheckConstraint(models.Q(name__contains='Bob'), 'name_contains_bob')
+        self.assertOperationAttributes(changes, 'testapp', 0, 0, model_name='author', constraint=added_constraint)
+
+    def test_remove_constraints(self):
+        """Test change detection of removed constraints."""
+        changes = self.get_changes([self.author_name_check_constraint], [self.author_name])
+        # Right number/type of migrations?
+        self.assertNumberMigrations(changes, 'testapp', 1)
+        self.assertOperationTypes(changes, 'testapp', 0, ['RemoveConstraint'])
+        self.assertOperationAttributes(changes, 'testapp', 0, 0, model_name='author', name='name_contains_bob')
+
     def test_add_foo_together(self):
         """Tests index/unique_together detection."""
         changes = self.get_changes([self.author_empty, self.book], [self.author_empty, self.book_foo_together])
@@ -1520,7 +1560,7 @@ class AutodetectorTests(TestCase):
         self.assertNumberMigrations(changes, "testapp", 1)
         self.assertOperationTypes(changes, "testapp", 0, ["CreateModel"])
         self.assertOperationAttributes(
-            changes, "testapp", 0, 0, name="AuthorProxy", options={"proxy": True, "indexes": []}
+            changes, "testapp", 0, 0, name="AuthorProxy", options={"proxy": True, "indexes": [], "constraints": []}
         )
         # Now, we test turning a proxy model into a non-proxy model
         # It should delete the proxy then make the real one
diff --git a/tests/migrations/test_base.py b/tests/migrations/test_base.py
index 84a5117751..7fcbaffd24 100644
--- a/tests/migrations/test_base.py
+++ b/tests/migrations/test_base.py
@@ -67,6 +67,17 @@ class MigrationTestBase(TransactionTestCase):
     def assertIndexNotExists(self, table, columns):
         return self.assertIndexExists(table, columns, False)
 
+    def assertConstraintExists(self, table, name, value=True, using='default'):
+        with connections[using].cursor() as cursor:
+            constraints = connections[using].introspection.get_constraints(cursor, table).items()
+            self.assertEqual(
+                value,
+                any(c['check'] for n, c in constraints if n == name),
+            )
+
+    def assertConstraintNotExists(self, table, name):
+        return self.assertConstraintExists(table, name, False)
+
     def assertFKExists(self, table, columns, to, value=True, using='default'):
         with connections[using].cursor() as cursor:
             self.assertEqual(
diff --git a/tests/migrations/test_operations.py b/tests/migrations/test_operations.py
index d70feaacdb..b1581042f7 100644
--- a/tests/migrations/test_operations.py
+++ b/tests/migrations/test_operations.py
@@ -53,7 +53,7 @@ class OperationTestBase(MigrationTestBase):
     def set_up_test_model(
             self, app_label, second_model=False, third_model=False, index=False, multicol_index=False,
             related_model=False, mti_model=False, proxy_model=False, manager_model=False,
-            unique_together=False, options=False, db_table=None, index_together=False):
+            unique_together=False, options=False, db_table=None, index_together=False, check_constraint=False):
         """
         Creates a test model state and database table.
         """
@@ -106,6 +106,11 @@ class OperationTestBase(MigrationTestBase):
                 "Pony",
                 models.Index(fields=["pink", "weight"], name="pony_test_idx")
             ))
+        if check_constraint:
+            operations.append(migrations.AddConstraint(
+                "Pony",
+                models.CheckConstraint(models.Q(pink__gt=2), name="pony_test_constraint")
+            ))
         if second_model:
             operations.append(migrations.CreateModel(
                 "Stable",
@@ -462,6 +467,45 @@ class OperationTests(OperationTestBase):
         self.assertTableNotExists("test_crummo_unmanagedpony")
         self.assertTableExists("test_crummo_pony")
 
+    @skipUnlessDBFeature('supports_table_check_constraints')
+    def test_create_model_with_constraint(self):
+        where = models.Q(pink__gt=2)
+        check_constraint = models.CheckConstraint(where, name='test_constraint_pony_pink_gt_2')
+        operation = migrations.CreateModel(
+            "Pony",
+            [
+                ("id", models.AutoField(primary_key=True)),
+                ("pink", models.IntegerField(default=3)),
+            ],
+            options={'constraints': [check_constraint]},
+        )
+
+        # Test the state alteration
+        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)
+
+        # Test database alteration
+        self.assertTableNotExists("test_crmo_pony")
+        with connection.schema_editor() as editor:
+            operation.database_forwards("test_crmo", editor, project_state, new_state)
+        self.assertTableExists("test_crmo_pony")
+        with connection.cursor() as cursor:
+            with self.assertRaises(IntegrityError):
+                cursor.execute("INSERT INTO test_crmo_pony (id, pink) VALUES (1, 1)")
+
+        # Test reversal
+        with connection.schema_editor() as editor:
+            operation.database_backwards("test_crmo", editor, new_state, project_state)
+        self.assertTableNotExists("test_crmo_pony")
+
+        # Test deconstruction
+        definition = operation.deconstruct()
+        self.assertEqual(definition[0], "CreateModel")
+        self.assertEqual(definition[1], [])
+        self.assertEqual(definition[2]['options']['constraints'], [check_constraint])
+
     def test_create_model_managers(self):
         """
         The managers on a model are set.
@@ -1708,6 +1752,87 @@ class OperationTests(OperationTestBase):
         operation = migrations.AlterIndexTogether("Pony", None)
         self.assertEqual(operation.describe(), "Alter index_together for Pony (0 constraint(s))")
 
+    @skipUnlessDBFeature('supports_table_check_constraints')
+    def test_add_constraint(self):
+        """Test the AddConstraint operation."""
+        project_state = self.set_up_test_model('test_addconstraint')
+
+        where = models.Q(pink__gt=2)
+        check_constraint = models.CheckConstraint(where, name='test_constraint_pony_pink_gt_2')
+        operation = migrations.AddConstraint('Pony', check_constraint)
+        self.assertEqual(operation.describe(), 'Create constraint test_constraint_pony_pink_gt_2 on model Pony')
+
+        new_state = project_state.clone()
+        operation.state_forwards('test_addconstraint', new_state)
+        self.assertEqual(len(new_state.models['test_addconstraint', 'pony'].options['constraints']), 1)
+
+        # Test database alteration
+        with connection.cursor() as cursor:
+            with atomic():
+                cursor.execute("INSERT INTO test_addconstraint_pony (id, pink, weight) VALUES (1, 1, 1.0)")
+            cursor.execute("DELETE FROM test_addconstraint_pony")
+
+        with connection.schema_editor() as editor:
+            operation.database_forwards("test_addconstraint", editor, project_state, new_state)
+
+        with connection.cursor() as cursor:
+            with self.assertRaises(IntegrityError):
+                cursor.execute("INSERT INTO test_addconstraint_pony (id, pink, weight) VALUES (1, 1, 1.0)")
+
+        # Test reversal
+        with connection.schema_editor() as editor:
+            operation.database_backwards("test_addconstraint", editor, new_state, project_state)
+
+        with connection.cursor() as cursor:
+            with atomic():
+                cursor.execute("INSERT INTO test_addconstraint_pony (id, pink, weight) VALUES (1, 1, 1.0)")
+            cursor.execute("DELETE FROM test_addconstraint_pony")
+
+        # Test deconstruction
+        definition = operation.deconstruct()
+        self.assertEqual(definition[0], "AddConstraint")
+        self.assertEqual(definition[1], [])
+        self.assertEqual(definition[2], {'model_name': "Pony", 'constraint': check_constraint})
+
+    @skipUnlessDBFeature('supports_table_check_constraints')
+    def test_remove_constraint(self):
+        """Test the RemoveConstraint operation."""
+        project_state = self.set_up_test_model("test_removeconstraint", check_constraint=True)
+        self.assertTableExists("test_removeconstraint_pony")
+        operation = migrations.RemoveConstraint("Pony", "pony_test_constraint")
+        self.assertEqual(operation.describe(), "Remove constraint pony_test_constraint from model Pony")
+        new_state = project_state.clone()
+        operation.state_forwards("test_removeconstraint", new_state)
+        # Test state alteration
+        self.assertEqual(len(new_state.models["test_removeconstraint", "pony"].options['constraints']), 0)
+
+        with connection.cursor() as cursor:
+            with self.assertRaises(IntegrityError):
+                cursor.execute("INSERT INTO test_removeconstraint_pony (id, pink, weight) VALUES (1, 1, 1.0)")
+
+        # Test database alteration
+        with connection.schema_editor() as editor:
+            operation.database_forwards("test_removeconstraint", editor, project_state, new_state)
+
+        with connection.cursor() as cursor:
+            with atomic():
+                cursor.execute("INSERT INTO test_removeconstraint_pony (id, pink, weight) VALUES (1, 1, 1.0)")
+            cursor.execute("DELETE FROM test_removeconstraint_pony")
+
+        # Test reversal
+        with connection.schema_editor() as editor:
+            operation.database_backwards("test_removeconstraint", editor, new_state, project_state)
+
+        with connection.cursor() as cursor:
+            with self.assertRaises(IntegrityError):
+                cursor.execute("INSERT INTO test_removeconstraint_pony (id, pink, weight) VALUES (1, 1, 1.0)")
+
+        # Test deconstruction
+        definition = operation.deconstruct()
+        self.assertEqual(definition[0], "RemoveConstraint")
+        self.assertEqual(definition[1], [])
+        self.assertEqual(definition[2], {'model_name': "Pony", 'name': "pony_test_constraint"})
+
     def test_alter_model_options(self):
         """
         Tests the AlterModelOptions operation.
diff --git a/tests/migrations/test_state.py b/tests/migrations/test_state.py
index 255e14beff..6a7a087ac5 100644
--- a/tests/migrations/test_state.py
+++ b/tests/migrations/test_state.py
@@ -127,7 +127,12 @@ class StateTests(SimpleTestCase):
         self.assertIs(author_state.fields[3][1].null, True)
         self.assertEqual(
             author_state.options,
-            {"unique_together": {("name", "bio")}, "index_together": {("bio", "age")}, "indexes": []}
+            {
+                "unique_together": {("name", "bio")},
+                "index_together": {("bio", "age")},
+                "indexes": [],
+                "constraints": [],
+            }
         )
         self.assertEqual(author_state.bases, (models.Model,))
 
@@ -139,14 +144,17 @@ class StateTests(SimpleTestCase):
         self.assertEqual(book_state.fields[3][1].__class__.__name__, "ManyToManyField")
         self.assertEqual(
             book_state.options,
-            {"verbose_name": "tome", "db_table": "test_tome", "indexes": [book_index]},
+            {"verbose_name": "tome", "db_table": "test_tome", "indexes": [book_index], "constraints": []},
         )
         self.assertEqual(book_state.bases, (models.Model,))
 
         self.assertEqual(author_proxy_state.app_label, "migrations")
         self.assertEqual(author_proxy_state.name, "AuthorProxy")
         self.assertEqual(author_proxy_state.fields, [])
-        self.assertEqual(author_proxy_state.options, {"proxy": True, "ordering": ["name"], "indexes": []})
+        self.assertEqual(
+            author_proxy_state.options,
+            {"proxy": True, "ordering": ["name"], "indexes": [], "constraints": []},
+        )
         self.assertEqual(author_proxy_state.bases, ("migrations.author",))
 
         self.assertEqual(sub_author_state.app_label, "migrations")
@@ -1002,7 +1010,7 @@ class ModelStateTests(SimpleTestCase):
         self.assertEqual(author_state.fields[1][1].max_length, 255)
         self.assertIs(author_state.fields[2][1].null, False)
         self.assertIs(author_state.fields[3][1].null, True)
-        self.assertEqual(author_state.options, {'swappable': 'TEST_SWAPPABLE_MODEL', 'indexes': []})
+        self.assertEqual(author_state.options, {'swappable': 'TEST_SWAPPABLE_MODEL', 'indexes': [], "constraints": []})
         self.assertEqual(author_state.bases, (models.Model,))
         self.assertEqual(author_state.managers, [])
 
@@ -1047,7 +1055,7 @@ class ModelStateTests(SimpleTestCase):
         self.assertEqual(station_state.fields[2][1].null, False)
         self.assertEqual(
             station_state.options,
-            {'abstract': False, 'swappable': 'TEST_SWAPPABLE_MODEL', 'indexes': []}
+            {'abstract': False, 'swappable': 'TEST_SWAPPABLE_MODEL', 'indexes': [], 'constraints': []}
         )
         self.assertEqual(station_state.bases, ('migrations.searchablelocation',))
         self.assertEqual(station_state.managers, [])
@@ -1129,6 +1137,21 @@ class ModelStateTests(SimpleTestCase):
         index_names = [index.name for index in model_state.options['indexes']]
         self.assertEqual(index_names, ['foo_idx'])
 
+    @isolate_apps('migrations')
+    def test_from_model_constraints(self):
+        class ModelWithConstraints(models.Model):
+            size = models.IntegerField()
+
+            class Meta:
+                constraints = [models.CheckConstraint(models.Q(size__gt=1), 'size_gt_1')]
+
+        state = ModelState.from_model(ModelWithConstraints)
+        model_constraints = ModelWithConstraints._meta.constraints
+        state_constraints = state.options['constraints']
+        self.assertEqual(model_constraints, state_constraints)
+        self.assertIsNot(model_constraints, state_constraints)
+        self.assertIsNot(model_constraints[0], state_constraints[0])
+
 
 class RelatedModelsTests(SimpleTestCase):
 
diff --git a/tests/queries/test_query.py b/tests/queries/test_query.py
new file mode 100644
index 0000000000..10ea8eb0f2
--- /dev/null
+++ b/tests/queries/test_query.py
@@ -0,0 +1,95 @@
+from datetime import datetime
+
+from django.core.exceptions import FieldError
+from django.db.models import CharField, F, Q
+from django.db.models.expressions import SimpleCol
+from django.db.models.fields.related_lookups import RelatedIsNull
+from django.db.models.functions import Lower
+from django.db.models.lookups import Exact, GreaterThan, IsNull, LessThan
+from django.db.models.sql.query import Query
+from django.db.models.sql.where import OR
+from django.test import TestCase
+
+from .models import Author, Item, ObjectC, Ranking
+
+
+class TestQuery(TestCase):
+    def test_simple_query(self):
+        query = Query(Author)
+        where = query.build_where(Q(num__gt=2))
+        lookup = where.children[0]
+        self.assertIsInstance(lookup, GreaterThan)
+        self.assertEqual(lookup.rhs, 2)
+        self.assertEqual(lookup.lhs.target, Author._meta.get_field('num'))
+
+    def test_complex_query(self):
+        query = Query(Author)
+        where = query.build_where(Q(num__gt=2) | Q(num__lt=0))
+        self.assertEqual(where.connector, OR)
+
+        lookup = where.children[0]
+        self.assertIsInstance(lookup, GreaterThan)
+        self.assertEqual(lookup.rhs, 2)
+        self.assertEqual(lookup.lhs.target, Author._meta.get_field('num'))
+
+        lookup = where.children[1]
+        self.assertIsInstance(lookup, LessThan)
+        self.assertEqual(lookup.rhs, 0)
+        self.assertEqual(lookup.lhs.target, Author._meta.get_field('num'))
+
+    def test_multiple_fields(self):
+        query = Query(Item)
+        where = query.build_where(Q(modified__gt=F('created')))
+        lookup = where.children[0]
+        self.assertIsInstance(lookup, GreaterThan)
+        self.assertIsInstance(lookup.rhs, SimpleCol)
+        self.assertIsInstance(lookup.lhs, SimpleCol)
+        self.assertEqual(lookup.rhs.target, Item._meta.get_field('created'))
+        self.assertEqual(lookup.lhs.target, Item._meta.get_field('modified'))
+
+    def test_transform(self):
+        query = Query(Author)
+        CharField.register_lookup(Lower, 'lower')
+        try:
+            where = query.build_where(~Q(name__lower='foo'))
+        finally:
+            CharField._unregister_lookup(Lower, 'lower')
+        lookup = where.children[0]
+        self.assertIsInstance(lookup, Exact)
+        self.assertIsInstance(lookup.lhs, Lower)
+        self.assertIsInstance(lookup.lhs.lhs, SimpleCol)
+        self.assertEqual(lookup.lhs.lhs.target, Author._meta.get_field('name'))
+
+    def test_negated_nullable(self):
+        query = Query(Item)
+        where = query.build_where(~Q(modified__lt=datetime(2017, 1, 1)))
+        self.assertTrue(where.negated)
+        lookup = where.children[0]
+        self.assertIsInstance(lookup, LessThan)
+        self.assertEqual(lookup.lhs.target, Item._meta.get_field('modified'))
+        lookup = where.children[1]
+        self.assertIsInstance(lookup, IsNull)
+        self.assertEqual(lookup.lhs.target, Item._meta.get_field('modified'))
+
+    def test_foreign_key(self):
+        query = Query(Item)
+        msg = 'Joined field references are not permitted in this query'
+        with self.assertRaisesMessage(FieldError, msg):
+            query.build_where(Q(creator__num__gt=2))
+
+    def test_foreign_key_f(self):
+        query = Query(Ranking)
+        with self.assertRaises(FieldError):
+            query.build_where(Q(rank__gt=F('author__num')))
+
+    def test_foreign_key_exclusive(self):
+        query = Query(ObjectC)
+        where = query.build_where(Q(objecta=None) | Q(objectb=None))
+        a_isnull = where.children[0]
+        self.assertIsInstance(a_isnull, RelatedIsNull)
+        self.assertIsInstance(a_isnull.lhs, SimpleCol)
+        self.assertEqual(a_isnull.lhs.target, ObjectC._meta.get_field('objecta'))
+        b_isnull = where.children[1]
+        self.assertIsInstance(b_isnull, RelatedIsNull)
+        self.assertIsInstance(b_isnull.lhs, SimpleCol)
+        self.assertEqual(b_isnull.lhs.target, ObjectC._meta.get_field('objectb'))