Context Navigation

Back to Ticket #7539

Ticket #7539: on_delete_on_update-r11733.diff

File on_delete_on_update-r11733.diff, 45.7 KB (added by glassfordm, 15 years ago)
Same as my previous patch, but including unintentionally omitted unit tests

django/db/models/base.py

 from django.db.models.fields import AutoField, FieldDoesNotExist
 from django.db.models.fields.related import OneToOneRel, ManyToOneRel, OneToOneField
 from django.db.models.query import delete_objects, Q
 from django.db.models.query_utils import CollectedObjects, DeferredAttribute
+from django.db.models.query_utils import CollectedFields, CollectedObjects, DeferredAttribute
 from django.db.models.options import Options
 from django.db import connection, transaction, DatabaseError
+from django.db import connection, transaction, DatabaseError, IntegrityError
 from django.db.models import signals
+from django.db.models.fields.related import CASCADE, PROTECT, SET_NULL, SET_DEFAULT
 from django.db.models.loading import register_models, get_model
 from django.utils.functional import curry
 from django.utils.encoding import smart_str, force_unicode, smart_unicode
 …
     save_base.alters_data = True
     def _collect_sub_objects(self, seen_objs, parent=None, nullable=False):
+    def _collect_sub_objects(self, seen_objs, fields_to_set, parent=None, nullable=False):
         """
         Recursively populates seen_objs with all objects related to this
         object.
 …
         pk_val = self._get_pk_val()
         if seen_objs.add(self.__class__, pk_val, self, parent, nullable):
             return
+        def _handle_sub_obj(related, sub_obj):
+            on_delete = related.field.rel.on_delete
+            if on_delete is None:
+                #If no explicit on_delete option is specified, use the old
+                #django behavior as the default: SET_NULL if the foreign
+                #key is nullable, otherwise CASCADE.
+                if related.field.null:
+                    on_delete = SET_NULL
+                else:
+                    on_delete = CASCADE
+            if on_delete == CASCADE:
+                sub_obj._collect_sub_objects(seen_objs, fields_to_set, self.__class__)
+            elif on_delete == PROTECT:
+                msg = '[Django] Cannot delete a parent object: a foreign key constraint fails (ForeignKey `%s` (pk=`%s`) references `%s` (pk=`%s`))' % (
+                    sub_obj.__class__,
+                    sub_obj._get_pk_val(),
+                    self.__class__,
+                    pk_val,
+                    )
+                raise IntegrityError(msg)
+            elif on_delete == SET_NULL:
+                if not related.field.null:
+                    msg = '[Django] Cannot delete a parent object: foreign key constraint on_delete=SET_NULL is specified for a non-nullable foreign key (ForeignKey `%s` (pk=`%s`) references `%s` (pk=`%s`))' % (
+                        sub_obj.__class__,
+                        sub_obj._get_pk_val(),
+                        self.__class__,
+                        pk_val,
+                        )
+                    raise IntegrityError(msg)
+                fields_to_set.add(sub_obj.__class__, sub_obj._get_pk_val(), sub_obj, related.field.name, None)
+            elif on_delete == SET_DEFAULT:
+                if not related.field.has_default():
+                    msg = '[Django] Cannot delete a parent object: foreign key constraint on_delete=SET_DEFAULT is specified for a foreign key with no default value (ForeignKey `%s` (pk=`%s`) references `%s` (pk=`%s`))' % (
+                        sub_obj.__class__,
+                        sub_obj._get_pk_val(),
+                        self.__class__,
+                        pk_val,
+                        )
+                    raise IntegrityError(msg)
+                fields_to_set.add(sub_obj.__class__, sub_obj._get_pk_val(), sub_obj, related.field.name, related.field.get_default())
+            else:
+                raise AttributeError('Unexpected value for on_delete')
         for related in self._meta.get_all_related_objects():
             rel_opts_name = related.get_accessor_name()
             if isinstance(related.field.rel, OneToOneRel):
                 try:
+                    # delattr(self, rel_opts_name) #Delete first to clear any stale cache
+                        #TODO: the above line is a bit of a hack
+                        #It's one way (not a very good one) to work around stale cache data causing
+                        #spurious RESTRICT errors, etc; it would be better to prevent the cache from
+                        #becoming stale in the first place.
                     sub_obj = getattr(self, rel_opts_name)
                 except ObjectDoesNotExist:
                     pass
                 else:
                     sub_obj._collect_sub_objects(seen_objs, self.__class__, related.field.null)
+                    _handle_sub_obj(related, sub_obj)
             else:
                 # To make sure we can access all elements, we can't use the
                 # normal manager on the related object. So we work directly
 …
                         continue
                 delete_qs = rel_descriptor.delete_manager(self).all()
                 for sub_obj in delete_qs:
                     sub_obj._collect_sub_objects(seen_objs, self.__class__, related.field.null)
+                    _handle_sub_obj(related, sub_obj)
         # Handle any ancestors (for the model-inheritance case). We do this by
         # traversing to the most remote parent classes -- those with no parents
 …
                 continue
             # At this point, parent_obj is base class (no ancestor models). So
             # delete it and all its descendents.
             parent_obj._collect_sub_objects(seen_objs)
+            parent_obj._collect_sub_objects(seen_objs, fields_to_set)
     def delete(self):
         assert self._get_pk_val() is not None, "%s object can't be deleted because its %s attribute is set to None." % (self._meta.object_name, self._meta.pk.attname)
         # Find all the objects than need to be deleted.
         seen_objs = CollectedObjects()
+        self._collect_sub_objects(seen_objs)
+        fields_to_set = CollectedFields()
+        self._collect_sub_objects(seen_objs, fields_to_set)
         # Actually delete the objects.
+        delete_objects(seen_objs)
+        delete_objects(seen_objs, fields_to_set)
     delete.alters_data = True
     def _get_FIELD_display(self, field):

django/db/models/fields/related.py

 pending_lookups = {}
+class CASCADE(object):
+    pass
+class PROTECT(object):
+    pass
+class SET_NULL(object):
+    pass
+class SET_DEFAULT(object):
+    pass
+ALLOWED_ON_DELETE_ACTION_TYPES = set([None, CASCADE, PROTECT, SET_NULL, SET_DEFAULT])
 def add_lazy_relation(cls, field, relation, operation):
     """
     Adds a lookup on ``cls`` when a related field is defined using a string,
 …
         # object you just set.
         setattr(instance, self.cache_name, value)
         setattr(value, self.related.field.get_cache_name(), instance)
