Repository: django/django
File: tests/bulk_create/tests.py (Python)

Code before:
        with self.assertNumQueries(1) as ctx:
            DbDefaultModel.objects.bulk_create(
                [
                    DbDefaultModel(name="foo", created_at=timezone.now()),
                    DbDefaultModel(name="bar"),
                ]
            )
        self.assertEqual(
            ctx[0]["sql"].count(created_at_quoted_name),
            2 if connection.features.can_return_rows_from_bulk_insert else 1,
        )

    @skipUnlessDBFeature(
        "can_return_rows_from_bulk_insert", "supports_expression_defaults"
    )
    def test_db_default_primary_key(self):
        (obj,) = DbDefaultPrimaryKey.objects.bulk_create([DbDefaultPrimaryKey()])
        self.assertIsInstance(obj.id, datetime)


@skipUnlessDBFeature("supports_transactions", "has_bulk_insert")
class BulkCreateTransactionTests(TransactionTestCase):
    available_apps = ["bulk_create"]

    def test_no_unnecessary_transaction(self):
        unused_id = getattr(Country.objects.last(), "id", 10) + 100
        with self.assertNumQueries(1):
            Country.objects.bulk_create(
                [Country(id=unused_id, name="France", iso_two_letter="FR")]
            )
        with self.assertNumQueries(1):
            Country.objects.bulk_create([Country(name="Canada", iso_two_letter="CA")])

    def test_objs_with_and_without_pk(self):
        unused_id = getattr(Country.objects.last(), "id", 10) + 100
        with self.assertNumQueries(4):
            Country.objects.bulk_create(
                [
                    Country(id=unused_id, name="France", iso_two_letter="FR"),
                    Country(name="Canada", iso_two_letter="CA"),
                ]
            )

    def test_multiple_batches(self):
        with self.assertNumQueries(4):
            Country.objects.bulk_create(
                [
                    Country(name="France", iso_two_letter="FR"),
                    Country(name="Canada", iso_two_letter="CA"),
                ],
                batch_size=1,

Review comment (suggestion):
Maybe


```suggestion
        # Find a serial ID that hasn't been used already and has enough of a buffer for the
        # following `bulk_create` call without an explicit pk not to conflict.
        unused_id = getattr(Country.objects.last(), "id", 10) + 100
```

Code after:
        with self.assertNumQueries(1) as ctx:
            DbDefaultModel.objects.bulk_create(
                [
                    DbDefaultModel(name="foo", created_at=timezone.now()),
                    DbDefaultModel(name="bar"),
                ]
            )
        self.assertEqual(
            ctx[0]["sql"].count(created_at_quoted_name),
            2 if connection.features.can_return_rows_from_bulk_insert else 1,
        )

    @skipUnlessDBFeature(
        "can_return_rows_from_bulk_insert", "supports_expression_defaults"
    )
    def test_db_default_primary_key(self):
        (obj,) = DbDefaultPrimaryKey.objects.bulk_create([DbDefaultPrimaryKey()])
        self.assertIsInstance(obj.id, datetime)


@skipUnlessDBFeature("supports_transactions", "has_bulk_insert")
class BulkCreateTransactionTests(TransactionTestCase):
    available_apps = ["bulk_create"]

    def get_unused_country_id(self):
        # Find a serial ID that hasn't been used already and has enough of a
        # buffer for the following `bulk_create` call without an explicit pk
        # not to conflict.
        return getattr(Country.objects.last(), "id", 10) + 100

    def test_no_unnecessary_transaction(self):
        unused_id = self.get_unused_country_id()
        with self.assertNumQueries(1):
            Country.objects.bulk_create(
                [Country(id=unused_id, name="France", iso_two_letter="FR")]
            )
        with self.assertNumQueries(1):
            Country.objects.bulk_create([Country(name="Canada", iso_two_letter="CA")])

    def test_objs_with_and_without_pk(self):
        unused_id = self.get_unused_country_id()
        with self.assertNumQueries(4):
            Country.objects.bulk_create(
                [
                    Country(id=unused_id, name="France", iso_two_letter="FR"),
                    Country(name="Canada", iso_two_letter="CA"),
                ]
            )

    def test_multiple_batches(self):
        with self.assertNumQueries(4):