Fixed #28668 -- Allowed QuerySet.bulk_create() to ignore insert conflicts.

1 files changed, 35 insertions, 1 deletions

diff --git a/tests/bulk_create/tests.py b/tests/bulk_create/tests.py
index 2b3e65594f..eb7d2a9e28 100644
--- a/tests/bulk_create/tests.py
+++ b/tests/bulk_create/tests.py
@@ -1,6 +1,6 @@
 from operator import attrgetter
 
-from django.db import connection
+from django.db import IntegrityError, NotSupportedError, connection
 from django.db.models import FileField, Value
 from django.db.models.functions import Lower
 from django.test import (
@@ -261,3 +261,37 @@ class BulkCreateTests(TestCase):
         # Objects save via bulk_create() and save() should have equal state.
         self.assertEqual(state_ca._state.adding, state_ny._state.adding)
         self.assertEqual(state_ca._state.db, state_ny._state.db)
+
+    @skipIfDBFeature('supports_ignore_conflicts')
+    def test_ignore_conflicts_value_error(self):
+        message = 'This database backend does not support ignoring conflicts.'
+        with self.assertRaisesMessage(NotSupportedError, message):
+            TwoFields.objects.bulk_create(self.data, ignore_conflicts=True)
+
+    @skipUnlessDBFeature('supports_ignore_conflicts')
+    def test_ignore_conflicts_ignore(self):
+        data = [
+            TwoFields(f1=1, f2=1),
+            TwoFields(f1=2, f2=2),
+            TwoFields(f1=3, f2=3),
+        ]
+        TwoFields.objects.bulk_create(data)
+        self.assertEqual(TwoFields.objects.count(), 3)
+        # With ignore_conflicts=True, conflicts are ignored.
+        conflicting_objects = [
+            TwoFields(f1=2, f2=2),
+            TwoFields(f1=3, f2=3),
+        ]
+        TwoFields.objects.bulk_create([conflicting_objects[0]], ignore_conflicts=True)
+        TwoFields.objects.bulk_create(conflicting_objects, ignore_conflicts=True)
+        self.assertEqual(TwoFields.objects.count(), 3)
+        self.assertIsNone(conflicting_objects[0].pk)
+        self.assertIsNone(conflicting_objects[1].pk)
+        # New objects are created and conflicts are ignored.
+        new_object = TwoFields(f1=4, f2=4)
+        TwoFields.objects.bulk_create(conflicting_objects + [new_object], ignore_conflicts=True)
+        self.assertEqual(TwoFields.objects.count(), 4)
+        self.assertIsNone(new_object.pk)
+        # Without ignore_conflicts=True, there's a problem.
+        with self.assertRaises(IntegrityError):
+            TwoFields.objects.bulk_create(conflicting_objects)