diff options
| author | django-bot <ops@djangoproject.com> | 2022-02-08 12:09:55 +0100 |
|---|---|---|
| committer | Mariusz Felisiak <felisiak.mariusz@gmail.com> | 2022-02-08 12:15:38 +0100 |
| commit | 6a682b38e75d4c975b4c4493565a59f1bc14397c (patch) | |
| tree | 0bd9cda550bea26238656d9f120d769e8b41bb9e /tests/bulk_create | |
| parent | e73ce08888e6f34d3f050377cfd2fbb733be94a9 (diff) | |
[4.0.x] Refs #33476 -- Reformatted code with Black.
Backport of 9c19aff7c7561e3a82978a272ecdaad40dda5c00 from main.
Diffstat (limited to 'tests/bulk_create')
| -rw-r--r-- | tests/bulk_create/models.py | 42 | ||||
| -rw-r--r-- | tests/bulk_create/tests.py | 296 |
2 files changed, 208 insertions, 130 deletions
diff --git a/tests/bulk_create/models.py b/tests/bulk_create/models.py index 586457b192..d04ff58c52 100644 --- a/tests/bulk_create/models.py +++ b/tests/bulk_create/models.py @@ -75,31 +75,43 @@ class BigAutoFieldModel(models.Model): class NullableFields(models.Model): # Fields in db.backends.oracle.BulkInsertMapper big_int_filed = models.BigIntegerField(null=True, default=1) - binary_field = models.BinaryField(null=True, default=b'data') + binary_field = models.BinaryField(null=True, default=b"data") date_field = models.DateField(null=True, default=timezone.now) datetime_field = models.DateTimeField(null=True, default=timezone.now) - decimal_field = models.DecimalField(null=True, max_digits=2, decimal_places=1, default=Decimal('1.1')) + decimal_field = models.DecimalField( + null=True, max_digits=2, decimal_places=1, default=Decimal("1.1") + ) duration_field = models.DurationField(null=True, default=datetime.timedelta(1)) float_field = models.FloatField(null=True, default=3.2) integer_field = models.IntegerField(null=True, default=2) null_boolean_field = models.BooleanField(null=True, default=False) - positive_big_integer_field = models.PositiveBigIntegerField(null=True, default=2 ** 63 - 1) + positive_big_integer_field = models.PositiveBigIntegerField( + null=True, default=2**63 - 1 + ) positive_integer_field = models.PositiveIntegerField(null=True, default=3) - positive_small_integer_field = models.PositiveSmallIntegerField(null=True, default=4) + positive_small_integer_field = models.PositiveSmallIntegerField( + null=True, default=4 + ) small_integer_field = models.SmallIntegerField(null=True, default=5) time_field = models.TimeField(null=True, default=timezone.now) auto_field = models.ForeignKey(NoFields, on_delete=models.CASCADE, null=True) - small_auto_field = models.ForeignKey(SmallAutoFieldModel, on_delete=models.CASCADE, null=True) - big_auto_field = models.ForeignKey(BigAutoFieldModel, on_delete=models.CASCADE, null=True) + small_auto_field = models.ForeignKey( + SmallAutoFieldModel, on_delete=models.CASCADE, null=True + ) + big_auto_field = models.ForeignKey( + BigAutoFieldModel, on_delete=models.CASCADE, null=True + ) # Fields not required in BulkInsertMapper - char_field = models.CharField(null=True, max_length=4, default='char') - email_field = models.EmailField(null=True, default='user@example.com') - file_field = models.FileField(null=True, default='file.txt') - file_path_field = models.FilePathField(path='/tmp', null=True, default='file.txt') - generic_ip_address_field = models.GenericIPAddressField(null=True, default='127.0.0.1') + char_field = models.CharField(null=True, max_length=4, default="char") + email_field = models.EmailField(null=True, default="user@example.com") + file_field = models.FileField(null=True, default="file.txt") + file_path_field = models.FilePathField(path="/tmp", null=True, default="file.txt") + generic_ip_address_field = models.GenericIPAddressField( + null=True, default="127.0.0.1" + ) if Image: - image_field = models.ImageField(null=True, default='image.jpg') - slug_field = models.SlugField(null=True, default='slug') - text_field = models.TextField(null=True, default='text') - url_field = models.URLField(null=True, default='/') + image_field = models.ImageField(null=True, default="image.jpg") + slug_field = models.SlugField(null=True, default="slug") + text_field = models.TextField(null=True, default="text") + url_field = models.URLField(null=True, default="/") uuid_field = models.UUIDField(null=True, default=uuid.uuid4) diff --git a/tests/bulk_create/tests.py b/tests/bulk_create/tests.py index 2ee54c382f..45111d612f 100644 --- a/tests/bulk_create/tests.py +++ b/tests/bulk_create/tests.py @@ -5,13 +5,26 @@ from django.db import IntegrityError, NotSupportedError, connection from django.db.models import FileField, Value from django.db.models.functions import Lower from django.test import ( - TestCase, override_settings, skipIfDBFeature, skipUnlessDBFeature, + TestCase, + override_settings, + skipIfDBFeature, + skipUnlessDBFeature, ) from .models import ( - BigAutoFieldModel, Country, NoFields, NullableFields, Pizzeria, - ProxyCountry, ProxyMultiCountry, ProxyMultiProxyCountry, ProxyProxyCountry, - Restaurant, SmallAutoFieldModel, State, TwoFields, + BigAutoFieldModel, + Country, + NoFields, + NullableFields, + Pizzeria, + ProxyCountry, + ProxyMultiCountry, + ProxyMultiProxyCountry, + ProxyProxyCountry, + Restaurant, + SmallAutoFieldModel, + State, + TwoFields, ) @@ -21,142 +34,179 @@ class BulkCreateTests(TestCase): Country(name="United States of America", iso_two_letter="US"), Country(name="The Netherlands", iso_two_letter="NL"), Country(name="Germany", iso_two_letter="DE"), - Country(name="Czech Republic", iso_two_letter="CZ") + Country(name="Czech Republic", iso_two_letter="CZ"), ] def test_simple(self): created = Country.objects.bulk_create(self.data) self.assertEqual(created, self.data) - self.assertQuerysetEqual(Country.objects.order_by("-name"), [ - "United States of America", "The Netherlands", "Germany", "Czech Republic" - ], attrgetter("name")) + self.assertQuerysetEqual( + Country.objects.order_by("-name"), + [ + "United States of America", + "The Netherlands", + "Germany", + "Czech Republic", + ], + attrgetter("name"), + ) created = Country.objects.bulk_create([]) self.assertEqual(created, []) self.assertEqual(Country.objects.count(), 4) - @skipUnlessDBFeature('has_bulk_insert') + @skipUnlessDBFeature("has_bulk_insert") def test_efficiency(self): with self.assertNumQueries(1): Country.objects.bulk_create(self.data) - @skipUnlessDBFeature('has_bulk_insert') + @skipUnlessDBFeature("has_bulk_insert") def test_long_non_ascii_text(self): """ Inserting non-ASCII values with a length in the range 2001 to 4000 characters, i.e. 4002 to 8000 bytes, must be set as a CLOB on Oracle (#22144). """ - Country.objects.bulk_create([Country(description='Ж' * 3000)]) + Country.objects.bulk_create([Country(description="Ж" * 3000)]) self.assertEqual(Country.objects.count(), 1) - @skipUnlessDBFeature('has_bulk_insert') + @skipUnlessDBFeature("has_bulk_insert") def test_long_and_short_text(self): - Country.objects.bulk_create([ - Country(description='a' * 4001), - Country(description='a'), - Country(description='Ж' * 2001), - Country(description='Ж'), - ]) + Country.objects.bulk_create( + [ + Country(description="a" * 4001), + Country(description="a"), + Country(description="Ж" * 2001), + Country(description="Ж"), + ] + ) self.assertEqual(Country.objects.count(), 4) def test_multi_table_inheritance_unsupported(self): expected_message = "Can't bulk create a multi-table inherited model" with self.assertRaisesMessage(ValueError, expected_message): - Pizzeria.objects.bulk_create([ - Pizzeria(name="The Art of Pizza"), - ]) + Pizzeria.objects.bulk_create( + [ + Pizzeria(name="The Art of Pizza"), + ] + ) with self.assertRaisesMessage(ValueError, expected_message): - ProxyMultiCountry.objects.bulk_create([ - ProxyMultiCountry(name="Fillory", iso_two_letter="FL"), - ]) + ProxyMultiCountry.objects.bulk_create( + [ + ProxyMultiCountry(name="Fillory", iso_two_letter="FL"), + ] + ) with self.assertRaisesMessage(ValueError, expected_message): - ProxyMultiProxyCountry.objects.bulk_create([ - ProxyMultiProxyCountry(name="Fillory", iso_two_letter="FL"), - ]) + ProxyMultiProxyCountry.objects.bulk_create( + [ + ProxyMultiProxyCountry(name="Fillory", iso_two_letter="FL"), + ] + ) def test_proxy_inheritance_supported(self): - ProxyCountry.objects.bulk_create([ - ProxyCountry(name="Qwghlm", iso_two_letter="QW"), - Country(name="Tortall", iso_two_letter="TA"), - ]) - self.assertQuerysetEqual(ProxyCountry.objects.all(), { - "Qwghlm", "Tortall" - }, attrgetter("name"), ordered=False) + ProxyCountry.objects.bulk_create( + [ + ProxyCountry(name="Qwghlm", iso_two_letter="QW"), + Country(name="Tortall", iso_two_letter="TA"), + ] + ) + self.assertQuerysetEqual( + ProxyCountry.objects.all(), + {"Qwghlm", "Tortall"}, + attrgetter("name"), + ordered=False, + ) - ProxyProxyCountry.objects.bulk_create([ - ProxyProxyCountry(name="Netherlands", iso_two_letter="NT"), - ]) - self.assertQuerysetEqual(ProxyProxyCountry.objects.all(), { - "Qwghlm", "Tortall", "Netherlands", - }, attrgetter("name"), ordered=False) + ProxyProxyCountry.objects.bulk_create( + [ + ProxyProxyCountry(name="Netherlands", iso_two_letter="NT"), + ] + ) + self.assertQuerysetEqual( + ProxyProxyCountry.objects.all(), + { + "Qwghlm", + "Tortall", + "Netherlands", + }, + attrgetter("name"), + ordered=False, + ) 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")) + 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') + @skipUnlessDBFeature("has_bulk_insert") def test_non_auto_increment_pk_efficiency(self): with self.assertNumQueries(1): - 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")) + 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"), + ) - @skipIfDBFeature('allows_auto_pk_0') + @skipIfDBFeature("allows_auto_pk_0") def test_zero_as_autoval(self): """ Zero as id for AutoField should raise exception in MySQL, because MySQL does not allow zero for automatic primary key if the NO_AUTO_VALUE_ON_ZERO SQL mode is not enabled. """ - valid_country = Country(name='Germany', iso_two_letter='DE') - invalid_country = Country(id=0, name='Poland', iso_two_letter='PL') - msg = 'The database backend does not accept 0 as a value for AutoField.' + valid_country = Country(name="Germany", iso_two_letter="DE") + invalid_country = Country(id=0, name="Poland", iso_two_letter="PL") + msg = "The database backend does not accept 0 as a value for AutoField." with self.assertRaisesMessage(ValueError, msg): Country.objects.bulk_create([valid_country, invalid_country]) def test_batch_same_vals(self): # SQLite had a problem where all the same-valued models were # collapsed to one insert. - Restaurant.objects.bulk_create([ - Restaurant(name='foo') for i in range(0, 2) - ]) + Restaurant.objects.bulk_create([Restaurant(name="foo") for i in range(0, 2)]) self.assertEqual(Restaurant.objects.count(), 2) def test_large_batch(self): - TwoFields.objects.bulk_create([ - TwoFields(f1=i, f2=i + 1) for i in range(0, 1001) - ]) + TwoFields.objects.bulk_create( + [TwoFields(f1=i, f2=i + 1) for i in range(0, 1001)] + ) self.assertEqual(TwoFields.objects.count(), 1001) self.assertEqual( - TwoFields.objects.filter(f1__gte=450, f1__lte=550).count(), - 101) + TwoFields.objects.filter(f1__gte=450, f1__lte=550).count(), 101 + ) self.assertEqual(TwoFields.objects.filter(f2__gte=901).count(), 101) - @skipUnlessDBFeature('has_bulk_insert') + @skipUnlessDBFeature("has_bulk_insert") def test_large_single_field_batch(self): # SQLite had a problem with more than 500 UNIONed selects in single # query. - Restaurant.objects.bulk_create([ - Restaurant() for i in range(0, 501) - ]) + Restaurant.objects.bulk_create([Restaurant() for i in range(0, 501)]) - @skipUnlessDBFeature('has_bulk_insert') + @skipUnlessDBFeature("has_bulk_insert") def test_large_batch_efficiency(self): with override_settings(DEBUG=True): connection.queries_log.clear() - TwoFields.objects.bulk_create([ - TwoFields(f1=i, f2=i + 1) for i in range(0, 1001) - ]) + TwoFields.objects.bulk_create( + [TwoFields(f1=i, f2=i + 1) for i in range(0, 1001)] + ) self.assertLess(len(connection.queries), 10) def test_large_batch_mixed(self): @@ -164,10 +214,12 @@ class BulkCreateTests(TestCase): Test inserting a large batch with objects having primary key set mixed together with objects without PK set. """ - TwoFields.objects.bulk_create([ - TwoFields(id=i if i % 2 == 0 else None, f1=i, f2=i + 1) - for i in range(100000, 101000) - ]) + 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.assertEqual(TwoFields.objects.count(), 1000) # We can't assume much about the ID's created, except that the above # created IDs must exist. @@ -175,7 +227,7 @@ class BulkCreateTests(TestCase): self.assertEqual(TwoFields.objects.filter(id__in=id_range).count(), 500) self.assertEqual(TwoFields.objects.exclude(id__in=id_range).count(), 500) - @skipUnlessDBFeature('has_bulk_insert') + @skipUnlessDBFeature("has_bulk_insert") def test_large_batch_mixed_efficiency(self): """ Test inserting a large batch with objects having primary key set @@ -183,9 +235,12 @@ class BulkCreateTests(TestCase): """ with override_settings(DEBUG=True): connection.queries_log.clear() - TwoFields.objects.bulk_create([ - TwoFields(id=i if i % 2 == 0 else None, f1=i, f2=i + 1) - for i in range(100000, 101000)]) + 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.assertLess(len(connection.queries), 10) def test_explicit_batch_size(self): @@ -207,7 +262,7 @@ class BulkCreateTests(TestCase): NoFields.objects.bulk_create([NoFields() for i in range(2)]) self.assertEqual(NoFields.objects.count(), 2) - @skipUnlessDBFeature('has_bulk_insert') + @skipUnlessDBFeature("has_bulk_insert") def test_explicit_batch_size_efficiency(self): objs = [TwoFields(f1=i, f2=i) for i in range(0, 100)] with self.assertNumQueries(2): @@ -216,50 +271,59 @@ class BulkCreateTests(TestCase): with self.assertNumQueries(1): TwoFields.objects.bulk_create(objs, len(objs)) - @skipUnlessDBFeature('has_bulk_insert') + @skipUnlessDBFeature("has_bulk_insert") def test_explicit_batch_size_respects_max_batch_size(self): objs = [Country() for i in range(1000)] - fields = ['name', 'iso_two_letter', 'description'] + fields = ["name", "iso_two_letter", "description"] max_batch_size = max(connection.ops.bulk_batch_size(fields, objs), 1) with self.assertNumQueries(ceil(len(objs) / max_batch_size)): Country.objects.bulk_create(objs, batch_size=max_batch_size + 1) - @skipUnlessDBFeature('has_bulk_insert') + @skipUnlessDBFeature("has_bulk_insert") def test_bulk_insert_expressions(self): - Restaurant.objects.bulk_create([ - Restaurant(name="Sam's Shake Shack"), - Restaurant(name=Lower(Value("Betty's Beetroot Bar"))) - ]) + Restaurant.objects.bulk_create( + [ + Restaurant(name="Sam's Shake Shack"), + Restaurant(name=Lower(Value("Betty's Beetroot Bar"))), + ] + ) bbb = Restaurant.objects.filter(name="betty's beetroot bar") self.assertEqual(bbb.count(), 1) - @skipUnlessDBFeature('has_bulk_insert') + @skipUnlessDBFeature("has_bulk_insert") def test_bulk_insert_nullable_fields(self): fk_to_auto_fields = { - 'auto_field': NoFields.objects.create(), - 'small_auto_field': SmallAutoFieldModel.objects.create(), - 'big_auto_field': BigAutoFieldModel.objects.create(), + "auto_field": NoFields.objects.create(), + "small_auto_field": SmallAutoFieldModel.objects.create(), + "big_auto_field": BigAutoFieldModel.objects.create(), } # NULL can be mixed with other values in nullable fields - nullable_fields = [field for field in NullableFields._meta.get_fields() if field.name != 'id'] - NullableFields.objects.bulk_create([ - NullableFields(**{**fk_to_auto_fields, field.name: None}) - for field in nullable_fields - ]) + nullable_fields = [ + field for field in NullableFields._meta.get_fields() if field.name != "id" + ] + NullableFields.objects.bulk_create( + [ + NullableFields(**{**fk_to_auto_fields, field.name: None}) + for field in nullable_fields + ] + ) self.assertEqual(NullableFields.objects.count(), len(nullable_fields)) for field in nullable_fields: with self.subTest(field=field): - field_value = '' if isinstance(field, FileField) else None - self.assertEqual(NullableFields.objects.filter(**{field.name: field_value}).count(), 1) + field_value = "" if isinstance(field, FileField) else None + self.assertEqual( + NullableFields.objects.filter(**{field.name: field_value}).count(), + 1, + ) - @skipUnlessDBFeature('can_return_rows_from_bulk_insert') + @skipUnlessDBFeature("can_return_rows_from_bulk_insert") def test_set_pk_and_insert_single_item(self): with self.assertNumQueries(1): countries = Country.objects.bulk_create([self.data[0]]) self.assertEqual(len(countries), 1) self.assertEqual(Country.objects.get(pk=countries[0].pk), countries[0]) - @skipUnlessDBFeature('can_return_rows_from_bulk_insert') + @skipUnlessDBFeature("can_return_rows_from_bulk_insert") def test_set_pk_and_query_efficiency(self): with self.assertNumQueries(1): countries = Country.objects.bulk_create(self.data) @@ -269,10 +333,10 @@ class BulkCreateTests(TestCase): self.assertEqual(Country.objects.get(pk=countries[2].pk), countries[2]) self.assertEqual(Country.objects.get(pk=countries[3].pk), countries[3]) - @skipUnlessDBFeature('can_return_rows_from_bulk_insert') + @skipUnlessDBFeature("can_return_rows_from_bulk_insert") def test_set_state(self): - country_nl = Country(name='Netherlands', iso_two_letter='NL') - country_be = Country(name='Belgium', iso_two_letter='BE') + country_nl = Country(name="Netherlands", iso_two_letter="NL") + country_be = Country(name="Belgium", iso_two_letter="BE") Country.objects.bulk_create([country_nl]) country_be.save() # Objects save via bulk_create() and save() should have equal state. @@ -280,21 +344,21 @@ class BulkCreateTests(TestCase): self.assertEqual(country_nl._state.db, country_be._state.db) def test_set_state_with_pk_specified(self): - state_ca = State(two_letter_code='CA') - state_ny = State(two_letter_code='NY') + state_ca = State(two_letter_code="CA") + state_ny = State(two_letter_code="NY") State.objects.bulk_create([state_ca]) state_ny.save() # Objects save via bulk_create() and save() should have equal state. self.assertEqual(state_ca._state.adding, state_ny._state.adding) self.assertEqual(state_ca._state.db, state_ny._state.db) - @skipIfDBFeature('supports_ignore_conflicts') + @skipIfDBFeature("supports_ignore_conflicts") def test_ignore_conflicts_value_error(self): - message = 'This database backend does not support ignoring conflicts.' + message = "This database backend does not support ignoring conflicts." with self.assertRaisesMessage(NotSupportedError, message): TwoFields.objects.bulk_create(self.data, ignore_conflicts=True) - @skipUnlessDBFeature('supports_ignore_conflicts') + @skipUnlessDBFeature("supports_ignore_conflicts") def test_ignore_conflicts_ignore(self): data = [ TwoFields(f1=1, f2=1), @@ -315,7 +379,9 @@ class BulkCreateTests(TestCase): self.assertIsNone(conflicting_objects[1].pk) # New objects are created and conflicts are ignored. new_object = TwoFields(f1=4, f2=4) - TwoFields.objects.bulk_create(conflicting_objects + [new_object], ignore_conflicts=True) + TwoFields.objects.bulk_create( + conflicting_objects + [new_object], ignore_conflicts=True + ) self.assertEqual(TwoFields.objects.count(), 4) self.assertIsNone(new_object.pk) # Without ignore_conflicts=True, there's a problem. @@ -330,7 +396,7 @@ class BulkCreateTests(TestCase): child = NullableFields.objects.get(integer_field=88) self.assertEqual(child.auto_field, parent) - @skipUnlessDBFeature('can_return_rows_from_bulk_insert') + @skipUnlessDBFeature("can_return_rows_from_bulk_insert") def test_nullable_fk_after_parent_bulk_create(self): parent = NoFields() child = NullableFields(auto_field=parent, integer_field=88) @@ -349,6 +415,6 @@ class BulkCreateTests(TestCase): NullableFields.objects.bulk_create([NullableFields(auto_field=parent)]) def test_invalid_batch_size_exception(self): - msg = 'Batch size must be a positive integer.' + msg = "Batch size must be a positive integer." with self.assertRaisesMessage(ValueError, msg): Country.objects.bulk_create([], batch_size=-1) |