+    #TODO: the following function is a bit of a hack
+        #It's one way (not a very good one) to work around stale cache data causing
+        #spurious RESTRICT errors, etc; it would be better to prevent the cache from
+        #becoming stale in the first place.
+    # def __delete__(self, instance):
+    #     try:
+    #         return delattr(instance, self.cache_name)
+    #     except AttributeError:
+    #         pass
 class ReverseSingleRelatedObjectDescriptor(object):
     # This class provides the functionality that makes the related-object
 …
 class ManyToOneRel(object):
     def __init__(self, to, field_name, related_name=None,
+            limit_choices_to=None, lookup_overrides=None, parent_link=False):
+            limit_choices_to=None, lookup_overrides=None, parent_link=False,
+            on_delete=None):
         try:
             to._meta
         except AttributeError: # to._meta doesn't exist, so it must be RECURSIVE_RELATIONSHIP_CONSTANT
 …
         self.lookup_overrides = lookup_overrides or {}
         self.multiple = True
         self.parent_link = parent_link
+        self.on_delete = on_delete
     def is_hidden(self):
         "Should the related object be hidden?"
 …
 class OneToOneRel(ManyToOneRel):
     def __init__(self, to, field_name, related_name=None,
+            limit_choices_to=None, lookup_overrides=None, parent_link=False):
+            limit_choices_to=None, lookup_overrides=None, parent_link=False,
+            on_delete=None):
         super(OneToOneRel, self).__init__(to, field_name,
                 related_name=related_name, limit_choices_to=limit_choices_to,
+                lookup_overrides=lookup_overrides, parent_link=parent_link)
+                lookup_overrides=lookup_overrides, parent_link=parent_link,
+                on_delete=on_delete)
         self.multiple = False
 class ManyToManyRel(object):
 …
             related_name=kwargs.pop('related_name', None),
             limit_choices_to=kwargs.pop('limit_choices_to', None),
             lookup_overrides=kwargs.pop('lookup_overrides', None),
+            parent_link=kwargs.pop('parent_link', False))
+            parent_link=kwargs.pop('parent_link', False),
+            on_delete=kwargs.pop('on_delete', None))
         Field.__init__(self, **kwargs)
         self.db_index = True
 …
             target = self.rel.to._meta.db_table
         cls._meta.duplicate_targets[self.column] = (target, "o2m")
+        on_delete = self.rel.on_delete
+        if on_delete not in ALLOWED_ON_DELETE_ACTION_TYPES:
+            raise ValueError("Invalid value 'on_delete=%s' specified for %s %s.%s." % (on_delete, type(self).__name__, cls.__name__, name))
+        if on_delete == SET_NULL and not self.null:
+            specification = "'on_delete=SET_NULL'"
+            raise ValueError("%s specified for %s '%s.%s', but the field is not nullable." % (specification, type(self).__name__, cls.__name__, name))
+        if on_delete == SET_DEFAULT and not self.has_default():
+            specification = "'on_delete=SET_DEFAULT'"
+            raise ValueError("%s specified for %s '%s.%s', but the field has no default value." % (specification, type(self).__name__, cls.__name__, name))
     def contribute_to_related_class(self, cls, related):
         # Internal FK's - i.e., those with a related name ending with '+' -
         # don't get a related descriptor.

django/db/models/init.py

 from django.db.models.fields.subclassing import SubfieldBase
 from django.db.models.fields.files import FileField, ImageField
 from django.db.models.fields.related import ForeignKey, OneToOneField, ManyToManyField, ManyToOneRel, ManyToManyRel, OneToOneRel
+from django.db.models.fields.related import CASCADE, PROTECT, SET_NULL, SET_DEFAULT
 from django.db.models import signals
 # Admin stages.

django/db/models/query.py

 from copy import deepcopy
 from django.db import connection, transaction, IntegrityError
 from django.db.models.aggregates import Aggregate
+from django.db.models.sql.constants import GET_ITERATOR_CHUNK_SIZE
 from django.db.models.fields import DateField
 from django.db.models.query_utils import Q, select_related_descend, CollectedObjects, CyclicDependency, deferred_class_factory
+from django.db.models.query_utils import Q, select_related_descend, CollectedFields, CollectedObjects, CyclicDependency, deferred_class_factory
 from django.db.models import signals, sql
 # Used to control how many objects are worked with at once in some cases (e.g.
 …
             # Collect all the objects to be deleted in this chunk, and all the
             # objects that are related to the objects that are to be deleted.
             seen_objs = CollectedObjects(seen_objs)
+            fields_to_set = CollectedFields()
             for object in del_query[:CHUNK_SIZE]:
                 object._collect_sub_objects(seen_objs)
+                object._collect_sub_objects(seen_objs, fields_to_set)
             if not seen_objs:
                 break
             delete_objects(seen_objs)
+            delete_objects(seen_objs, fields_to_set)
         # Clear the result cache, in case this QuerySet gets reused.
         self._result_cache = None
 …
                 setattr(obj, f.get_cache_name(), rel_obj)
     return obj, index_end
 def delete_objects(seen_objs):
