1 from django.dispatch.dispatcher import *

     2 from django.dispatch import dispatcher, robust

     3 import unittest

     4 import copy

     5 

     6 def x(a):

     7     return a

     8 

     9 class Dummy(object):

    10     pass

    11 

    12 class Callable(object):

    13     def __call__(self, a):

    14         return a

    15     

    16     def a(self, a):

    17         return a

    18 

    19 class DispatcherTests(unittest.TestCase):

    20     """Test suite for dispatcher (barely started)"""

    21     

    22     def setUp(self):

    23         # track the initial state, since it's possible that others have bleed receivers in

    24         self.sendersBack = copy.copy(dispatcher.sendersBack)

    25         self.connections = copy.copy(dispatcher.connections)

    26         self.senders = copy.copy(dispatcher.senders)

    27     

    28     def _testIsClean(self):

    29         """Assert that everything has been cleaned up automatically"""

    30         self.assertEqual(dispatcher.sendersBack, self.sendersBack)

    31         self.assertEqual(dispatcher.connections, self.connections)

    32         self.assertEqual(dispatcher.senders, self.senders)

    33     

    34     def testExact(self):

    35         a = Dummy()

    36         signal = 'this'

    37         connect(x, signal, a)

    38         expected = [(x,a)]

    39         result = send('this',a, a=a)

    40         self.assertEqual(result, expected)

    41         disconnect(x, signal, a)

    42         self.assertEqual(list(getAllReceivers(a,signal)), [])

    43         self._testIsClean()

    44     

    45     def testAnonymousSend(self):

    46         a = Dummy()

    47         signal = 'this'

    48         connect(x, signal)

    49         expected = [(x,a)]

    50         result = send(signal,None, a=a)

    51         self.assertEqual(result, expected)

    52         disconnect(x, signal)

    53         self.assertEqual(list(getAllReceivers(None,signal)), [])

    54         self._testIsClean()

    55     

    56     def testAnyRegistration(self):

    57         a = Dummy()

    58         signal = 'this'

    59         connect(x, signal, Any)

    60         expected = [(x,a)]

    61         result = send('this',object(), a=a)

    62         self.assertEqual(result, expected)

    63         disconnect(x, signal, Any)

    64         expected = []

    65         result = send('this',object(), a=a)

    66         self.assertEqual(result, expected)

    67         self.assertEqual(list(getAllReceivers(Any,signal)), [])

    68         

    69         self._testIsClean()

    70     

    71     def testAnyRegistration2(self):

    72         a = Dummy()

    73         signal = 'this'

    74         connect(x, Any, a)

    75         expected = [(x,a)]

    76         result = send('this',a, a=a)

    77         self.assertEqual(result, expected)

    78         disconnect(x, Any, a)

    79         self.assertEqual(list(getAllReceivers(a,Any)), [])

    80         self._testIsClean()

    81     

    82     def testGarbageCollected(self):

    83         a = Callable()

    84         b = Dummy()

    85         signal = 'this'

    86         connect(a.a, signal, b)

    87         expected = []

    88         del a

    89         result = send('this',b, a=b)

    90         self.assertEqual(result, expected)

    91         self.assertEqual(list(getAllReceivers(b,signal)), [])

    92         self._testIsClean()

    93     

    94     def testGarbageCollectedObj(self):

    95         class x:

    96             def __call__(self, a):

    97                 return a

    98         a = Callable()

    99         b = Dummy()

   100         signal = 'this'

   101         connect(a, signal, b)

   102         expected = []

   103         del a

   104         result = send('this',b, a=b)

   105         self.assertEqual(result, expected)

   106         self.assertEqual(list(getAllReceivers(b,signal)), [])

   107         self._testIsClean()

   108 

   109     

   110     def testMultipleRegistration(self):

   111         a = Callable()

   112         b = Dummy()

   113         signal = 'this'

   114         connect(a, signal, b)

   115         connect(a, signal, b)

   116         connect(a, signal, b)

   117         connect(a, signal, b)

   118         connect(a, signal, b)

   119         connect(a, signal, b)

   120         result = send('this',b, a=b)

   121         self.assertEqual(len(result), 1)

   122         self.assertEqual(len(list(getAllReceivers(b,signal))), 1)

   123         del a

   124         del b

   125         del result

   126         self._testIsClean()

   127     

   128     def testRobust(self):

   129         """Test the sendRobust function"""

   130         def fails():

   131             raise ValueError('this')

   132         a = object()

   133         signal = 'this'

   134         connect(fails, Any, a)

   135         result = robust.sendRobust('this',a, a=a)

   136         err = result[0][1]

   137         self.assert_(isinstance(err, ValueError))

   138         self.assertEqual(err.args, ('this',))

   139 

   140 def getSuite():

   141     return unittest.makeSuite(DispatcherTests,'test')

   142 

   143 if __name__ == "__main__":

   144     unittest.main ()