mirror of
https://github.com/django/django.git
synced 2024-12-23 09:36:06 +00:00
7acef095d7
This required moving the entirety of DELETE SQL generation to the compiler where it should have been in the first place and implementing a specialized compiler on MySQL/MariaDB. The MySQL compiler relies on the "DELETE table FROM table JOIN" syntax for queries spanning over multiple tables.
591 lines
22 KiB
Python
591 lines
22 KiB
Python
from math import ceil
|
|
|
|
from django.db import IntegrityError, connection, models
|
|
from django.db.models.deletion import Collector
|
|
from django.db.models.sql.constants import GET_ITERATOR_CHUNK_SIZE
|
|
from django.test import TestCase, skipIfDBFeature, skipUnlessDBFeature
|
|
|
|
from .models import (
|
|
MR, A, Avatar, Base, Child, HiddenUser, HiddenUserProfile, M, M2MFrom,
|
|
M2MTo, MRNull, Origin, Parent, R, RChild, RChildChild, Referrer, S, T,
|
|
User, create_a, get_default_r,
|
|
)
|
|
|
|
|
|
class OnDeleteTests(TestCase):
|
|
def setUp(self):
|
|
self.DEFAULT = get_default_r()
|
|
|
|
def test_auto(self):
|
|
a = create_a('auto')
|
|
a.auto.delete()
|
|
self.assertFalse(A.objects.filter(name='auto').exists())
|
|
|
|
def test_non_callable(self):
|
|
msg = 'on_delete must be callable.'
|
|
with self.assertRaisesMessage(TypeError, msg):
|
|
models.ForeignKey('self', on_delete=None)
|
|
with self.assertRaisesMessage(TypeError, msg):
|
|
models.OneToOneField('self', on_delete=None)
|
|
|
|
def test_auto_nullable(self):
|
|
a = create_a('auto_nullable')
|
|
a.auto_nullable.delete()
|
|
self.assertFalse(A.objects.filter(name='auto_nullable').exists())
|
|
|
|
def test_setvalue(self):
|
|
a = create_a('setvalue')
|
|
a.setvalue.delete()
|
|
a = A.objects.get(pk=a.pk)
|
|
self.assertEqual(self.DEFAULT, a.setvalue.pk)
|
|
|
|
def test_setnull(self):
|
|
a = create_a('setnull')
|
|
a.setnull.delete()
|
|
a = A.objects.get(pk=a.pk)
|
|
self.assertIsNone(a.setnull)
|
|
|
|
def test_setdefault(self):
|
|
a = create_a('setdefault')
|
|
a.setdefault.delete()
|
|
a = A.objects.get(pk=a.pk)
|
|
self.assertEqual(self.DEFAULT, a.setdefault.pk)
|
|
|
|
def test_setdefault_none(self):
|
|
a = create_a('setdefault_none')
|
|
a.setdefault_none.delete()
|
|
a = A.objects.get(pk=a.pk)
|
|
self.assertIsNone(a.setdefault_none)
|
|
|
|
def test_cascade(self):
|
|
a = create_a('cascade')
|
|
a.cascade.delete()
|
|
self.assertFalse(A.objects.filter(name='cascade').exists())
|
|
|
|
def test_cascade_nullable(self):
|
|
a = create_a('cascade_nullable')
|
|
a.cascade_nullable.delete()
|
|
self.assertFalse(A.objects.filter(name='cascade_nullable').exists())
|
|
|
|
def test_protect(self):
|
|
a = create_a('protect')
|
|
msg = (
|
|
"Cannot delete some instances of model 'R' because they are "
|
|
"referenced through a protected foreign key: 'A.protect'"
|
|
)
|
|
with self.assertRaisesMessage(IntegrityError, msg):
|
|
a.protect.delete()
|
|
|
|
def test_do_nothing(self):
|
|
# Testing DO_NOTHING is a bit harder: It would raise IntegrityError for a normal model,
|
|
# so we connect to pre_delete and set the fk to a known value.
|
|
replacement_r = R.objects.create()
|
|
|
|
def check_do_nothing(sender, **kwargs):
|
|
obj = kwargs['instance']
|
|
obj.donothing_set.update(donothing=replacement_r)
|
|
models.signals.pre_delete.connect(check_do_nothing)
|
|
a = create_a('do_nothing')
|
|
a.donothing.delete()
|
|
a = A.objects.get(pk=a.pk)
|
|
self.assertEqual(replacement_r, a.donothing)
|
|
models.signals.pre_delete.disconnect(check_do_nothing)
|
|
|
|
def test_do_nothing_qscount(self):
|
|
"""
|
|
A models.DO_NOTHING relation doesn't trigger a query.
|
|
"""
|
|
b = Base.objects.create()
|
|
with self.assertNumQueries(1):
|
|
# RelToBase should not be queried.
|
|
b.delete()
|
|
self.assertEqual(Base.objects.count(), 0)
|
|
|
|
def test_inheritance_cascade_up(self):
|
|
child = RChild.objects.create()
|
|
child.delete()
|
|
self.assertFalse(R.objects.filter(pk=child.pk).exists())
|
|
|
|
def test_inheritance_cascade_down(self):
|
|
child = RChild.objects.create()
|
|
parent = child.r_ptr
|
|
parent.delete()
|
|
self.assertFalse(RChild.objects.filter(pk=child.pk).exists())
|
|
|
|
def test_cascade_from_child(self):
|
|
a = create_a('child')
|
|
a.child.delete()
|
|
self.assertFalse(A.objects.filter(name='child').exists())
|
|
self.assertFalse(R.objects.filter(pk=a.child_id).exists())
|
|
|
|
def test_cascade_from_parent(self):
|
|
a = create_a('child')
|
|
R.objects.get(pk=a.child_id).delete()
|
|
self.assertFalse(A.objects.filter(name='child').exists())
|
|
self.assertFalse(RChild.objects.filter(pk=a.child_id).exists())
|
|
|
|
def test_setnull_from_child(self):
|
|
a = create_a('child_setnull')
|
|
a.child_setnull.delete()
|
|
self.assertFalse(R.objects.filter(pk=a.child_setnull_id).exists())
|
|
|
|
a = A.objects.get(pk=a.pk)
|
|
self.assertIsNone(a.child_setnull)
|
|
|
|
def test_setnull_from_parent(self):
|
|
a = create_a('child_setnull')
|
|
R.objects.get(pk=a.child_setnull_id).delete()
|
|
self.assertFalse(RChild.objects.filter(pk=a.child_setnull_id).exists())
|
|
|
|
a = A.objects.get(pk=a.pk)
|
|
self.assertIsNone(a.child_setnull)
|
|
|
|
def test_o2o_setnull(self):
|
|
a = create_a('o2o_setnull')
|
|
a.o2o_setnull.delete()
|
|
a = A.objects.get(pk=a.pk)
|
|
self.assertIsNone(a.o2o_setnull)
|
|
|
|
|
|
class DeletionTests(TestCase):
|
|
|
|
def test_m2m(self):
|
|
m = M.objects.create()
|
|
r = R.objects.create()
|
|
MR.objects.create(m=m, r=r)
|
|
r.delete()
|
|
self.assertFalse(MR.objects.exists())
|
|
|
|
r = R.objects.create()
|
|
MR.objects.create(m=m, r=r)
|
|
m.delete()
|
|
self.assertFalse(MR.objects.exists())
|
|
|
|
m = M.objects.create()
|
|
r = R.objects.create()
|
|
m.m2m.add(r)
|
|
r.delete()
|
|
through = M._meta.get_field('m2m').remote_field.through
|
|
self.assertFalse(through.objects.exists())
|
|
|
|
r = R.objects.create()
|
|
m.m2m.add(r)
|
|
m.delete()
|
|
self.assertFalse(through.objects.exists())
|
|
|
|
m = M.objects.create()
|
|
r = R.objects.create()
|
|
MRNull.objects.create(m=m, r=r)
|
|
r.delete()
|
|
self.assertFalse(not MRNull.objects.exists())
|
|
self.assertFalse(m.m2m_through_null.exists())
|
|
|
|
def test_bulk(self):
|
|
s = S.objects.create(r=R.objects.create())
|
|
for i in range(2 * GET_ITERATOR_CHUNK_SIZE):
|
|
T.objects.create(s=s)
|
|
# 1 (select related `T` instances)
|
|
# + 1 (select related `U` instances)
|
|
# + 2 (delete `T` instances in batches)
|
|
# + 1 (delete `s`)
|
|
self.assertNumQueries(5, s.delete)
|
|
self.assertFalse(S.objects.exists())
|
|
|
|
def test_instance_update(self):
|
|
deleted = []
|
|
related_setnull_sets = []
|
|
|
|
def pre_delete(sender, **kwargs):
|
|
obj = kwargs['instance']
|
|
deleted.append(obj)
|
|
if isinstance(obj, R):
|
|
related_setnull_sets.append([a.pk for a in obj.setnull_set.all()])
|
|
|
|
models.signals.pre_delete.connect(pre_delete)
|
|
a = create_a('update_setnull')
|
|
a.setnull.delete()
|
|
|
|
a = create_a('update_cascade')
|
|
a.cascade.delete()
|
|
|
|
for obj in deleted:
|
|
self.assertIsNone(obj.pk)
|
|
|
|
for pk_list in related_setnull_sets:
|
|
for a in A.objects.filter(id__in=pk_list):
|
|
self.assertIsNone(a.setnull)
|
|
|
|
models.signals.pre_delete.disconnect(pre_delete)
|
|
|
|
def test_deletion_order(self):
|
|
pre_delete_order = []
|
|
post_delete_order = []
|
|
|
|
def log_post_delete(sender, **kwargs):
|
|
pre_delete_order.append((sender, kwargs['instance'].pk))
|
|
|
|
def log_pre_delete(sender, **kwargs):
|
|
post_delete_order.append((sender, kwargs['instance'].pk))
|
|
|
|
models.signals.post_delete.connect(log_post_delete)
|
|
models.signals.pre_delete.connect(log_pre_delete)
|
|
|
|
r = R.objects.create(pk=1)
|
|
s1 = S.objects.create(pk=1, r=r)
|
|
s2 = S.objects.create(pk=2, r=r)
|
|
T.objects.create(pk=1, s=s1)
|
|
T.objects.create(pk=2, s=s2)
|
|
RChild.objects.create(r_ptr=r)
|
|
r.delete()
|
|
self.assertEqual(
|
|
pre_delete_order, [(T, 2), (T, 1), (RChild, 1), (S, 2), (S, 1), (R, 1)]
|
|
)
|
|
self.assertEqual(
|
|
post_delete_order, [(T, 1), (T, 2), (RChild, 1), (S, 1), (S, 2), (R, 1)]
|
|
)
|
|
|
|
models.signals.post_delete.disconnect(log_post_delete)
|
|
models.signals.pre_delete.disconnect(log_pre_delete)
|
|
|
|
def test_relational_post_delete_signals_happen_before_parent_object(self):
|
|
deletions = []
|
|
|
|
def log_post_delete(instance, **kwargs):
|
|
self.assertTrue(R.objects.filter(pk=instance.r_id))
|
|
self.assertIs(type(instance), S)
|
|
deletions.append(instance.id)
|
|
|
|
r = R.objects.create(pk=1)
|
|
S.objects.create(pk=1, r=r)
|
|
|
|
models.signals.post_delete.connect(log_post_delete, sender=S)
|
|
|
|
try:
|
|
r.delete()
|
|
finally:
|
|
models.signals.post_delete.disconnect(log_post_delete)
|
|
|
|
self.assertEqual(len(deletions), 1)
|
|
self.assertEqual(deletions[0], 1)
|
|
|
|
@skipUnlessDBFeature("can_defer_constraint_checks")
|
|
def test_can_defer_constraint_checks(self):
|
|
u = User.objects.create(
|
|
avatar=Avatar.objects.create()
|
|
)
|
|
a = Avatar.objects.get(pk=u.avatar_id)
|
|
# 1 query to find the users for the avatar.
|
|
# 1 query to delete the user
|
|
# 1 query to delete the avatar
|
|
# The important thing is that when we can defer constraint checks there
|
|
# is no need to do an UPDATE on User.avatar to null it out.
|
|
|
|
# Attach a signal to make sure we will not do fast_deletes.
|
|
calls = []
|
|
|
|
def noop(*args, **kwargs):
|
|
calls.append('')
|
|
models.signals.post_delete.connect(noop, sender=User)
|
|
|
|
self.assertNumQueries(3, a.delete)
|
|
self.assertFalse(User.objects.exists())
|
|
self.assertFalse(Avatar.objects.exists())
|
|
self.assertEqual(len(calls), 1)
|
|
models.signals.post_delete.disconnect(noop, sender=User)
|
|
|
|
@skipIfDBFeature("can_defer_constraint_checks")
|
|
def test_cannot_defer_constraint_checks(self):
|
|
u = User.objects.create(
|
|
avatar=Avatar.objects.create()
|
|
)
|
|
# Attach a signal to make sure we will not do fast_deletes.
|
|
calls = []
|
|
|
|
def noop(*args, **kwargs):
|
|
calls.append('')
|
|
models.signals.post_delete.connect(noop, sender=User)
|
|
|
|
a = Avatar.objects.get(pk=u.avatar_id)
|
|
# The below doesn't make sense... Why do we need to null out
|
|
# user.avatar if we are going to delete the user immediately after it,
|
|
# and there are no more cascades.
|
|
# 1 query to find the users for the avatar.
|
|
# 1 query to delete the user
|
|
# 1 query to null out user.avatar, because we can't defer the constraint
|
|
# 1 query to delete the avatar
|
|
self.assertNumQueries(4, a.delete)
|
|
self.assertFalse(User.objects.exists())
|
|
self.assertFalse(Avatar.objects.exists())
|
|
self.assertEqual(len(calls), 1)
|
|
models.signals.post_delete.disconnect(noop, sender=User)
|
|
|
|
def test_hidden_related(self):
|
|
r = R.objects.create()
|
|
h = HiddenUser.objects.create(r=r)
|
|
HiddenUserProfile.objects.create(user=h)
|
|
|
|
r.delete()
|
|
self.assertEqual(HiddenUserProfile.objects.count(), 0)
|
|
|
|
def test_large_delete(self):
|
|
TEST_SIZE = 2000
|
|
objs = [Avatar() for i in range(0, TEST_SIZE)]
|
|
Avatar.objects.bulk_create(objs)
|
|
# Calculate the number of queries needed.
|
|
batch_size = connection.ops.bulk_batch_size(['pk'], objs)
|
|
# The related fetches are done in batches.
|
|
batches = ceil(len(objs) / batch_size)
|
|
# One query for Avatar.objects.all() and then one related fast delete for
|
|
# each batch.
|
|
fetches_to_mem = 1 + batches
|
|
# The Avatar objects are going to be deleted in batches of GET_ITERATOR_CHUNK_SIZE
|
|
queries = fetches_to_mem + TEST_SIZE // GET_ITERATOR_CHUNK_SIZE
|
|
self.assertNumQueries(queries, Avatar.objects.all().delete)
|
|
self.assertFalse(Avatar.objects.exists())
|
|
|
|
def test_large_delete_related(self):
|
|
TEST_SIZE = 2000
|
|
s = S.objects.create(r=R.objects.create())
|
|
for i in range(TEST_SIZE):
|
|
T.objects.create(s=s)
|
|
|
|
batch_size = max(connection.ops.bulk_batch_size(['pk'], range(TEST_SIZE)), 1)
|
|
|
|
# TEST_SIZE / batch_size (select related `T` instances)
|
|
# + 1 (select related `U` instances)
|
|
# + TEST_SIZE / GET_ITERATOR_CHUNK_SIZE (delete `T` instances in batches)
|
|
# + 1 (delete `s`)
|
|
expected_num_queries = ceil(TEST_SIZE / batch_size)
|
|
expected_num_queries += ceil(TEST_SIZE / GET_ITERATOR_CHUNK_SIZE) + 2
|
|
|
|
self.assertNumQueries(expected_num_queries, s.delete)
|
|
self.assertFalse(S.objects.exists())
|
|
self.assertFalse(T.objects.exists())
|
|
|
|
def test_delete_with_keeping_parents(self):
|
|
child = RChild.objects.create()
|
|
parent_id = child.r_ptr_id
|
|
child.delete(keep_parents=True)
|
|
self.assertFalse(RChild.objects.filter(id=child.id).exists())
|
|
self.assertTrue(R.objects.filter(id=parent_id).exists())
|
|
|
|
def test_delete_with_keeping_parents_relationships(self):
|
|
child = RChild.objects.create()
|
|
parent_id = child.r_ptr_id
|
|
parent_referent_id = S.objects.create(r=child.r_ptr).pk
|
|
child.delete(keep_parents=True)
|
|
self.assertFalse(RChild.objects.filter(id=child.id).exists())
|
|
self.assertTrue(R.objects.filter(id=parent_id).exists())
|
|
self.assertTrue(S.objects.filter(pk=parent_referent_id).exists())
|
|
|
|
childchild = RChildChild.objects.create()
|
|
parent_id = childchild.rchild_ptr.r_ptr_id
|
|
child_id = childchild.rchild_ptr_id
|
|
parent_referent_id = S.objects.create(r=childchild.rchild_ptr.r_ptr).pk
|
|
childchild.delete(keep_parents=True)
|
|
self.assertFalse(RChildChild.objects.filter(id=childchild.id).exists())
|
|
self.assertTrue(RChild.objects.filter(id=child_id).exists())
|
|
self.assertTrue(R.objects.filter(id=parent_id).exists())
|
|
self.assertTrue(S.objects.filter(pk=parent_referent_id).exists())
|
|
|
|
def test_queryset_delete_returns_num_rows(self):
|
|
"""
|
|
QuerySet.delete() should return the number of deleted rows and a
|
|
dictionary with the number of deletions for each object type.
|
|
"""
|
|
Avatar.objects.bulk_create([Avatar(desc='a'), Avatar(desc='b'), Avatar(desc='c')])
|
|
avatars_count = Avatar.objects.count()
|
|
deleted, rows_count = Avatar.objects.all().delete()
|
|
self.assertEqual(deleted, avatars_count)
|
|
|
|
# more complex example with multiple object types
|
|
r = R.objects.create()
|
|
h1 = HiddenUser.objects.create(r=r)
|
|
HiddenUser.objects.create(r=r)
|
|
HiddenUserProfile.objects.create(user=h1)
|
|
existed_objs = {
|
|
R._meta.label: R.objects.count(),
|
|
HiddenUser._meta.label: HiddenUser.objects.count(),
|
|
A._meta.label: A.objects.count(),
|
|
MR._meta.label: MR.objects.count(),
|
|
HiddenUserProfile._meta.label: HiddenUserProfile.objects.count(),
|
|
}
|
|
deleted, deleted_objs = R.objects.all().delete()
|
|
for k, v in existed_objs.items():
|
|
self.assertEqual(deleted_objs[k], v)
|
|
|
|
def test_model_delete_returns_num_rows(self):
|
|
"""
|
|
Model.delete() should return the number of deleted rows and a
|
|
dictionary with the number of deletions for each object type.
|
|
"""
|
|
r = R.objects.create()
|
|
h1 = HiddenUser.objects.create(r=r)
|
|
h2 = HiddenUser.objects.create(r=r)
|
|
HiddenUser.objects.create(r=r)
|
|
HiddenUserProfile.objects.create(user=h1)
|
|
HiddenUserProfile.objects.create(user=h2)
|
|
m1 = M.objects.create()
|
|
m2 = M.objects.create()
|
|
MR.objects.create(r=r, m=m1)
|
|
r.m_set.add(m1)
|
|
r.m_set.add(m2)
|
|
r.save()
|
|
existed_objs = {
|
|
R._meta.label: R.objects.count(),
|
|
HiddenUser._meta.label: HiddenUser.objects.count(),
|
|
A._meta.label: A.objects.count(),
|
|
MR._meta.label: MR.objects.count(),
|
|
HiddenUserProfile._meta.label: HiddenUserProfile.objects.count(),
|
|
M.m2m.through._meta.label: M.m2m.through.objects.count(),
|
|
}
|
|
deleted, deleted_objs = r.delete()
|
|
self.assertEqual(deleted, sum(existed_objs.values()))
|
|
for k, v in existed_objs.items():
|
|
self.assertEqual(deleted_objs[k], v)
|
|
|
|
def test_proxied_model_duplicate_queries(self):
|
|
"""
|
|
#25685 - Deleting instances of a model with existing proxy
|
|
classes should not issue multiple queries during cascade
|
|
deletion of referring models.
|
|
"""
|
|
avatar = Avatar.objects.create()
|
|
# One query for the Avatar table and a second for the User one.
|
|
with self.assertNumQueries(2):
|
|
avatar.delete()
|
|
|
|
def test_only_referenced_fields_selected(self):
|
|
"""
|
|
Only referenced fields are selected during cascade deletion SELECT
|
|
unless deletion signals are connected.
|
|
"""
|
|
origin = Origin.objects.create()
|
|
expected_sql = str(
|
|
Referrer.objects.only(
|
|
# Both fields are referenced by SecondReferrer.
|
|
'id', 'unique_field',
|
|
).filter(origin__in=[origin]).query
|
|
)
|
|
with self.assertNumQueries(2) as ctx:
|
|
origin.delete()
|
|
self.assertEqual(ctx.captured_queries[0]['sql'], expected_sql)
|
|
|
|
def receiver(instance, **kwargs):
|
|
pass
|
|
|
|
# All fields are selected if deletion signals are connected.
|
|
for signal_name in ('pre_delete', 'post_delete'):
|
|
with self.subTest(signal=signal_name):
|
|
origin = Origin.objects.create()
|
|
signal = getattr(models.signals, signal_name)
|
|
signal.connect(receiver, sender=Referrer)
|
|
with self.assertNumQueries(2) as ctx:
|
|
origin.delete()
|
|
self.assertIn(
|
|
connection.ops.quote_name('large_field'),
|
|
ctx.captured_queries[0]['sql'],
|
|
)
|
|
signal.disconnect(receiver, sender=Referrer)
|
|
|
|
|
|
class FastDeleteTests(TestCase):
|
|
|
|
def test_fast_delete_fk(self):
|
|
u = User.objects.create(
|
|
avatar=Avatar.objects.create()
|
|
)
|
|
a = Avatar.objects.get(pk=u.avatar_id)
|
|
# 1 query to fast-delete the user
|
|
# 1 query to delete the avatar
|
|
self.assertNumQueries(2, a.delete)
|
|
self.assertFalse(User.objects.exists())
|
|
self.assertFalse(Avatar.objects.exists())
|
|
|
|
def test_fast_delete_m2m(self):
|
|
t = M2MTo.objects.create()
|
|
f = M2MFrom.objects.create()
|
|
f.m2m.add(t)
|
|
# 1 to delete f, 1 to fast-delete m2m for f
|
|
self.assertNumQueries(2, f.delete)
|
|
|
|
def test_fast_delete_revm2m(self):
|
|
t = M2MTo.objects.create()
|
|
f = M2MFrom.objects.create()
|
|
f.m2m.add(t)
|
|
# 1 to delete t, 1 to fast-delete t's m_set
|
|
self.assertNumQueries(2, f.delete)
|
|
|
|
def test_fast_delete_qs(self):
|
|
u1 = User.objects.create()
|
|
u2 = User.objects.create()
|
|
self.assertNumQueries(1, User.objects.filter(pk=u1.pk).delete)
|
|
self.assertEqual(User.objects.count(), 1)
|
|
self.assertTrue(User.objects.filter(pk=u2.pk).exists())
|
|
|
|
def test_fast_delete_instance_set_pk_none(self):
|
|
u = User.objects.create()
|
|
# User can be fast-deleted.
|
|
collector = Collector(using='default')
|
|
self.assertTrue(collector.can_fast_delete(u))
|
|
u.delete()
|
|
self.assertIsNone(u.pk)
|
|
|
|
def test_fast_delete_joined_qs(self):
|
|
a = Avatar.objects.create(desc='a')
|
|
User.objects.create(avatar=a)
|
|
u2 = User.objects.create()
|
|
self.assertNumQueries(1, User.objects.filter(avatar__desc='a').delete)
|
|
self.assertEqual(User.objects.count(), 1)
|
|
self.assertTrue(User.objects.filter(pk=u2.pk).exists())
|
|
|
|
def test_fast_delete_inheritance(self):
|
|
c = Child.objects.create()
|
|
p = Parent.objects.create()
|
|
# 1 for self, 1 for parent
|
|
self.assertNumQueries(2, c.delete)
|
|
self.assertFalse(Child.objects.exists())
|
|
self.assertEqual(Parent.objects.count(), 1)
|
|
self.assertEqual(Parent.objects.filter(pk=p.pk).count(), 1)
|
|
# 1 for self delete, 1 for fast delete of empty "child" qs.
|
|
self.assertNumQueries(2, p.delete)
|
|
self.assertFalse(Parent.objects.exists())
|
|
# 1 for self delete, 1 for fast delete of empty "child" qs.
|
|
c = Child.objects.create()
|
|
p = c.parent_ptr
|
|
self.assertNumQueries(2, p.delete)
|
|
self.assertFalse(Parent.objects.exists())
|
|
self.assertFalse(Child.objects.exists())
|
|
|
|
def test_fast_delete_large_batch(self):
|
|
User.objects.bulk_create(User() for i in range(0, 2000))
|
|
# No problems here - we aren't going to cascade, so we will fast
|
|
# delete the objects in a single query.
|
|
self.assertNumQueries(1, User.objects.all().delete)
|
|
a = Avatar.objects.create(desc='a')
|
|
User.objects.bulk_create(User(avatar=a) for i in range(0, 2000))
|
|
# We don't hit parameter amount limits for a, so just one query for
|
|
# that + fast delete of the related objs.
|
|
self.assertNumQueries(2, a.delete)
|
|
self.assertEqual(User.objects.count(), 0)
|
|
|
|
def test_fast_delete_empty_no_update_can_self_select(self):
|
|
"""
|
|
#25932 - Fast deleting on backends that don't have the
|
|
`no_update_can_self_select` feature should work even if the specified
|
|
filter doesn't match any row.
|
|
"""
|
|
with self.assertNumQueries(1):
|
|
self.assertEqual(
|
|
User.objects.filter(avatar__desc='missing').delete(),
|
|
(0, {'delete.User': 0})
|
|
)
|
|
|
|
def test_fast_delete_combined_relationships(self):
|
|
# The cascading fast-delete of SecondReferrer should be combined
|
|
# in a single DELETE WHERE referrer_id OR unique_field.
|
|
origin = Origin.objects.create()
|
|
referer = Referrer.objects.create(origin=origin, unique_field=42)
|
|
with self.assertNumQueries(2):
|
|
referer.delete()
|