django/tests/regressiontests/dispatch/tests/test_dispatcher.py

from django.dispatch.dispatcher import *
from django.dispatch import dispatcher, robust
import unittest
import copy

def x(a):
    return a

class Dummy(object):
    pass

class Callable(object):
    def __call__(self, a):
        return a
    
    def a(self, a):
        return a

class DispatcherTests(unittest.TestCase):
    """Test suite for dispatcher (barely started)"""
    
    def setUp(self):
        # track the initial state, since it's possible that others have bleed receivers in
        self.sendersBack = copy.copy(dispatcher.sendersBack)
        self.connections = copy.copy(dispatcher.connections)
        self.senders = copy.copy(dispatcher.senders)
    
    def _testIsClean(self):
        """Assert that everything has been cleaned up automatically"""
        self.assertEqual(dispatcher.sendersBack, self.sendersBack)
        self.assertEqual(dispatcher.connections, self.connections)
        self.assertEqual(dispatcher.senders, self.senders)
    
    def testExact(self):
        a = Dummy()
        signal = 'this'
        connect(x, signal, a)
        expected = [(x,a)]
        result = send('this',a, a=a)
        self.assertEqual(result, expected)
        disconnect(x, signal, a)
        self.assertEqual(list(getAllReceivers(a,signal)), [])
        self._testIsClean()
    
    def testAnonymousSend(self):
        a = Dummy()
        signal = 'this'
        connect(x, signal)
        expected = [(x,a)]
        result = send(signal,None, a=a)
        self.assertEqual(result, expected)
        disconnect(x, signal)
        self.assertEqual(list(getAllReceivers(None,signal)), [])
        self._testIsClean()
    
    def testAnyRegistration(self):
        a = Dummy()
        signal = 'this'
        connect(x, signal, Any)
        expected = [(x,a)]
        result = send('this',object(), a=a)
        self.assertEqual(result, expected)
        disconnect(x, signal, Any)
        expected = []
        result = send('this',object(), a=a)
        self.assertEqual(result, expected)
        self.assertEqual(list(getAllReceivers(Any,signal)), [])
        
        self._testIsClean()
    
    def testAnyRegistration2(self):
        a = Dummy()
        signal = 'this'
        connect(x, Any, a)
        expected = [(x,a)]
        result = send('this',a, a=a)
        self.assertEqual(result, expected)
        disconnect(x, Any, a)
        self.assertEqual(list(getAllReceivers(a,Any)), [])
        self._testIsClean()
    
    def testGarbageCollected(self):
        a = Callable()
        b = Dummy()
        signal = 'this'
        connect(a.a, signal, b)
        expected = []
        del a
        result = send('this',b, a=b)
        self.assertEqual(result, expected)
        self.assertEqual(list(getAllReceivers(b,signal)), [])
        self._testIsClean()
    
    def testGarbageCollectedObj(self):
        class x:
            def __call__(self, a):
                return a
        a = Callable()
        b = Dummy()
        signal = 'this'
        connect(a, signal, b)
        expected = []
        del a
        result = send('this',b, a=b)
        self.assertEqual(result, expected)
        self.assertEqual(list(getAllReceivers(b,signal)), [])
        self._testIsClean()

    
    def testMultipleRegistration(self):
        a = Callable()
        b = Dummy()
        signal = 'this'
        connect(a, signal, b)
        connect(a, signal, b)
        connect(a, signal, b)
        connect(a, signal, b)
        connect(a, signal, b)
        connect(a, signal, b)
        result = send('this',b, a=b)
        self.assertEqual(len(result), 1)
        self.assertEqual(len(list(getAllReceivers(b,signal))), 1)
        del a
        del b
        del result
        self._testIsClean()
    
    def testRobust(self):
        """Test the sendRobust function"""
        def fails():
            raise ValueError('this')
        a = object()
        signal = 'this'
        connect(fails, Any, a)
        result = robust.sendRobust('this',a, a=a)
        err = result[0][1]
        self.assert_(isinstance(err, ValueError))
        self.assertEqual(err.args, ('this',))

def getSuite():
    return unittest.makeSuite(DispatcherTests,'test')

if __name__ == "__main__":
    unittest.main ()