Skip to content
This repository

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Browse code

Fixed #18927 -- Fixed bulk_create tests when no has_bulk_insert

  • Loading branch information...
commit 32ac067a6ddca465cd94fc7ee35bfe54a675979b 1 parent b625e82
Michael Manfre authored September 28, 2012 akaariai committed October 10, 2012
46  tests/regressiontests/bulk_create/tests.py
@@ -3,7 +3,7 @@
3 3
 from operator import attrgetter
4 4
 
5 5
 from django.db import connection
6  
-from django.test import TestCase, skipIfDBFeature
  6
+from django.test import TestCase, skipIfDBFeature, skipUnlessDBFeature
7 7
 from django.test.utils import override_settings
8 8
 
9 9
 from .models import Country, Restaurant, Pizzeria, State, TwoFields
@@ -29,6 +29,7 @@ def test_simple(self):
29 29
         self.assertEqual(created, [])
30 30
         self.assertEqual(Country.objects.count(), 4)
31 31
 
  32
+    @skipUnlessDBFeature('has_bulk_insert')
32 33
     def test_efficiency(self):
33 34
         with self.assertNumQueries(1):
34 35
             Country.objects.bulk_create(self.data)
@@ -50,6 +51,16 @@ def test_inheritance(self):
50 51
         ], attrgetter("name"))
51 52
 
52 53
     def test_non_auto_increment_pk(self):
  54
+        State.objects.bulk_create([
  55
+            State(two_letter_code=s)
  56
+            for s in ["IL", "NY", "CA", "ME"]
  57
+        ])
  58
+        self.assertQuerysetEqual(State.objects.order_by("two_letter_code"), [
  59
+            "CA", "IL", "ME", "NY",
  60
+        ], attrgetter("two_letter_code"))
  61
+
  62
+    @skipUnlessDBFeature('has_bulk_insert')
  63
+    def test_non_auto_increment_pk_efficiency(self):
53 64
         with self.assertNumQueries(1):
54 65
             State.objects.bulk_create([
55 66
                 State(two_letter_code=s)
@@ -77,13 +88,21 @@ def test_large_batch(self):
77 88
             TwoFields.objects.bulk_create([
78 89
                    TwoFields(f1=i, f2=i+1) for i in range(0, 1001)
79 90
                 ])
80  
-            self.assertTrue(len(connection.queries) < 10)
81 91
         self.assertEqual(TwoFields.objects.count(), 1001)
82 92
         self.assertEqual(
83 93
             TwoFields.objects.filter(f1__gte=450, f1__lte=550).count(),
84 94
             101)
85 95
         self.assertEqual(TwoFields.objects.filter(f2__gte=901).count(), 101)
86 96
 
  97
+    @skipUnlessDBFeature('has_bulk_insert')
  98
+    def test_large_batch_efficiency(self):
  99
+        with override_settings(DEBUG=True):
  100
+            connection.queries = []
  101
+            TwoFields.objects.bulk_create([
  102
+                   TwoFields(f1=i, f2=i+1) for i in range(0, 1001)
  103
+                ])
  104
+            self.assertTrue(len(connection.queries) < 10)
  105
+
87 106
     def test_large_batch_mixed(self):
88 107
         """
89 108
         Test inserting a large batch with objects having primary key set
@@ -94,7 +113,6 @@ def test_large_batch_mixed(self):
94 113
             TwoFields.objects.bulk_create([
95 114
                 TwoFields(id=i if i % 2 == 0 else None, f1=i, f2=i+1)
96 115
                 for i in range(100000, 101000)])
97  
-            self.assertTrue(len(connection.queries) < 10)
98 116
         self.assertEqual(TwoFields.objects.count(), 1000)
99 117
         # We can't assume much about the ID's created, except that the above
100 118
         # created IDs must exist.
@@ -102,7 +120,29 @@ def test_large_batch_mixed(self):
102 120
         self.assertEqual(TwoFields.objects.filter(id__in=id_range).count(), 500)
103 121
         self.assertEqual(TwoFields.objects.exclude(id__in=id_range).count(), 500)
104 122
 
  123
+    @skipUnlessDBFeature('has_bulk_insert')
  124
+    def test_large_batch_mixed_efficiency(self):
  125
+        """
  126
+        Test inserting a large batch with objects having primary key set
  127
+        mixed together with objects without PK set.
  128
+        """
  129
+        with override_settings(DEBUG=True):
  130
+            connection.queries = []
  131
+            TwoFields.objects.bulk_create([
  132
+                TwoFields(id=i if i % 2 == 0 else None, f1=i, f2=i+1)
  133
+                for i in range(100000, 101000)])
  134
+            self.assertTrue(len(connection.queries) < 10)
  135
+
105 136
     def test_explicit_batch_size(self):
  137
+        objs = [TwoFields(f1=i, f2=i) for i in range(0, 4)]
  138
+        TwoFields.objects.bulk_create(objs, 2)
  139
+        self.assertEqual(TwoFields.objects.count(), len(objs))
  140
+        TwoFields.objects.all().delete()
  141
+        TwoFields.objects.bulk_create(objs, len(objs))
  142
+        self.assertEqual(TwoFields.objects.count(), len(objs))
  143
+
  144
+    @skipUnlessDBFeature('has_bulk_insert')
  145
+    def test_explicit_batch_size_efficiency(self):
106 146
         objs = [TwoFields(f1=i, f2=i) for i in range(0, 100)]
107 147
         with self.assertNumQueries(2):
108 148
             TwoFields.objects.bulk_create(objs, 50)

0 notes on commit 32ac067

Please sign in to comment.
Something went wrong with that request. Please try again.