Ticket #2227: tests.py

import unittest
from django.db import transaction, connection, models

class Item(models.Model):    
    content = models.TextField()
    
    def __unicode__(self):
        return self.content

class ItemTests(object):
    def setUp(self):
        Item.objects.create(id=1, content="a")
        Item.objects.create(id=2, content="b")

    def tearDown(self):
        Item.objects.all().delete()

    def items_are(self, where, *values):
        ids = [id for (id, value) in values]
        expect = [value for (id, value) in values]
        self.assertEqual([Item.objects.get(id=id).content for id in ids], list(expect))    

class NestedOnSuccess(ItemTests, unittest.TestCase):

    def testSuccess(self):
        """
        This test asserts that an explicit commit done for an inner
        transaction should not actually be committed until the outer
        transaction is finished.
        """
        @transaction.commit_on_success
        def outer():
            item = Item.objects.get(id=1)                   
            inner()
            transaction.rollback()
            item.content = "changed a"
            item.save()
            self.items_are('outer end', (1, "changed a"), (2, "b"))

        @transaction.commit_on_success
        def inner():
            item = Item.objects.get(id=2)
            item.content = "changed b"
            item.save()

        self.items_are('initially', (1, "a"), (2, "b"))
        outer()
        self.items_are('finally', (1, "changed a"), (2, "b"))

class NestedAutocommit(ItemTests, unittest.TestCase):

    def testFail(self):
        """
        This test asserts that an inner `autocommit' transaction
        should not commit itself until the outer transaction is
        finished.  This implies that transaction.managed(False) should
        not override transaction.managed(True).
        """
        @transaction.commit_on_success
        def outer():
            item = Item.objects.get(id=1)
            item.content = inner()
            item.save()
            raise RuntimeError  

        @transaction.autocommit
        def inner():
            item = Item.objects.get(id=2)
            item.content = "changed b"
            item.save()
            return "changed a"

        self.items_are('initially', (1, "a"), (2, "b"))
        try:
            outer()
        except RuntimeError:
            pass        
        self.items_are('finally', (1, "a"), (2, "b"))