Fixed #470 -- Added support for database defaults on fields.

Special thanks to Hannes Ljungberg for finding multiple implementation
gaps.

Thanks also to Simon Charette, Adam Johnson, and Mariusz Felisiak for
reviews.

2 files changed, 234 insertions, 2 deletions

diff --git a/tests/field_defaults/models.py b/tests/field_defaults/models.py
index b95005192a..5f9c38a5a4 100644
--- a/tests/field_defaults/models.py
+++ b/tests/field_defaults/models.py
@@ -12,6 +12,8 @@ field.
 from datetime import datetime
 
 from django.db import models
+from django.db.models.functions import Coalesce, ExtractYear, Now, Pi
+from django.db.models.lookups import GreaterThan
 
 
 class Article(models.Model):
@@ -20,3 +22,45 @@ class Article(models.Model):
 
     def __str__(self):
         return self.headline
+
+
+class DBArticle(models.Model):
+    """
+    Values or expressions can be passed as the db_default parameter to a field.
+    When the object is created without an explicit value passed in, the
+    database will insert the default value automatically.
+    """
+
+    headline = models.CharField(max_length=100, db_default="Default headline")
+    pub_date = models.DateTimeField(db_default=Now())
+
+    class Meta:
+        required_db_features = {"supports_expression_defaults"}
+
+
+class DBDefaults(models.Model):
+    both = models.IntegerField(default=1, db_default=2)
+    null = models.FloatField(null=True, db_default=1.1)
+
+
+class DBDefaultsFunction(models.Model):
+    number = models.FloatField(db_default=Pi())
+    year = models.IntegerField(db_default=ExtractYear(Now()))
+    added = models.FloatField(db_default=Pi() + 4.5)
+    multiple_subfunctions = models.FloatField(db_default=Coalesce(4.5, Pi()))
+    case_when = models.IntegerField(
+        db_default=models.Case(models.When(GreaterThan(2, 1), then=3), default=4)
+    )
+
+    class Meta:
+        required_db_features = {"supports_expression_defaults"}
+
+
+class DBDefaultsPK(models.Model):
+    language_code = models.CharField(primary_key=True, max_length=2, db_default="en")
+
+
+class DBDefaultsFK(models.Model):
+    language_code = models.ForeignKey(
+        DBDefaultsPK, db_default="fr", on_delete=models.CASCADE
+    )
diff --git a/tests/field_defaults/tests.py b/tests/field_defaults/tests.py
index 19b05aa537..76d01f7a5a 100644
--- a/tests/field_defaults/tests.py
+++ b/tests/field_defaults/tests.py
@@ -1,8 +1,28 @@
 from datetime import datetime
+from math import pi
 
-from django.test import TestCase
+from django.db import connection
+from django.db.models import Case, F, FloatField, Value, When
+from django.db.models.expressions import (
+    Expression,
+    ExpressionList,
+    ExpressionWrapper,
+    Func,
+    OrderByList,
+    RawSQL,
+)
+from django.db.models.functions import Collate
+from django.db.models.lookups import GreaterThan
+from django.test import SimpleTestCase, TestCase, skipIfDBFeature, skipUnlessDBFeature
 
-from .models import Article
+from .models import (
+    Article,
+    DBArticle,
+    DBDefaults,
+    DBDefaultsFK,
+    DBDefaultsFunction,
+    DBDefaultsPK,
+)
 
 
 class DefaultTests(TestCase):
@@ -14,3 +34,171 @@ class DefaultTests(TestCase):
         self.assertIsInstance(a.id, int)
         self.assertEqual(a.headline, "Default headline")
         self.assertLess((now - a.pub_date).seconds, 5)
