summaryrefslogtreecommitdiff
path: root/tests/bulk_create/tests.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/bulk_create/tests.py')
-rw-r--r--tests/bulk_create/tests.py36
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)