Ticket #12760: tests.py

from django.db import models
from django.contrib.auth.models import User


# #
# THE SETUP
# #

class Human(models.Model):
  """A Human MAY have a user assigned"""
  name = models.CharField(blank=False,null=False,max_length=100) 
  auth_user = models.ForeignKey(User,blank=True,null=True)
  def __unicode__(self):
      return self.name

class Joke(models.Model):
  """a Joke must have owner """
  text = models.TextField(blank=False,null=False)
  owner = models.ForeignKey(Human,blank=False,null=False)

# #
# THE TEST
# #

from django.test import TestCase
from django.db.models.query_utils import CollectedObjects

class SimpleTest(TestCase):
    
    def setUp(self):
        user = User(username='Authtester',password="secret")
        user.save()

        human = Human(name='HumanTester',auth_user=user)
        human.save()

        joke = Joke(text='phunny Joke....',owner=human)
        joke.save()
        
        self.user ,self.human,self.joke= user,human,joke

    def test1(self):
        """ 
        The deletion of a User should disconnect it from a Human, but should not delete the Human, neither its Joke)
        """
        user, human, joke = self.user, self.human, self.joke
        user.delete()

        self.assertTrue(len(Human.objects.all()), u'there is no human left')
        self.assertTrue(len(Joke.objects.all()), u'there is no Joke left')

    def test2(self):
        """
        inspecting the objects getting collected by calling delete on a user
        """
        user, human, joke = self.user, self.human, self.joke

        seen = CollectedObjects()
        # calling the collection method used in delete
        user._collect_sub_objects(seen)

        collected_instances = []
        for cls,instances in seen.items():
            collected_instances.extend(instances.values())

        self.assertFalse(joke in collected_instances,u'%s should not get collected because the human.user can be set to "NULL".'%joke)
        # _collect_sub_objects keeps collection relations and their subrelations
        # There should be no futher evaluation of sub-relations if a relation can be set to null
        
    def test3(self):
        """
        printing out what happens in the delete_objects funciton
        """
        user, human, joke = self.user, self.human, self.joke

        seen = CollectedObjects()
        user._collect_sub_objects(seen)
        copy_of_delete_objects(seen)


# ############################################################################## #        
# # this is copy of db.models.query.py delete_objects (django version: 1.1.1.)
# # spiced with some ugly prints to see what happens there
# ############################################################################## #
from django.db import connection, transaction
from django.db.models import signals, sql

def copy_of_delete_objects(seen_objs):
    print
    print "."*40
    print "  the delte_object slow-mo"
    print "."*40
    """
    Iterate through a list of seen classes, and remove any instances that are
    referred to.
    """
    if not transaction.is_managed():
        transaction.enter_transaction_management()
        forced_managed = True
    else:
        forced_managed = False
    try:
        ordered_classes = seen_objs.keys()
    except CyclicDependency:
        # If there is a cyclic dependency, we cannot in general delete the
        # objects.  However, if an appropriate transaction is set up, or if the
        # database is lax enough, it will succeed. So for now, we go ahead and
        # try anyway.
        ordered_classes = seen_objs.unordered_keys()

    obj_pairs = {}
    print
    try:
        print "1. LOOP over ordered_classes"
        for cls in ordered_classes:
            print 'Class: ',cls
            items = seen_objs[cls].items()
            items.sort()
            obj_pairs[cls] = items

            # Pre-notify all instances to be deleted.
            for pk_val, instance in items:
                signals.pre_delete.send(sender=cls, instance=instance)

            pk_list = [pk for pk,instance in items]
            print cls.__name__,"Primarykeys used for delete_batch_related: ",tuple(pk_list)
            del_query = sql.DeleteQuery(cls, connection)
            del_query.delete_batch_related(pk_list)

            update_query = sql.UpdateQuery(cls, connection)
            print cls.__name__, 'UPDATE Queries used'
            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)

            print update_query or 'NONE'
        print 
        print "2. LOOP over ordered_classes"
        print "# 'ordered_classes' has not been modified. objectes updated previousely are still in that list"
        # Now delete the actual data.
        for cls in ordered_classes:
            print 'Class: ',cls
            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)
            print '  ',del_query
            # 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.
            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:
                        setattr(instance, field.attname, None)
                        print '  ','the object %s gets its %s set to None'%(instance,field.attname)

                signals.post_delete.send(sender=cls, instance=instance)
                print '  ',"post_delete signal sent"
                setattr(instance, cls._meta.pk.attname, None)
                print '  ','the object %s gets its %s set to None'%(instance,cls._meta.pk.attname)
        if forced_managed:
            transaction.commit()
        else:
            transaction.commit_unless_managed()
    finally:
        if forced_managed:
            transaction.leave_transaction_management()
    print "."*40