diff options
Diffstat (limited to 'tests/regressiontests/bulk_create/tests.py')
| -rw-r--r-- | tests/regressiontests/bulk_create/tests.py | 48 |
1 files changed, 44 insertions, 4 deletions
diff --git a/tests/regressiontests/bulk_create/tests.py b/tests/regressiontests/bulk_create/tests.py index 0fa142b795..b4c3e7f17f 100644 --- a/tests/regressiontests/bulk_create/tests.py +++ b/tests/regressiontests/bulk_create/tests.py @@ -2,9 +2,11 @@ from __future__ import with_statement, absolute_import from operator import attrgetter -from django.test import TestCase, skipUnlessDBFeature +from django.db import connection +from django.test import TestCase, skipIfDBFeature +from django.test.utils import override_settings -from .models import Country, Restaurant, Pizzeria, State +from .models import Country, Restaurant, Pizzeria, State, TwoFields class BulkCreateTests(TestCase): @@ -27,7 +29,6 @@ class BulkCreateTests(TestCase): self.assertEqual(created, []) self.assertEqual(Country.objects.count(), 4) - @skipUnlessDBFeature("has_bulk_insert") def test_efficiency(self): with self.assertNumQueries(1): Country.objects.bulk_create(self.data) @@ -56,4 +57,43 @@ class BulkCreateTests(TestCase): ]) self.assertQuerysetEqual(State.objects.order_by("two_letter_code"), [ "CA", "IL", "ME", "NY", - ], attrgetter("two_letter_code"))
\ No newline at end of file + ], attrgetter("two_letter_code")) + + def test_large_batch(self): + with override_settings(DEBUG=True): + connection.queries = [] + TwoFields.objects.bulk_create([ + TwoFields(f1=i, f2=i+1) for i in range(0, 1001) + ]) + self.assertTrue(len(connection.queries) < 10) + self.assertEqual(TwoFields.objects.count(), 1001) + self.assertEqual( + TwoFields.objects.filter(f1__gte=450, f1__lte=550).count(), + 101) + self.assertEqual(TwoFields.objects.filter(f2__gte=901).count(), 101) + + def test_large_batch_mixed(self): + """ + Test inserting a large batch with objects having primary key set + mixed together with objects without PK set. + """ + with override_settings(DEBUG=True): + connection.queries = [] + TwoFields.objects.bulk_create([ + TwoFields(id=i if i % 2 == 0 else None, f1=i, f2=i+1) + for i in range(100000, 101000)]) + self.assertTrue(len(connection.queries) < 10) + self.assertEqual(TwoFields.objects.count(), 1000) + # We can't assume much about the ID's created, except that the above + # created IDs must exist. + id_range = range(100000, 101000, 2) + self.assertEqual(TwoFields.objects.filter(id__in=id_range).count(), 500) + self.assertEqual(TwoFields.objects.exclude(id__in=id_range).count(), 500) + + def test_explicit_batch_size(self): + objs = [TwoFields(f1=i, f2=i) for i in range(0, 100)] + with self.assertNumQueries(2): + TwoFields.objects.bulk_create(objs, 50) + TwoFields.objects.all().delete() + with self.assertNumQueries(1): + TwoFields.objects.bulk_create(objs, len(objs)) |