Refs #36822 -- Hoisted bulk_batch_size() implementations to base backend.

4 files changed, 59 insertions, 60 deletions

diff --git a/tests/backends/base/test_operations.py b/tests/backends/base/test_operations.py
index 96fadb4c7a..8be1d3e841 100644
--- a/tests/backends/base/test_operations.py
+++ b/tests/backends/base/test_operations.py
@@ -1,7 +1,7 @@
 import decimal
 
 from django.core.management.color import no_style
-from django.db import NotSupportedError, connection, transaction
+from django.db import NotSupportedError, connection, models, transaction
 from django.db.backends.base.operations import BaseDatabaseOperations
 from django.db.models import DurationField
 from django.db.models.expressions import Col
@@ -11,10 +11,11 @@ from django.test import (
     TransactionTestCase,
     override_settings,
     skipIfDBFeature,
+    skipUnlessDBFeature,
 )
 from django.utils import timezone
 
-from ..models import Author, Book
+from ..models import Author, Book, Person
 
 
 class SimpleDatabaseOperationTests(SimpleTestCase):
@@ -201,6 +202,55 @@ class DatabaseOperationTests(TestCase):
         with self.assertRaisesMessage(NotSupportedError, msg):
             self.ops.subtract_temporals(duration_field_internal_type, None, None)
 
+    @skipUnlessDBFeature("max_query_params")
+    def test_bulk_batch_size_limited(self):
+        max_query_params = connection.features.max_query_params
+        objects = range(max_query_params + 1)
+        first_name_field = Person._meta.get_field("first_name")
+        last_name_field = Person._meta.get_field("last_name")
+        composite_pk = models.CompositePrimaryKey("first_name", "last_name")
+        composite_pk.fields = [first_name_field, last_name_field]
+
+        self.assertEqual(connection.ops.bulk_batch_size([], objects), len(objects))
+        self.assertEqual(
+            connection.ops.bulk_batch_size([first_name_field], objects),
+            max_query_params,
+        )
+        self.assertEqual(
+            connection.ops.bulk_batch_size(
+                [first_name_field, last_name_field], objects
+            ),
+            max_query_params // 2,
+        )
+        self.assertEqual(
+            connection.ops.bulk_batch_size([composite_pk, first_name_field], objects),
+            max_query_params // 3,
+        )
+
+    @skipIfDBFeature("max_query_params")
+    def test_bulk_batch_size_unlimited(self):
+        objects = range(2**16 + 1)
+        first_name_field = Person._meta.get_field("first_name")
+        last_name_field = Person._meta.get_field("last_name")
+        composite_pk = models.CompositePrimaryKey("first_name", "last_name")
+        composite_pk.fields = [first_name_field, last_name_field]
+
+        self.assertEqual(connection.ops.bulk_batch_size([], objects), len(objects))
+        self.assertEqual(
+            connection.ops.bulk_batch_size([first_name_field], objects),
+            len(objects),
+        )
+        self.assertEqual(
+            connection.ops.bulk_batch_size(
+                [first_name_field, last_name_field], objects
+            ),
+            len(objects),
+        )
+        self.assertEqual(
+            connection.ops.bulk_batch_size([composite_pk, first_name_field], objects),
+            len(objects),
+        )
+
 
 class SqlFlushTests(TransactionTestCase):
     available_apps = ["backends"]
diff --git a/tests/backends/oracle/test_operations.py b/tests/backends/oracle/test_operations.py
index 1f9447bde7..197ac5a422 100644
--- a/tests/backends/oracle/test_operations.py
+++ b/tests/backends/oracle/test_operations.py
@@ -1,7 +1,7 @@
 import unittest
 
 from django.core.management.color import no_style
-from django.db import connection, models
+from django.db import connection
 from django.test import TransactionTestCase
 
 from ..models import Person, Tag
@@ -17,31 +17,6 @@ class OperationsTests(TransactionTestCase):
         )
         self.assertEqual(seq_name, "SCHEMA_AUTHORWITHEVENLOB0B8_SQ")
 