+
+    @skipUnlessDBFeature(
+        "can_return_columns_from_insert", "supports_expression_defaults"
+    )
+    def test_field_db_defaults_returning(self):
+        a = DBArticle()
+        a.save()
+        self.assertIsInstance(a.id, int)
+        self.assertEqual(a.headline, "Default headline")
+        self.assertIsInstance(a.pub_date, datetime)
+
+    @skipIfDBFeature("can_return_columns_from_insert")
+    @skipUnlessDBFeature("supports_expression_defaults")
+    def test_field_db_defaults_refresh(self):
+        a = DBArticle()
+        a.save()
+        a.refresh_from_db()
+        self.assertIsInstance(a.id, int)
+        self.assertEqual(a.headline, "Default headline")
+        self.assertIsInstance(a.pub_date, datetime)
+
+    def test_null_db_default(self):
+        obj1 = DBDefaults.objects.create()
+        if not connection.features.can_return_columns_from_insert:
+            obj1.refresh_from_db()
+        self.assertEqual(obj1.null, 1.1)
+
+        obj2 = DBDefaults.objects.create(null=None)
+        self.assertIsNone(obj2.null)
+
+    @skipUnlessDBFeature("supports_expression_defaults")
+    def test_db_default_function(self):
+        m = DBDefaultsFunction.objects.create()
+        if not connection.features.can_return_columns_from_insert:
+            m.refresh_from_db()
+        self.assertAlmostEqual(m.number, pi)
+        self.assertEqual(m.year, datetime.now().year)
+        self.assertAlmostEqual(m.added, pi + 4.5)
+        self.assertEqual(m.multiple_subfunctions, 4.5)
+
+    @skipUnlessDBFeature("insert_test_table_with_defaults")
+    def test_both_default(self):
+        create_sql = connection.features.insert_test_table_with_defaults
+        with connection.cursor() as cursor:
+            cursor.execute(create_sql.format(DBDefaults._meta.db_table))
+        obj1 = DBDefaults.objects.get()
+        self.assertEqual(obj1.both, 2)
+
+        obj2 = DBDefaults.objects.create()
+        self.assertEqual(obj2.both, 1)
+
+    def test_pk_db_default(self):
+        obj1 = DBDefaultsPK.objects.create()
+        if not connection.features.can_return_columns_from_insert:
+            # refresh_from_db() cannot be used because that needs the pk to
+            # already be known to Django.
+            obj1 = DBDefaultsPK.objects.get(pk="en")
+        self.assertEqual(obj1.pk, "en")
+        self.assertEqual(obj1.language_code, "en")
+
+        obj2 = DBDefaultsPK.objects.create(language_code="de")
+        self.assertEqual(obj2.pk, "de")
+        self.assertEqual(obj2.language_code, "de")
+
+    def test_foreign_key_db_default(self):
+        parent1 = DBDefaultsPK.objects.create(language_code="fr")
+        child1 = DBDefaultsFK.objects.create()
+        if not connection.features.can_return_columns_from_insert:
+            child1.refresh_from_db()
+        self.assertEqual(child1.language_code, parent1)
+
+        parent2 = DBDefaultsPK.objects.create()
+        if not connection.features.can_return_columns_from_insert:
+            # refresh_from_db() cannot be used because that needs the pk to
+            # already be known to Django.
+            parent2 = DBDefaultsPK.objects.get(pk="en")
+        child2 = DBDefaultsFK.objects.create(language_code=parent2)
+        self.assertEqual(child2.language_code, parent2)
+
+    @skipUnlessDBFeature(
+        "can_return_columns_from_insert", "supports_expression_defaults"
+    )
+    def test_case_when_db_default_returning(self):
+        m = DBDefaultsFunction.objects.create()
+        self.assertEqual(m.case_when, 3)
+
+    @skipIfDBFeature("can_return_columns_from_insert")
+    @skipUnlessDBFeature("supports_expression_defaults")
+    def test_case_when_db_default_no_returning(self):
+        m = DBDefaultsFunction.objects.create()
+        m.refresh_from_db()
+        self.assertEqual(m.case_when, 3)
+
+    @skipUnlessDBFeature("supports_expression_defaults")
+    def test_bulk_create_all_db_defaults(self):
+        articles = [DBArticle(), DBArticle()]
+        DBArticle.objects.bulk_create(articles)
+
+        headlines = DBArticle.objects.values_list("headline", flat=True)
+        self.assertSequenceEqual(headlines, ["Default headline", "Default headline"])
+
+    @skipUnlessDBFeature("supports_expression_defaults")
+    def test_bulk_create_all_db_defaults_one_field(self):
+        pub_date = datetime.now()
+        articles = [DBArticle(pub_date=pub_date), DBArticle(pub_date=pub_date)]
+        DBArticle.objects.bulk_create(articles)
+
+        headlines = DBArticle.objects.values_list("headline", "pub_date")
+        self.assertSequenceEqual(
+            headlines,
+            [
+                ("Default headline", pub_date),
+                ("Default headline", pub_date),
+            ],
+        )
+
+    @skipUnlessDBFeature("supports_expression_defaults")
+    def test_bulk_create_mixed_db_defaults(self):
+        articles = [DBArticle(), DBArticle(headline="Something else")]
+        DBArticle.objects.bulk_create(articles)
+
+        headlines = DBArticle.objects.values_list("headline", flat=True)
+        self.assertCountEqual(headlines, ["Default headline", "Something else"])
+
+    @skipUnlessDBFeature("supports_expression_defaults")
+    def test_bulk_create_mixed_db_defaults_function(self):
+        instances = [DBDefaultsFunction(), DBDefaultsFunction(year=2000)]
+        DBDefaultsFunction.objects.bulk_create(instances)
+
+        years = DBDefaultsFunction.objects.values_list("year", flat=True)
+        self.assertCountEqual(years, [2000, datetime.now().year])
+
+
+class AllowedDefaultTests(SimpleTestCase):
+    def test_allowed(self):
+        class Max(Func):
+            function = "MAX"
+
+        tests = [
+            Value(10),
+            Max(1, 2),
+            RawSQL("Now()", ()),
+            Value(10) + Value(7),  # Combined expression.
+            ExpressionList(Value(1), Value(2)),
+            ExpressionWrapper(Value(1), output_field=FloatField()),
+            Case(When(GreaterThan(2, 1), then=3), default=4),
+        ]
+        for expression in tests:
+            with self.subTest(expression=expression):
+                self.assertIs(expression.allowed_default, True)
+
+    def test_disallowed(self):
+        class Max(Func):
+            function = "MAX"
+
+        tests = [
+            Expression(),
+            F("field"),
+            Max(F("count"), 1),
+            Value(10) + F("count"),  # Combined expression.
+            ExpressionList(F("count"), Value(2)),
+            ExpressionWrapper(F("count"), output_field=FloatField()),
+            Collate(Value("John"), "nocase"),
+            OrderByList("field"),
+        ]
+        for expression in tests:
+            with self.subTest(expression=expression):
+                self.assertIs(expression.allowed_default, False)