Fixed #18927 -- Fixed bulk_create tests when no has_bulk_insert

This commit is contained in:
Michael Manfre 2012-09-28 13:25:09 -04:00 committed by Anssi Kääriäinen
parent b625e8272b
commit 32ac067a6d
1 changed files with 43 additions and 3 deletions

View File

@ -3,7 +3,7 @@ from __future__ import absolute_import
from operator import attrgetter from operator import attrgetter
from django.db import connection from django.db import connection
from django.test import TestCase, skipIfDBFeature from django.test import TestCase, skipIfDBFeature, skipUnlessDBFeature
from django.test.utils import override_settings from django.test.utils import override_settings
from .models import Country, Restaurant, Pizzeria, State, TwoFields from .models import Country, Restaurant, Pizzeria, State, TwoFields
@ -29,6 +29,7 @@ class BulkCreateTests(TestCase):
self.assertEqual(created, []) self.assertEqual(created, [])
self.assertEqual(Country.objects.count(), 4) self.assertEqual(Country.objects.count(), 4)
@skipUnlessDBFeature('has_bulk_insert')
def test_efficiency(self): def test_efficiency(self):
with self.assertNumQueries(1): with self.assertNumQueries(1):
Country.objects.bulk_create(self.data) Country.objects.bulk_create(self.data)
@ -50,6 +51,16 @@ class BulkCreateTests(TestCase):
], attrgetter("name")) ], attrgetter("name"))
def test_non_auto_increment_pk(self): def test_non_auto_increment_pk(self):
State.objects.bulk_create([
State(two_letter_code=s)
for s in ["IL", "NY", "CA", "ME"]
])
self.assertQuerysetEqual(State.objects.order_by("two_letter_code"), [
"CA", "IL", "ME", "NY",
], attrgetter("two_letter_code"))
@skipUnlessDBFeature('has_bulk_insert')
def test_non_auto_increment_pk_efficiency(self):
with self.assertNumQueries(1): with self.assertNumQueries(1):
State.objects.bulk_create([ State.objects.bulk_create([
State(two_letter_code=s) State(two_letter_code=s)
@ -77,13 +88,21 @@ class BulkCreateTests(TestCase):
TwoFields.objects.bulk_create([ TwoFields.objects.bulk_create([
TwoFields(f1=i, f2=i+1) for i in range(0, 1001) 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.count(), 1001)
self.assertEqual( self.assertEqual(
TwoFields.objects.filter(f1__gte=450, f1__lte=550).count(), TwoFields.objects.filter(f1__gte=450, f1__lte=550).count(),
101) 101)
self.assertEqual(TwoFields.objects.filter(f2__gte=901).count(), 101) self.assertEqual(TwoFields.objects.filter(f2__gte=901).count(), 101)
@skipUnlessDBFeature('has_bulk_insert')
def test_large_batch_efficiency(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)
def test_large_batch_mixed(self): def test_large_batch_mixed(self):
""" """
Test inserting a large batch with objects having primary key set Test inserting a large batch with objects having primary key set
@ -94,7 +113,6 @@ class BulkCreateTests(TestCase):
TwoFields.objects.bulk_create([ TwoFields.objects.bulk_create([
TwoFields(id=i if i % 2 == 0 else None, f1=i, f2=i+1) TwoFields(id=i if i % 2 == 0 else None, f1=i, f2=i+1)
for i in range(100000, 101000)]) for i in range(100000, 101000)])
self.assertTrue(len(connection.queries) < 10)
self.assertEqual(TwoFields.objects.count(), 1000) self.assertEqual(TwoFields.objects.count(), 1000)
# We can't assume much about the ID's created, except that the above # We can't assume much about the ID's created, except that the above
# created IDs must exist. # created IDs must exist.
@ -102,7 +120,29 @@ class BulkCreateTests(TestCase):
self.assertEqual(TwoFields.objects.filter(id__in=id_range).count(), 500) self.assertEqual(TwoFields.objects.filter(id__in=id_range).count(), 500)
self.assertEqual(TwoFields.objects.exclude(id__in=id_range).count(), 500) self.assertEqual(TwoFields.objects.exclude(id__in=id_range).count(), 500)
@skipUnlessDBFeature('has_bulk_insert')
def test_large_batch_mixed_efficiency(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)
def test_explicit_batch_size(self): def test_explicit_batch_size(self):
objs = [TwoFields(f1=i, f2=i) for i in range(0, 4)]
TwoFields.objects.bulk_create(objs, 2)
self.assertEqual(TwoFields.objects.count(), len(objs))
TwoFields.objects.all().delete()
TwoFields.objects.bulk_create(objs, len(objs))
self.assertEqual(TwoFields.objects.count(), len(objs))
@skipUnlessDBFeature('has_bulk_insert')
def test_explicit_batch_size_efficiency(self):
objs = [TwoFields(f1=i, f2=i) for i in range(0, 100)] objs = [TwoFields(f1=i, f2=i) for i in range(0, 100)]
with self.assertNumQueries(2): with self.assertNumQueries(2):
TwoFields.objects.bulk_create(objs, 50) TwoFields.objects.bulk_create(objs, 50)