+def delete_objects(seen_objs, fields_to_set):
     """
     Iterate through a list of seen classes, and remove any instances that are
     referred to.
 …
     obj_pairs = {}
     try:
+        for cls, cls_dct in fields_to_set.iteritems():
+            #TODO: batch these, similar to UpdateQuery.clear_related?
+            #(Note that it may be harder to do here because the default value
+            #for a given field may be different for each instance,
+            #while UpdateQuery.clear_related always uses the value None).
+            query = sql.UpdateQuery(cls, connection)
+            for instance, field_names_and_values in cls_dct.itervalues():
+                query.where = query.where_class()
+                pk = query.model._meta.pk
+                query.where.add((sql.where.Constraint(None, pk.column, pk), 'exact', instance.pk), sql.where.AND)
+                query.add_update_values(field_names_and_values)
+                query.execute_sql()
         for cls in ordered_classes:
             items = seen_objs[cls].items()
             items.sort()
 …
                 if not cls._meta.auto_created:
                     signals.pre_delete.send(sender=cls, instance=instance)
+            # Handle related GenericRelation and ManyToManyField instances
             pk_list = [pk for pk,instance in items]
             del_query = sql.DeleteQuery(cls, connection)
             del_query.delete_batch_related(pk_list)
-            update_query = sql.UpdateQuery(cls, connection)
-            for field, model in cls._meta.get_fields_with_model():
-                if (field.rel and field.null and field.rel.to in seen_objs and
-                        filter(lambda f: f.column == field.rel.get_related_field().column,
-                        field.rel.to._meta.fields)):
-                    if model:
-                        sql.UpdateQuery(model, connection).clear_related(field,
-                                pk_list)
-                    else:
-                        update_query.clear_related(field, pk_list)
-        # Now delete the actual data.
         for cls in ordered_classes:
             items = obj_pairs[cls]
             items.reverse()
             pk_list = [pk for pk,instance in items]
             del_query = sql.DeleteQuery(cls, connection)
             del_query.delete_batch(pk_list)
+            # Last cleanup; set NULLs where there once was a reference to the
+            # object, NULL the primary key of the found objects, and perform
+            # post-notification.
+        #Last cleanup; set NULLs and default values where there once was a
+        #reference to the object, NULL the primary key of the found objects,
+        #and perform post-notification.
+        for cls, cls_dct in fields_to_set.iteritems():
+            for instance, field_names_and_values in cls_dct.itervalues():
+                for field_name, field_value in field_names_and_values.iteritems():
+                    field = cls._meta.get_field_by_name(field_name)[0]
+                    setattr(instance, field.attname, field_value)
+        for cls in ordered_classes:
+            items = obj_pairs[cls]
+            items.reverse()
             for pk_val, instance in items:
                 for field in cls._meta.fields:
                     if field.rel and field.null and field.rel.to in seen_objs:

django/db/models/query_utils.py

         """
         return self.data.keys()
