Skip to content

Commit

Permalink
Fixed #18927 - Fix bulk_create tests when not has_bulk_insert
Browse files Browse the repository at this point in the history
  • Loading branch information
manfre committed Sep 28, 2012
1 parent 31e59bb commit e575f5e
Showing 1 changed file with 43 additions and 3 deletions.
46 changes: 43 additions & 3 deletions tests/regressiontests/bulk_create/tests.py
Original file line number Original file line Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
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
Expand All @@ -29,6 +29,7 @@ def test_simple(self):
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)
Expand All @@ -50,6 +51,16 @@ def test_inheritance(self):
], 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)
Expand Down Expand Up @@ -77,13 +88,21 @@ def test_large_batch(self):
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
Expand All @@ -94,15 +113,36 @@ def test_large_batch_mixed(self):
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.
id_range = range(100000, 101000, 2) id_range = range(100000, 101000, 2)
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)
Expand Down

0 comments on commit e575f5e

Please sign in to comment.