diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/modeltests/delete/__init__.py | 1 | ||||
| -rw-r--r-- | tests/modeltests/delete/models.py | 124 | ||||
| -rw-r--r-- | tests/modeltests/delete/tests.py | 328 | ||||
| -rw-r--r-- | tests/modeltests/invalid_models/models.py | 9 | ||||
| -rw-r--r-- | tests/regressiontests/admin_util/models.py | 4 | ||||
| -rw-r--r-- | tests/regressiontests/admin_util/tests.py | 55 |
6 files changed, 357 insertions, 164 deletions
diff --git a/tests/modeltests/delete/__init__.py b/tests/modeltests/delete/__init__.py index 8b13789179..e69de29bb2 100644 --- a/tests/modeltests/delete/__init__.py +++ b/tests/modeltests/delete/__init__.py @@ -1 +0,0 @@ - diff --git a/tests/modeltests/delete/models.py b/tests/modeltests/delete/models.py index 9c81f6b8f8..ee9e99975f 100644 --- a/tests/modeltests/delete/models.py +++ b/tests/modeltests/delete/models.py @@ -1,42 +1,106 @@ -# coding: utf-8 -""" -Tests for some corner cases with deleting. -""" +from django.db import models, IntegrityError -from django.db import models -class DefaultRepr(object): - def __repr__(self): - return u"<%s: %s>" % (self.__class__.__name__, self.__dict__) +class R(models.Model): + is_default = models.BooleanField(default=False) -class A(DefaultRepr, models.Model): + def __str__(self): + return "%s" % self.pk + + +get_default_r = lambda: R.objects.get_or_create(is_default=True)[0] + + +class S(models.Model): + r = models.ForeignKey(R) + + +class T(models.Model): + s = models.ForeignKey(S) + + +class U(models.Model): + t = models.ForeignKey(T) + + +class RChild(R): pass -class B(DefaultRepr, models.Model): - a = models.ForeignKey(A) -class C(DefaultRepr, models.Model): - b = models.ForeignKey(B) +class A(models.Model): + name = models.CharField(max_length=30) + + auto = models.ForeignKey(R, related_name="auto_set") + auto_nullable = models.ForeignKey(R, null=True, + related_name='auto_nullable_set') + setvalue = models.ForeignKey(R, on_delete=models.SET(get_default_r), + related_name='setvalue') + setnull = models.ForeignKey(R, on_delete=models.SET_NULL, null=True, + related_name='setnull_set') + setdefault = models.ForeignKey(R, on_delete=models.SET_DEFAULT, + default=get_default_r, related_name='setdefault_set') + setdefault_none = models.ForeignKey(R, on_delete=models.SET_DEFAULT, + default=None, null=True, related_name='setnull_nullable_set') + cascade = models.ForeignKey(R, on_delete=models.CASCADE, + related_name='cascade_set') + cascade_nullable = models.ForeignKey(R, on_delete=models.CASCADE, null=True, + related_name='cascade_nullable_set') + protect = models.ForeignKey(R, on_delete=models.PROTECT, null=True) + donothing = models.ForeignKey(R, on_delete=models.DO_NOTHING, null=True, + related_name='donothing_set') + child = models.ForeignKey(RChild, related_name="child") + child_setnull = models.ForeignKey(RChild, on_delete=models.SET_NULL, null=True, + related_name="child_setnull") + + # A OneToOneField is just a ForeignKey unique=True, so we don't duplicate + # all the tests; just one smoke test to ensure on_delete works for it as + # well. + o2o_setnull = models.ForeignKey(R, null=True, + on_delete=models.SET_NULL, related_name="o2o_nullable_set") + + +def create_a(name): + a = A(name=name) + for name in ('auto', 'auto_nullable', 'setvalue', 'setnull', 'setdefault', + 'setdefault_none', 'cascade', 'cascade_nullable', 'protect', + 'donothing', 'o2o_setnull'): + r = R.objects.create() + setattr(a, name, r) + a.child = RChild.objects.create() + a.child_setnull = RChild.objects.create() + a.save() + return a + + +class M(models.Model): + m2m = models.ManyToManyField(R, related_name="m_set") + m2m_through = models.ManyToManyField(R, through="MR", + related_name="m_through_set") + m2m_through_null = models.ManyToManyField(R, through="MRNull", + related_name="m_through_null_set") + + +class MR(models.Model): + m = models.ForeignKey(M) + r = models.ForeignKey(R) + + +class MRNull(models.Model): + m = models.ForeignKey(M) + r = models.ForeignKey(R, null=True, on_delete=models.SET_NULL) + + +class Avatar(models.Model): + pass -class D(DefaultRepr, models.Model): - c = models.ForeignKey(C) - a = models.ForeignKey(A) -# Simplified, we have: -# A -# B -> A -# C -> B -# D -> C -# D -> A +class User(models.Model): + avatar = models.ForeignKey(Avatar, null=True) -# So, we must delete Ds first of all, then Cs then Bs then As. -# However, if we start at As, we might find Bs first (in which -# case things will be nice), or find Ds first. -# Some mutually dependent models, but nullable -class E(DefaultRepr, models.Model): - f = models.ForeignKey('F', null=True, related_name='e_rel') +class HiddenUser(models.Model): + r = models.ForeignKey(R, related_name="+") -class F(DefaultRepr, models.Model): - e = models.ForeignKey(E, related_name='f_rel') +class HiddenUserProfile(models.Model): + user = models.ForeignKey(HiddenUser) diff --git a/tests/modeltests/delete/tests.py b/tests/modeltests/delete/tests.py index 7927cce1c1..2a209b1630 100644 --- a/tests/modeltests/delete/tests.py +++ b/tests/modeltests/delete/tests.py @@ -1,135 +1,253 @@ -from django.db.models import sql -from django.db.models.loading import cache -from django.db.models.query import CollectedObjects -from django.db.models.query_utils import CyclicDependency -from django.test import TestCase +from django.db import models, IntegrityError +from django.test import TestCase, skipUnlessDBFeature, skipIfDBFeature -from models import A, B, C, D, E, F +from modeltests.delete.models import (R, RChild, S, T, U, A, M, MR, MRNull, + create_a, get_default_r, User, Avatar, HiddenUser, HiddenUserProfile) -class DeleteTests(TestCase): - def clear_rel_obj_caches(self, *models): - for m in models: - if hasattr(m._meta, '_related_objects_cache'): - del m._meta._related_objects_cache +class OnDeleteTests(TestCase): + def setUp(self): + self.DEFAULT = get_default_r() - def order_models(self, *models): - cache.app_models["delete"].keyOrder = models + def test_auto(self): + a = create_a('auto') + a.auto.delete() + self.assertFalse(A.objects.filter(name='auto').exists()) - def setUp(self): - self.order_models("a", "b", "c", "d", "e", "f") - self.clear_rel_obj_caches(A, B, C, D, E, F) + 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) + + def test_setnull(self): + a = create_a('setnull') + a.setnull.delete() + a = A.objects.get(pk=a.pk) + self.assertEqual(None, 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) + + def test_setdefault_none(self): + a = create_a('setdefault_none') + a.setdefault_none.delete() + a = A.objects.get(pk=a.pk) + self.assertEqual(None, 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') + self.assertRaises(IntegrityError, 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_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.assertEqual(None, 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()) - def tearDown(self): - self.order_models("a", "b", "c", "d", "e", "f") - self.clear_rel_obj_caches(A, B, C, D, E, F) + a = A.objects.get(pk=a.pk) + self.assertEqual(None, a.child_setnull) - def test_collected_objects(self): - g = CollectedObjects() - self.assertFalse(g.add("key1", 1, "item1", None)) - self.assertEqual(g["key1"], {1: "item1"}) + def test_o2o_setnull(self): + a = create_a('o2o_setnull') + a.o2o_setnull.delete() + a = A.objects.get(pk=a.pk) + self.assertEqual(None, a.o2o_setnull) - self.assertFalse(g.add("key2", 1, "item1", "key1")) - self.assertFalse(g.add("key2", 2, "item2", "key1")) - self.assertEqual(g["key2"], {1: "item1", 2: "item2"}) +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()) - self.assertFalse(g.add("key3", 1, "item1", "key1")) - self.assertTrue(g.add("key3", 1, "item1", "key2")) - self.assertEqual(g.ordered_keys(), ["key3", "key2", "key1"]) + r = R.objects.create() + MR.objects.create(m=m, r=r) + m.delete() + self.assertFalse(MR.objects.exists()) - self.assertTrue(g.add("key2", 1, "item1", "key3")) - self.assertRaises(CyclicDependency, g.ordered_keys) + m = M.objects.create() + r = R.objects.create() + m.m2m.add(r) + r.delete() + through = M._meta.get_field('m2m').rel.through + self.assertFalse(through.objects.exists()) - def test_delete(self): - ## Second, test the usage of CollectedObjects by Model.delete() + r = R.objects.create() + m.m2m.add(r) + m.delete() + self.assertFalse(through.objects.exists()) - # Due to the way that transactions work in the test harness, doing - # m.delete() here can work but fail in a real situation, since it may - # delete all objects, but not in the right order. So we manually check - # that the order of deletion is correct. + 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()) - # Also, it is possible that the order is correct 'accidentally', due - # solely to order of imports etc. To check this, we set the order that - # 'get_models()' will retrieve to a known 'nice' order, and then try - # again with a known 'tricky' order. Slightly naughty access to - # internals here :-) + def test_bulk(self): + from django.db.models.sql.constants import GET_ITERATOR_CHUNK_SIZE + s = S.objects.create(r=R.objects.create()) + for i in xrange(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()) - # If implementation changes, then the tests may need to be simplified: - # - remove the lines that set the .keyOrder and clear the related - # object caches - # - remove the second set of tests (with a2, b2 etc) + 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(list(a.pk for a in obj.setnull_set.all())) - a1 = A.objects.create() - b1 = B.objects.create(a=a1) - c1 = C.objects.create(b=b1) - d1 = D.objects.create(c=c1, a=a1) + models.signals.pre_delete.connect(pre_delete) + a = create_a('update_setnull') + a.setnull.delete() - o = CollectedObjects() - a1._collect_sub_objects(o) - self.assertEqual(o.keys(), [D, C, B, A]) - a1.delete() + a = create_a('update_cascade') + a.cascade.delete() - # Same again with a known bad order - self.order_models("d", "c", "b", "a") - self.clear_rel_obj_caches(A, B, C, D) + for obj in deleted: + self.assertEqual(None, obj.pk) - a2 = A.objects.create() - b2 = B.objects.create(a=a2) - c2 = C.objects.create(b=b2) - d2 = D.objects.create(c=c2, a=a2) + for pk_list in related_setnull_sets: + for a in A.objects.filter(id__in=pk_list): + self.assertEqual(None, a.setnull) - o = CollectedObjects() - a2._collect_sub_objects(o) - self.assertEqual(o.keys(), [D, C, B, A]) - a2.delete() + models.signals.pre_delete.disconnect(pre_delete) - def test_collected_objects_null(self): - g = CollectedObjects() - self.assertFalse(g.add("key1", 1, "item1", None)) - self.assertFalse(g.add("key2", 1, "item1", "key1", nullable=True)) - self.assertTrue(g.add("key1", 1, "item1", "key2")) - self.assertEqual(g.ordered_keys(), ["key1", "key2"]) + def test_deletion_order(self): + pre_delete_order = [] + post_delete_order = [] - def test_delete_nullable(self): - e1 = E.objects.create() - f1 = F.objects.create(e=e1) - e1.f = f1 - e1.save() + def log_post_delete(sender, **kwargs): + pre_delete_order.append((sender, kwargs['instance'].pk)) - # Since E.f is nullable, we should delete F first (after nulling out - # the E.f field), then E. + def log_pre_delete(sender, **kwargs): + post_delete_order.append((sender, kwargs['instance'].pk)) - o = CollectedObjects() - e1._collect_sub_objects(o) - self.assertEqual(o.keys(), [F, E]) + models.signals.post_delete.connect(log_post_delete) + models.signals.pre_delete.connect(log_pre_delete) - # temporarily replace the UpdateQuery class to verify that E.f is - # actually nulled out first + r = R.objects.create(pk=1) + s1 = S.objects.create(pk=1, r=r) + s2 = S.objects.create(pk=2, r=r) + t1 = T.objects.create(pk=1, s=s1) + t2 = T.objects.create(pk=2, s=s2) + r.delete() + self.assertEqual( + pre_delete_order, [(T, 2), (T, 1), (S, 2), (S, 1), (R, 1)] + ) + self.assertEqual( + post_delete_order, [(T, 1), (T, 2), (S, 1), (S, 2), (R, 1)] + ) - logged = [] - class LoggingUpdateQuery(sql.UpdateQuery): - def clear_related(self, related_field, pk_list, using): - logged.append(related_field.name) - return super(LoggingUpdateQuery, self).clear_related(related_field, pk_list, using) - original = sql.UpdateQuery - sql.UpdateQuery = LoggingUpdateQuery + models.signals.post_delete.disconnect(log_post_delete) + models.signals.post_delete.disconnect(log_pre_delete) - e1.delete() - self.assertEqual(logged, ["f"]) - logged = [] + @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. + self.assertNumQueries(3, a.delete) + self.assertFalse(User.objects.exists()) + self.assertFalse(Avatar.objects.exists()) - e2 = E.objects.create() - f2 = F.objects.create(e=e2) - e2.f = f2 - e2.save() + @skipIfDBFeature("can_defer_constraint_checks") + def test_cannot_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 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()) - # Same deal as before, though we are starting from the other object. - o = CollectedObjects() - f2._collect_sub_objects(o) - self.assertEqual(o.keys(), [F, E]) - f2.delete() - self.assertEqual(logged, ["f"]) - logged = [] + def test_hidden_related(self): + r = R.objects.create() + h = HiddenUser.objects.create(r=r) + p = HiddenUserProfile.objects.create(user=h) - sql.UpdateQuery = original + r.delete() + self.assertEqual(HiddenUserProfile.objects.count(), 0) diff --git a/tests/modeltests/invalid_models/models.py b/tests/modeltests/invalid_models/models.py index ef3edd8563..6e2b579a33 100644 --- a/tests/modeltests/invalid_models/models.py +++ b/tests/modeltests/invalid_models/models.py @@ -210,6 +210,13 @@ class NonExistingOrderingWithSingleUnderscore(models.Model): class Meta: ordering = ("does_not_exist",) +class InvalidSetNull(models.Model): + fk = models.ForeignKey('self', on_delete=models.SET_NULL) + +class InvalidSetDefault(models.Model): + fk = models.ForeignKey('self', on_delete=models.SET_DEFAULT) + + model_errors = """invalid_models.fielderrors: "charfield": CharFields require a "max_length" attribute that is a positive integer. invalid_models.fielderrors: "charfield2": CharFields require a "max_length" attribute that is a positive integer. invalid_models.fielderrors: "charfield3": CharFields require a "max_length" attribute that is a positive integer. @@ -315,4 +322,6 @@ invalid_models.uniquem2m: ManyToManyFields cannot be unique. Remove the unique invalid_models.nonuniquefktarget1: Field 'bad' under model 'FKTarget' must have a unique=True constraint. invalid_models.nonuniquefktarget2: Field 'bad' under model 'FKTarget' must have a unique=True constraint. invalid_models.nonexistingorderingwithsingleunderscore: "ordering" refers to "does_not_exist", a field that doesn't exist. +invalid_models.invalidsetnull: 'fk' specifies on_delete=SET_NULL, but cannot be null. +invalid_models.invalidsetdefault: 'fk' specifies on_delete=SET_DEFAULT, but has no default value. """ diff --git a/tests/regressiontests/admin_util/models.py b/tests/regressiontests/admin_util/models.py index 3191a55a2b..d5e29e9aee 100644 --- a/tests/regressiontests/admin_util/models.py +++ b/tests/regressiontests/admin_util/models.py @@ -18,6 +18,10 @@ class Article(models.Model): class Count(models.Model): num = models.PositiveSmallIntegerField() + parent = models.ForeignKey('self', null=True) + + def __unicode__(self): + return unicode(self.num) class Event(models.Model): date = models.DateTimeField(auto_now_add=True) diff --git a/tests/regressiontests/admin_util/tests.py b/tests/regressiontests/admin_util/tests.py index 518a16d151..24dde41c25 100644 --- a/tests/regressiontests/admin_util/tests.py +++ b/tests/regressiontests/admin_util/tests.py @@ -6,7 +6,7 @@ from django.contrib.admin.util import display_for_field, label_for_field, lookup from django.contrib.admin.util import NestedObjects from django.contrib.admin.views.main import EMPTY_CHANGELIST_VALUE from django.contrib.sites.models import Site -from django.db import models +from django.db import models, DEFAULT_DB_ALIAS from django.test import TestCase from django.utils import unittest from django.utils.formats import localize @@ -20,51 +20,50 @@ class NestedObjectsTests(TestCase): """ def setUp(self): - self.n = NestedObjects() + self.n = NestedObjects(using=DEFAULT_DB_ALIAS) self.objs = [Count.objects.create(num=i) for i in range(5)] def _check(self, target): self.assertEquals(self.n.nested(lambda obj: obj.num), target) - def _add(self, obj, parent=None): - # don't bother providing the extra args that NestedObjects ignores - self.n.add(None, None, obj, None, parent) + def _connect(self, i, j): + self.objs[i].parent = self.objs[j] + self.objs[i].save() - def test_unrelated_roots(self): - self._add(self.objs[0]) - self._add(self.objs[1]) - self._add(self.objs[2], self.objs[1]) + def _collect(self, *indices): + self.n.collect([self.objs[i] for i in indices]) + def test_unrelated_roots(self): + self._connect(2, 1) + self._collect(0) + self._collect(1) self._check([0, 1, [2]]) def test_siblings(self): - self._add(self.objs[0]) - self._add(self.objs[1], self.objs[0]) - self._add(self.objs[2], self.objs[0]) - + self._connect(1, 0) + self._connect(2, 0) + self._collect(0) self._check([0, [1, 2]]) - def test_duplicate_instances(self): - self._add(self.objs[0]) - self._add(self.objs[1]) - dupe = Count.objects.get(num=1) - self._add(dupe, self.objs[0]) - - self._check([0, 1]) - def test_non_added_parent(self): - self._add(self.objs[0], self.objs[1]) - + self._connect(0, 1) + self._collect(0) self._check([0]) def test_cyclic(self): - self._add(self.objs[0], self.objs[2]) - self._add(self.objs[1], self.objs[0]) - self._add(self.objs[2], self.objs[1]) - self._add(self.objs[0], self.objs[2]) - + self._connect(0, 2) + self._connect(1, 0) + self._connect(2, 1) + self._collect(0) self._check([0, [1, [2]]]) + def test_queries(self): + self._connect(1, 0) + self._connect(2, 0) + # 1 query to fetch all children of 0 (1 and 2) + # 1 query to fetch all children of 1 and 2 (none) + # Should not require additional queries to populate the nested graph. + self.assertNumQueries(2, self._collect, 0) class UtilTests(unittest.TestCase): def test_values_from_lookup_field(self): |
