Ticket #30827: ticket30827RegressionTests.diff

tests/bulk_create/tests.py

@@ +1 @@
+from math import ceil
 from operator import attrgetter
 
-from django.db import IntegrityError, NotSupportedError, connection
+from django.db import IntegrityError, NotSupportedError, OperationalError, connection, connections
 from django.db.models import FileField, Value
+from django.db.models.fields import AutoField
 from django.db.models.functions import Lower
 from django.test import (
     TestCase, override_settings, skipIfDBFeature, skipUnlessDBFeature,
@@ -295 +297 @@
         # Without ignore_conflicts=True, there's a problem.
         with self.assertRaises(IntegrityError):
             TwoFields.objects.bulk_create(conflicting_objects)
+
+    # Regression tests for #30827
+    def test_batched_insert_on_model_without_pk(self):
+        objs = [Country(name=str(i), iso_two_letter=str(i), description=str(i))
+                for i in range(1000)]
+        ops = connections[Country.objects.db].ops
+        opts = Country.objects.model._meta
+        fields = [f for f in opts.concrete_fields if not isinstance(f, AutoField)]
+        max_batch_size = max(ops.bulk_batch_size(fields, objs), 1)
+        with self.assertNumQueries(ceil(1000/max_batch_size)):
+            Country.objects.bulk_create(objs, batch_size = max_batch_size + 1)
+
+    # Regression tests for #30827
+    def test_batched_insert_upper_limit(self):
+        ops = connections[State.objects.db].ops
+        opts = State.objects.model._meta
+        fields = opts.concrete_fields
+        objs = [State(two_letter_code=str(i))
+                for i in range(1000)]
+        max_batch_size = max(ops.bulk_batch_size(fields, objs), 1)
+        try:
+            State.objects.bulk_create(objs, batch_size = max_batch_size + 1)
+        except OperationalError as e:
+            self.fail("Unexpected failure while performing regresion test in "
+                      "bulk_create: %s" % e)