-    def test_bulk_batch_size(self):
-        # Oracle restricts the number of parameters in a query.
-        objects = range(2**16)
-        self.assertEqual(connection.ops.bulk_batch_size([], objects), len(objects))
-        # Each field is a parameter for each object.
-        first_name_field = Person._meta.get_field("first_name")
-        last_name_field = Person._meta.get_field("last_name")
-        self.assertEqual(
-            connection.ops.bulk_batch_size([first_name_field], objects),
-            connection.features.max_query_params,
-        )
-        self.assertEqual(
-            connection.ops.bulk_batch_size(
-                [first_name_field, last_name_field],
-                objects,
-            ),
-            connection.features.max_query_params // 2,
-        )
-        composite_pk = models.CompositePrimaryKey("first_name", "last_name")
-        composite_pk.fields = [first_name_field, last_name_field]
-        self.assertEqual(
-            connection.ops.bulk_batch_size([composite_pk, first_name_field], objects),
-            connection.features.max_query_params // 3,
-        )
-
     def test_sql_flush(self):
         statements = connection.ops.sql_flush(
             no_style(),
diff --git a/tests/backends/sqlite/test_operations.py b/tests/backends/sqlite/test_operations.py
index 0c2772301b..dee55b6390 100644
--- a/tests/backends/sqlite/test_operations.py
+++ b/tests/backends/sqlite/test_operations.py
@@ -2,7 +2,7 @@ import sqlite3
 import unittest
 
 from django.core.management.color import no_style
-from django.db import connection, models
+from django.db import connection
 from django.test import TestCase
 
 from ..models import Person, Tag
@@ -88,29 +88,6 @@ class SQLiteOperationsTests(TestCase):
             statements[-1],
         )
 
-    def test_bulk_batch_size(self):
-        self.assertEqual(connection.ops.bulk_batch_size([], [Person()]), 1)
-        first_name_field = Person._meta.get_field("first_name")
-        last_name_field = Person._meta.get_field("last_name")
-        self.assertEqual(
-            connection.ops.bulk_batch_size([first_name_field], [Person()]),
-            connection.features.max_query_params,
-        )
-        self.assertEqual(
-            connection.ops.bulk_batch_size(
-                [first_name_field, last_name_field], [Person()]
-            ),
-            connection.features.max_query_params // 2,
-        )
-        composite_pk = models.CompositePrimaryKey("first_name", "last_name")
-        composite_pk.fields = [first_name_field, last_name_field]
-        self.assertEqual(
-            connection.ops.bulk_batch_size(
-                [composite_pk, first_name_field], [Person()]
-            ),
-            connection.features.max_query_params // 3,
-        )
-
     def test_bulk_batch_size_respects_variable_limit(self):
         first_name_field = Person._meta.get_field("first_name")
         last_name_field = Person._meta.get_field("last_name")
diff --git a/tests/composite_pk/tests.py b/tests/composite_pk/tests.py
index 3001847455..3653beceed 100644
--- a/tests/composite_pk/tests.py
+++ b/tests/composite_pk/tests.py
@@ -149,21 +149,18 @@ class CompositePKTests(TestCase):
 
     def test_in_bulk_batching(self):
         Comment.objects.all().delete()
-        batching_required = connection.features.max_query_params is not None
-        expected_queries = 2 if batching_required else 1
+        num_objects = 10
+        connection.features.__dict__.pop("max_query_params", None)
         with unittest.mock.patch.object(
-            type(connection.features), "max_query_params", 10
+            type(connection.features), "max_query_params", num_objects
         ):
-            num_requiring_batching = (
-                connection.ops.bulk_batch_size([Comment._meta.pk], []) + 1
-            )
             comments = [
                 Comment(id=i, tenant=self.tenant, user=self.user)
-                for i in range(1, num_requiring_batching + 1)
+                for i in range(1, num_objects + 1)
             ]
             Comment.objects.bulk_create(comments)
             id_list = list(Comment.objects.values_list("pk", flat=True))
-            with self.assertNumQueries(expected_queries):
+            with self.assertNumQueries(2):
                 comment_dict = Comment.objects.in_bulk(id_list=id_list)
         self.assertQuerySetEqual(comment_dict, id_list)