+class CollectedFields(object):
+    """
+    A container that stores the model object and field name
+    for fields that need to  be set to enforce on_delete=SET_NULL
+    and on_delete=SET_DEFAULT ForeigKey constraints.
+    """
+    def __init__(self):
+        self.data = {}
+    def add(self, model, pk, obj, field_name, field_value):
+        """
+        Adds an item.
+        model is the class of the object being added,
+        pk is the primary key, obj is the object itself,
+        field_name is the name of the field to be set,
+        field_value is the value it needs to be set to.
+        """
+        d = self.data.setdefault(model, SortedDict())
+        obj, field_names_and_values = d.setdefault(pk, (obj, dict()))
+        assert field_name not in field_names_and_values or field_names_and_values[field_name] == field_value
+        field_names_and_values[field_name] = field_value
+    def __contains__(self, key):
+        return self.data.__contains__(key)
+    def __getitem__(self, key):
+        return self.data[key]
+    def __nonzero__(self):
+        return bool(self.data)
+    def iteritems(self):
+        return self.data.iteritems()
+    def iterkeys(self):
+        return self.data.iterkeys()
+    def itervalues(self):
+        return self.data.itervalues()
+    def items(self):
+        return self.data.items()
+    def keys(self):
+        return self.data.keys()
+    def values(self):
+        return self.data.values()
 class QueryWrapper(object):
     """
     A type that indicates the contents are an SQL fragment and the associate

tests/modeltests/on_delete_django/tests.py

Property changes on: tests/modeltests/on_delete_django/__init__.py
___________________________________________________________________
Name: svn:keywords
   + Id

+from django.conf import settings
+from django.db import IntegrityError, models
+from django.db.models.fields.related import CASCADE, PROTECT, SET_NULL, SET_DEFAULT
+from django.test import TestCase
+from models import *
+class ON_DELETE_Tests(TestCase):
+    ### ForeignKey tests
+    def test_ForeignKey_CASCADE(self):
+        self._CASCADE_test(ForeignKey_CASCADE, True)
+    def test_ForeignKey_PROTECT(self):
+        self._PROTECT_test(ForeignKey_PROTECT, True)
+    def test_ForeignKey_SET_NULL(self):
+        self._SET_NULL_test(ForeignKey_SET_NULL, True)
+    def test_ForeignKey_SET_DEFAULT(self):
+        self._SET_DEFAULT_test(ForeignKey_SET_DEFAULT, True)
+    def test_ForeignKey_None_Null_True(self):
+        #If no on_delete behavior is specified,
+        #Django treats it as SET_NULL if the field is nullable
+        self._SET_NULL_test(ForeignKey_None_Null_True, True)
+    def test_ForeignKey_None_Null_False(self):
+        #If no on_delete behavior is specified,
+        #Django treats it as CASCADE if it is not nullable.
+        self._CASCADE_test(ForeignKey_None_Null_False, True)
+    def test_ForeignKey_Combined(self):
+        self._combined_test(ForeignKey_CASCADE, ForeignKey_PROTECT, ForeignKey_SET_NULL, ForeignKey_SET_DEFAULT)
+    def test_ForeignKey_SET_NULL_requires_nullable_field(self):
+        try:
+            class Test(models.Model):
+                fk = models.ForeignKey(Data, to_field='data', on_delete=models.SET_NULL)
+        except ValueError, e:
+            self.assertEqual(str(e), "'on_delete=SET_NULL' specified for ForeignKey 'Test.fk', but the field is not nullable.")
+        else:
+            self.assertTrue(False, 'Expected exception not raised.')
+    def test_ForeignKey_SET_DEFAULT_requires_default_value(self):
+        try:
+            class Test(models.Model):
+                fk = models.ForeignKey(Data, to_field='data', on_delete=models.SET_DEFAULT)
+        except ValueError, e:
+            self.assertEqual(str(e), "'on_delete=SET_DEFAULT' specified for ForeignKey 'Test.fk', but the field has no default value.")
+        else:
+            self.assertTrue(False, 'Expected exception not raised.')
+    ### OneToOneField tests
+    def test_OneToOne_CASCADE(self):
+        self._CASCADE_test(OneToOne_CASCADE, False)
+    def test_OneToOne_PROTECT(self):
+        self._PROTECT_test(OneToOne_PROTECT, False)
+    def test_OneToOne_SET_NULL(self):
+        self._SET_NULL_test(OneToOne_SET_NULL, False)
+    def test_OneToOne_SET_DEFAULT(self):
+        self._SET_DEFAULT_test(OneToOne_SET_DEFAULT, False)
+    def test_OneToOne_None_Null_True(self):
+        #If no on_delete behavior is specified,
+        #Django treats it as SET_NULL if the field is nullable
+        self._SET_NULL_test(OneToOne_None_Null_True, False)
+    def test_OneToOne_None_Null_False(self):
+        #If no on_delete behavior is specified,
+        #Django treats it as CASCADE if it is not nullable.
+        self._CASCADE_test(OneToOne_None_Null_False, False)
+    def test_OneToOne_Combined(self):
+        self._combined_test(OneToOne_CASCADE, OneToOne_PROTECT, OneToOne_SET_NULL, OneToOne_SET_DEFAULT)
+    def test_OneToOne_SET_NULL_requires_nullable_field(self):
+        try:
+            class Test(models.Model):
+                fk = models.OneToOneField(Data, to_field='data', on_delete=models.SET_NULL)
+        except ValueError, e:
+            self.assertEqual(str(e), "'on_delete=SET_NULL' specified for OneToOneField 'Test.fk', but the field is not nullable.")
+        else:
+            self.assertTrue(False, 'Expected exception not raised.')
+    def test_OneToOne_SET_DEFAULT_requires_default_value(self):
+        try:
+            class Test(models.Model):
+                fk = models.OneToOneField(Data, to_field='data', on_delete=models.SET_DEFAULT)
+        except ValueError, e:
+            self.assertEqual(str(e), "'on_delete=SET_DEFAULT' specified for OneToOneField 'Test.fk', but the field has no default value.")
+        else:
+            self.assertTrue(False, 'Expected exception not raised.')
+    ### Multiple FK tests
+    def test_All_FK_Model_1(self):
+        data = Data.objects.create(data=1)
+        fk = All_FK_Model.objects.create(
+            fk_ForeignKey_Cascade=data,
+            fk_ForeignKey_Protect=data,
+            fk_ForeignKey_Set_Null=data,
+            fk_ForeignKey_Set_Default=data,
+            fk_OneToOneField_Cascade=data,
+            fk_OneToOneField_Protect=data,
+            fk_OneToOneField_Set_Null=data,
+            fk_OneToOneField_Set_Default=data,
+            )
+        fk_id = fk.pk
+        self.assertEqual(data, fk.fk_ForeignKey_Cascade)
+        self.assertEqual(data, fk.fk_ForeignKey_Protect)
+        self.assertEqual(data, fk.fk_ForeignKey_Set_Null)
+        self.assertEqual(data, fk.fk_ForeignKey_Set_Default)
+        self.assertEqual(data, fk.fk_OneToOneField_Cascade)
+        self.assertEqual(data, fk.fk_OneToOneField_Protect)
+        self.assertEqual(data, fk.fk_OneToOneField_Set_Null)
+        self.assertEqual(data, fk.fk_OneToOneField_Set_Default)
+        self.assertRaises(IntegrityError, data.delete)
+        fk.fk_ForeignKey_Protect = None
+        fk.fk_OneToOneField_Protect = data
+        fk.save()
+        self.assertRaises(IntegrityError, data.delete)
+        fk.fk_ForeignKey_Protect = data
+        fk.fk_OneToOneField_Protect = None
+        fk.save()
+        self.assertRaises(IntegrityError, data.delete)
+        fk.fk_ForeignKey_Protect = None
+        fk.fk_OneToOneField_Protect = None
+        fk.save()
+        data.delete()
+        self.assertRaises(All_FK_Model.DoesNotExist, All_FK_Model.objects.get, id=fk_id)
+    def test_All_FK_Model_2(self):
+        default_value = default_data()
+        data = Data.objects.create(data=1)
+        fk = All_FK_Model.objects.create(
+            fk_ForeignKey_Cascade=None,
+            fk_ForeignKey_Protect=None,
+            fk_ForeignKey_Set_Null=data,
+            fk_ForeignKey_Set_Default=data,
+            fk_OneToOneField_Cascade=None,
+            fk_OneToOneField_Protect=None,
+            fk_OneToOneField_Set_Null=data,
+            fk_OneToOneField_Set_Default=data,
+            )
+        fk_id = fk.pk
+        self.assertEqual(None, fk.fk_ForeignKey_Cascade)
+        self.assertEqual(None, fk.fk_ForeignKey_Protect)
+        self.assertEqual(data, fk.fk_ForeignKey_Set_Null)
+        self.assertEqual(data, fk.fk_ForeignKey_Set_Default)
+        self.assertEqual(None, fk.fk_OneToOneField_Cascade)
+        self.assertEqual(None, fk.fk_OneToOneField_Protect)
+        self.assertEqual(data, fk.fk_OneToOneField_Set_Null)
+        self.assertEqual(data, fk.fk_OneToOneField_Set_Default)
+        data.delete()
+        fk = All_FK_Model.objects.get(pk=fk_id) #TODO: fix bug that makes this necessary
+        self.assertEqual(None, fk.fk_ForeignKey_Cascade)
+        self.assertEqual(None, fk.fk_ForeignKey_Protect)
+        self.assertEqual(None, fk.fk_ForeignKey_Set_Null)
+        self.assertEqual(default_value, fk.fk_ForeignKey_Set_Default)
+        self.assertEqual(None, fk.fk_OneToOneField_Cascade)
+        self.assertEqual(None, fk.fk_OneToOneField_Protect)
+        self.assertEqual(None, fk.fk_OneToOneField_Set_Null)
+        self.assertEqual(default_value, fk.fk_OneToOneField_Set_Default)
+    #TODO: not done yet:
+    # def test_All_FK_Model_3(self):
+    #     default_value = default_data()
+    #
+    #     data = Data.objects.create(data=1)
+    #     fk1 = All_FK_Model.objects.create(
+    #         fk_ForeignKey_Cascade=None,
+    #         fk_ForeignKey_Protect=None,
+    #         fk_ForeignKey_Set_Null=data,
+    #         fk_ForeignKey_Set_Default=data,
+    #         fk_OneToOneField_Cascade=None,
+    #         fk_OneToOneField_Protect=None,
+    #         fk_OneToOneField_Set_Null=data,
+    #         fk_OneToOneField_Set_Default=data,
+    #         )
+    #     fk2 = All_FK_Model.objects.create(
+    #         fk_ForeignKey_Cascade=None,
+    #         fk_ForeignKey_Protect=None,
+    #         fk_ForeignKey_Set_Null=data,
+    #         fk_ForeignKey_Set_Default=data,
+    #         fk_OneToOneField_Cascade=None,
+    #         fk_OneToOneField_Protect=None,
+    #         fk_OneToOneField_Set_Null=None,
+    #         fk_OneToOneField_Set_Default=None,
+    #         )
+    #     fk1_id = fk1.pk
+    #     fk2_id = fk2.pk
+    #     self.assertEqual(None, fk1.fk_ForeignKey_Cascade)
+    #     self.assertEqual(None, fk1.fk_ForeignKey_Protect)
+    #     self.assertEqual(data, fk1.fk_ForeignKey_Set_Null)
+    #     self.assertEqual(data, fk1.fk_ForeignKey_Set_Default)
+    #     self.assertEqual(None, fk1.fk_OneToOneField_Cascade)
+    #     self.assertEqual(None, fk1.fk_OneToOneField_Protect)
+    #     self.assertEqual(data, fk1.fk_OneToOneField_Set_Null)
+    #     self.assertEqual(data, fk1.fk_OneToOneField_Set_Default)
+    #     self.assertEqual(None, fk2.fk_ForeignKey_Cascade)
+    #     self.assertEqual(None, fk2.fk_ForeignKey_Protect)
+    #     self.assertEqual(data, fk2.fk_ForeignKey_Set_Null)
+    #     self.assertEqual(data, fk2.fk_ForeignKey_Set_Default)
+    #     self.assertEqual(None, fk2.fk_OneToOneField_Cascade)
+    #     self.assertEqual(None, fk2.fk_OneToOneField_Protect)
+    #     self.assertEqual(None, fk2.fk_OneToOneField_Set_Null)
+    #     self.assertEqual(None, fk2.fk_OneToOneField_Set_Default)
+    #
+    #     data.delete()
+    #
+    #     fk1 = All_FK_Model.objects.get(pk=fk1_id) #TODO: fix bug that makes this necessary
+    #     fk2 = All_FK_Model.objects.get(pk=fk2_id) #TODO: fix bug that makes this necessary
+    #     self.assertEqual(None, fk1.fk_ForeignKey_Cascade)
+    #     self.assertEqual(None, fk1.fk_ForeignKey_Protect)
+    #     self.assertEqual(None, fk1.fk_ForeignKey_Set_Null)
+    #     self.assertEqual(default_value, fk1.fk_ForeignKey_Set_Default)
+    #     self.assertEqual(None, fk1.fk_OneToOneField_Cascade)
+    #     self.assertEqual(None, fk1.fk_OneToOneField_Protect)
+    #     self.assertEqual(None, fk1.fk_OneToOneField_Set_Null)
+    #     self.assertEqual(default_value, fk1.fk_OneToOneField_Set_Default)
+    #     self.assertEqual(None, fk2.fk_ForeignKey_Cascade)
+    #     self.assertEqual(None, fk2.fk_ForeignKey_Protect)
+    #     self.assertEqual(None, fk2.fk_ForeignKey_Set_Null)
+    #     self.assertEqual(default_value, fk2.fk_ForeignKey_Set_Default)
+    #     self.assertEqual(None, fk2.fk_OneToOneField_Cascade)
+    #     self.assertEqual(None, fk2.fk_OneToOneField_Protect)
+    #     self.assertEqual(None, fk2.fk_OneToOneField_Set_Null)
+    #     self.assertEqual(default_value, fk2.fk_OneToOneField_Set_Default)
+    ### Cache tests
+    #These tests illustrate a caching problem with OneToOneKeyField.
+    def test_issue_1a(self):
+        data = Data.objects.create(data=1)
+        fk = OneToOne_PROTECT.objects.create(fk=data)
+        fk.delete()
+        data.delete() #This succeeds
+    def test_issue_1b(self):
+        data = Data.objects.create(data=1)
+        fk = OneToOne_PROTECT.objects.create(fk=data)
+        self.assertRaises(IntegrityError, data.delete) #This line added
+            #This loads the item into the SingleRelatedObjectDescriptor cache;
+            #this cached item is not unloaded after the fk.delete(),
+            #which later causes the data.delete() to fail.
+        fk.delete()
+        data.delete() #This fails
+    def test_issue_1c(self):
+        data = Data.objects.create(data=1)
+        fk = ForeignKey_PROTECT.objects.create(fk=data) #This line changed
+            #This problem applies only to OneToOneField;
+            #ForeignKey doesn't have the same problem
+        self.assertRaises(IntegrityError, data.delete)
+        fk.delete()
+        data.delete() #This succeeds
+    def test_issue_1d(self):
+        data = Data.objects.create(data=1)
+        fk = OneToOne_PROTECT.objects.create(fk=data)
+        self.assertRaises(IntegrityError, data.delete)
+        fk = OneToOne_PROTECT.objects.get(pk=fk.pk)
+        fk.delete()
+        data.delete() #This fails
+    def test_issue_1e(self):
+        data = Data.objects.create(data=1)
+        fk = OneToOne_PROTECT.objects.create(fk=data)
+        self.assertRaises(IntegrityError, data.delete)
+        fk.delete()
+        data = Data.objects.get(pk=data.pk)
+        data.delete() #This succeeds
+            #Refreshing the data object unloads/refreshes its cache
+    #These tests illustrate a problem with both ForeignKey and OneToOneKeyField.
+    #The database is changed correctly, but instances in memory are not updated correctly.
+    #The problem is that, although some instances are changed in memory, other instances
+    #referring to the same database record are not, and these become out of sync
+    #with the db data. If these out of sync instances are then used, issues occur.
+    def test_issue_2a(self):
+        default_value = default_data()
+        data = Data.objects.create(data=1)
+        fk = ForeignKey_SET_DEFAULT.objects.create(fk=data)
+        data.delete()
+        self.assertEqual(default_value, fk.fk) #Fails
+    def test_issue_2b(self):
+        data = Data.objects.create(data=1)
+        fk = ForeignKey_SET_NULL.objects.create(fk=data)
+        data.delete()
+        self.assertEqual(None, fk.fk) #Fails
+    def test_issue_2c(self):
+        data = Data.objects.create(data=1)
+        fk = ForeignKey_SET_NULL.objects.create(fk=data)
+        data.delete()
+        fk = ForeignKey_SET_NULL.objects.get(pk=fk.pk) #Added this line
+        self.assertEqual(None, fk.fk) #Succeeds
+    def test_issue_2d(self):
+        default_value = default_data()
+        data = Data.objects.create(data=1)
+        fk = OneToOne_SET_DEFAULT.objects.create(fk=data)
+        data.delete()
+        self.assertEqual(default_value, fk.fk) #Fails
+    def test_issue_2e(self):
+        data = Data.objects.create(data=1)
+        fk = OneToOne_SET_NULL.objects.create(fk=data)
+        data.delete()
+        self.assertEqual(None, fk.fk) #Fails
+    def test_issue_2f(self):
+        data = Data.objects.create(data=1)
+        fk = OneToOne_SET_NULL.objects.create(fk=data)
+        data.delete()
+        fk = OneToOne_SET_NULL.objects.get(pk=fk.pk) #Added this line
+        self.assertEqual(None, fk.fk) #Succeeds
+    ### Utility methods
+    def _CASCADE_test(self, foreign_key_model, many_to_one):
+        #Create a data item and foreign key(s) to it
+        data, fk_id_1 = self._create_data_item_and_foreign_key(1, foreign_key_model)
+        if many_to_one:
+            _, fk_id_2 = self._create_data_item_and_foreign_key(1, foreign_key_model)
+        #Create an unrelated data item and foreign key
+        _, fk_id_unlrelated = self._create_data_item_and_foreign_key(2, foreign_key_model)
+        #Delete the data item and make sure the foreign key is affected appropriately
+        data.delete()
+        self.assertEqual(None, data.pk)
+        self.assertRaises(foreign_key_model.DoesNotExist, foreign_key_model.objects.get, id=fk_id_1)
+        if many_to_one:
+            self.assertRaises(foreign_key_model.DoesNotExist, foreign_key_model.objects.get, id=fk_id_2)
+        #Make sure the unrelated data item and foreign key are unaffected
+        self._test_data_item_and_foreign_key(None, 2, foreign_key_model, fk_id_unlrelated)
+    def _PROTECT_test(self, foreign_key_model, many_to_one):
+        #Create a data item and foreign key(s) to it
+        data, fk_id_1 = self._create_data_item_and_foreign_key(1, foreign_key_model)
+        if many_to_one:
+            _, fk_id_2 = self._create_data_item_and_foreign_key(1, foreign_key_model)
+        #Create an unrelated data item and foreign key
+        data_unrelated, fk_id_unlrelated = self._create_data_item_and_foreign_key(2, foreign_key_model)
+        #Delete the data item and make sure the foreign key is affected appropriately
+        self.assertRaises(IntegrityError, data.delete)
+        self.assertNotEqual(None, data.pk)
+        self._test_data_item_and_foreign_key(data, 1, foreign_key_model, fk_id_1)
+        if many_to_one:
+            self._test_data_item_and_foreign_key(None, 1, foreign_key_model, fk_id_2)
+        #Make sure the unrelated data item and foreign key are unaffected
+        self._test_data_item_and_foreign_key(data_unrelated, 2, foreign_key_model, fk_id_unlrelated)
+    def _SET_NULL_test(self, foreign_key_model, many_to_one):
+        #Create a data item and foreign key(s) to it
+        data, fk_id_1 = self._create_data_item_and_foreign_key(1, foreign_key_model)
+        if many_to_one:
+            _, fk_id_2 = self._create_data_item_and_foreign_key(1, foreign_key_model)
+        #Create an unrelated data item and foreign key
+        data_unrelated, fk_id_unlrelated = self._create_data_item_and_foreign_key(2, foreign_key_model)
+        #Delete the data item and make sure the foreign key is affected appropriately
+        data.delete()
+        self.assertEqual(None, data.pk)
+        self.assertEqual(None, foreign_key_model.objects.get(id=fk_id_1).fk)
+        self.assertEqual(None, foreign_key_model.objects.get(id=fk_id_1).fk_id)
+        if many_to_one:
+            self.assertEqual(None, foreign_key_model.objects.get(id=fk_id_2).fk)
+            self.assertEqual(None, foreign_key_model.objects.get(id=fk_id_2).fk_id)
+        #Make sure the unrelated data item and foreign key are unaffected
+        self._test_data_item_and_foreign_key(data_unrelated, 2, foreign_key_model, fk_id_unlrelated)
+    def _SET_DEFAULT_test(self, foreign_key_model, many_to_one):
+        #Create a data item and foreign key(s) to it
+        data, fk_id_1 = self._create_data_item_and_foreign_key(1, foreign_key_model)
+        if many_to_one:
+            _, fk_id_2 = self._create_data_item_and_foreign_key(1, foreign_key_model)
+        #Create an unrelated data item and foreign key
+        data_unrelated, fk_id_unlrelated = self._create_data_item_and_foreign_key(2, foreign_key_model)
+        #Delete the data item and make sure the foreign key is affected appropriately
+        data.delete()
+        self.assertEqual(None, data.pk)
+        default_value = default_data()
+        self.assertEqual(default_value, foreign_key_model.objects.get(id=fk_id_1).fk)
+        if many_to_one:
+            self.assertEqual(default_value, foreign_key_model.objects.get(id=fk_id_2).fk)
+        #Make sure the unrelated data item and foreign key are unaffected
+        self._test_data_item_and_foreign_key(data_unrelated, 2, foreign_key_model, fk_id_unlrelated)
+    def _combined_test(self, cascade_model, protect_model, set_null_model, set_default_model):
+        #Create data items and foreign key
+        data, fk_id_cascade = self._create_data_item_and_foreign_key(1, cascade_model)
+        _, fk_id_protect = self._create_data_item_and_foreign_key(1, protect_model)
+        _, fk_id_set_null = self._create_data_item_and_foreign_key(1, set_null_model)
+        _, fk_id_set_default = self._create_data_item_and_foreign_key(1, set_default_model)
+        data_id = data.pk
+        #Attempt to delete--should do nothing due to PROTECT foreign key
+        self.assertRaises(IntegrityError, data.delete)
+        self.assertNotEqual(None, data.pk)
+        self._test_data_item_and_foreign_key(data, 1, cascade_model, fk_id_cascade)
+        self._test_data_item_and_foreign_key(data, 1, protect_model, fk_id_protect)
+        self._test_data_item_and_foreign_key(data, 1, set_null_model, fk_id_set_null)
+        self._test_data_item_and_foreign_key(data, 1, set_default_model, fk_id_set_default)
+        #Delete protect foreign key and try again--this time, it should work
+        fk = protect_model.objects.get(id=fk_id_protect)
+        fk.delete()
+        data = Data.objects.get(pk=data_id)
+        data.delete()
+        self.assertEqual(None, data.pk)
+        self.assertRaises(cascade_model.DoesNotExist, cascade_model.objects.get, id=fk_id_cascade)
+        self.assertEqual(None, set_null_model.objects.get(id=fk_id_set_null).fk)
+        self.assertEqual(None, set_null_model.objects.get(id=fk_id_set_null).fk_id)
+        self.assertEqual(default_data(), set_default_model.objects.get(id=fk_id_set_default).fk)
+    def _create_data_item_and_foreign_key(self, data_value, fk_model):
+        data, _ = Data.objects.get_or_create(data=data_value)
+        fk = fk_model.objects.create(fk=data)
+        self._test_data_item_and_foreign_key(data, data_value, fk_model, fk.id)
+        return data, fk.id
+    def _test_data_item_and_foreign_key(self, data_item, data_value, fk_model, fk_id):
+        data = Data.objects.get(data=data_value)
+        if data_item:
+            self.assertEqual(data.pk, data_item.pk)
+            self.assertEqual(data.data, data_item.data)
+        fk = fk_model.objects.get(id=fk_id)
+        self.assertEqual(data, fk.fk)
+ No newline at end of file

tests/modeltests/on_delete_django/models.py

Property changes on: tests/modeltests/on_delete_django/tests.py
___________________________________________________________________
Name: svn:keywords
   + Id

+"""
+Test ON DELETE behavior for foreign keys
+when it's handled by the Django.
+"""
+from django.db import models
+### Data model
+class Data(models.Model):
+    data = models.IntegerField(unique=True)
+def default_data():
+    data, _ = Data.objects.get_or_create(data=1000)
+    return data
+### ForeignKey models
+class ForeignKey_CASCADE(models.Model):
+    fk = models.ForeignKey(Data, on_delete=models.CASCADE)
+class ForeignKey_PROTECT(models.Model):
+    fk = models.ForeignKey(Data, on_delete=models.PROTECT)
+class ForeignKey_SET_NULL(models.Model):
+    fk = models.ForeignKey(Data, null=True, on_delete=models.SET_NULL)
+class ForeignKey_SET_DEFAULT(models.Model):
+    fk = models.ForeignKey(Data, default=default_data, on_delete=models.SET_DEFAULT)
+class ForeignKey_None_Null_True(models.Model):
+    fk = models.ForeignKey(Data, null=True, on_delete=None)
+class ForeignKey_None_Null_False(models.Model):
+    fk = models.ForeignKey(Data, on_delete=None)
+### OneToOneField models
+class OneToOne_CASCADE(models.Model):
+    fk = models.OneToOneField(Data, on_delete=models.CASCADE)
+class OneToOne_PROTECT(models.Model):
+    fk = models.OneToOneField(Data, on_delete=models.PROTECT)
+class OneToOne_SET_NULL(models.Model):
+    fk = models.OneToOneField(Data, null=True, on_delete=models.SET_NULL)
+class OneToOne_SET_DEFAULT(models.Model):
+    fk = models.OneToOneField(Data, default=default_data, on_delete=models.SET_DEFAULT)
+class OneToOne_None_Null_True(models.Model):
+    fk = models.OneToOneField(Data, null=True, on_delete=None)
+class OneToOne_None_Null_False(models.Model):
+    fk = models.OneToOneField(Data, on_delete=None)
+### Multiple FK models
+class All_FK_Model(models.Model):
+    fk_ForeignKey_Cascade = models.ForeignKey(Data, null=True, on_delete=models.CASCADE, related_name='fk1')
+    fk_ForeignKey_Protect = models.ForeignKey(Data, null=True, on_delete=models.PROTECT, related_name='fk2')
+    fk_ForeignKey_Set_Null = models.ForeignKey(Data, null=True, on_delete=models.SET_NULL, related_name='fk3')
+    fk_ForeignKey_Set_Default = models.ForeignKey(Data, null=True, default=default_data, on_delete=models.SET_DEFAULT, related_name='fk4')
+    fk_OneToOneField_Cascade = models.OneToOneField(Data, null=True, on_delete=models.CASCADE, related_name='fk5')
+    fk_OneToOneField_Protect = models.OneToOneField(Data, null=True, on_delete=models.PROTECT, related_name='fk6')
+    fk_OneToOneField_Set_Null = models.OneToOneField(Data, null=True, on_delete=models.SET_NULL, related_name='fk7')
+    fk_OneToOneField_Set_Default = models.OneToOneField(Data, null=True, default=default_data, on_delete=models.SET_DEFAULT, related_name='fk8')
+ No newline at end of file

tests/modeltests/delete/models.py

Property changes on: tests/modeltests/on_delete_django/models.py
___________________________________________________________________
Name: svn:keywords
   + Id

 ## First, test the CollectedObjects data structure directly
 >>> from django.db.models.query import CollectedObjects
+>>> from django.db.models.query_utils import CollectedFields, CollectedObjects
 >>> g = CollectedObjects()
 >>> g.add("key1", 1, "item1", None)
 …
 >>> d1 = D(c=c1, a=a1)
 >>> d1.save()
 >>> o = CollectedObjects()
 >>> a1._collect_sub_objects(o)
+>>> o, f = CollectedObjects(), CollectedFields()
+>>> a1._collect_sub_objects(o, f)
 >>> o.keys()
 [<class 'modeltests.delete.models.D'>, <class 'modeltests.delete.models.C'>, <class 'modeltests.delete.models.B'>, <class 'modeltests.delete.models.A'>]
+>>> f.keys()
+[]
 >>> a1.delete()
 # Same again with a known bad order
 …
 >>> d2 = D(c=c2, a=a2)
 >>> d2.save()
 >>> o = CollectedObjects()
 >>> a2._collect_sub_objects(o)
+>>> o, f = CollectedObjects(), CollectedFields()
+>>> a2._collect_sub_objects(o, f)
 >>> o.keys()
 [<class 'modeltests.delete.models.D'>, <class 'modeltests.delete.models.C'>, <class 'modeltests.delete.models.B'>, <class 'modeltests.delete.models.A'>]
+>>> f.keys()
+[]
 >>> a2.delete()
 ### Tests for models E,F - nullable related fields ###
 …
 # Since E.f is nullable, we should delete F first (after nulling out
 # the E.f field), then E.
 >>> o = CollectedObjects()
 >>> e1._collect_sub_objects(o)
+>>> o, f = CollectedObjects(), CollectedFields()
+>>> e1._collect_sub_objects(o, f)
 >>> o.keys()
 [<class 'modeltests.delete.models.F'>, <class 'modeltests.delete.models.E'>]
+>>> f.keys()
+[<class 'modeltests.delete.models.E'>]
-# temporarily replace the UpdateQuery class to verify that E.f is actually nulled out first
->>> import django.db.models.sql
->>> class LoggingUpdateQuery(django.db.models.sql.UpdateQuery):
-...     def clear_related(self, related_field, pk_list):
-...         print "CLEARING FIELD",related_field.name
-...         return super(LoggingUpdateQuery, self).clear_related(related_field, pk_list)
->>> original_class = django.db.models.sql.UpdateQuery
->>> django.db.models.sql.UpdateQuery = LoggingUpdateQuery
 >>> e1.delete()
-CLEARING FIELD f
 >>> e2 = E()
 >>> e2.save()
 …
 # Same deal as before, though we are starting from the other object.
 >>> o = CollectedObjects()
 >>> f2._collect_sub_objects(o)
+>>> o, f = CollectedObjects(), CollectedFields()
+>>> f2._collect_sub_objects(o, f)
 >>> o.keys()
+[<class 'modeltests.delete.models.F'>, <class 'modeltests.delete.models.E'>]
+[<class 'modeltests.delete.models.F'>]
+>>> f.keys()
+[<class 'modeltests.delete.models.E'>]
 >>> f2.delete()
-CLEARING FIELD f
-# Put this back to normal
->>> django.db.models.sql.UpdateQuery = original_class
 """
+}

Download in other formats:

Original Format

Issues

Context Navigation

Ticket #7539: on_delete_on_update-r11733.diff

django/db/models/base.py

django/db/models/fields/related.py

django/db/models/init.py

django/db/models/query.py

django/db/models/query_utils.py

tests/modeltests/on_delete_django/tests.py

tests/modeltests/on_delete_django/models.py

tests/modeltests/delete/models.py

Download in other formats:

Django Links

Learn More

Get Involved

Get Help

Follow Us

Support Us