Changeset 8223 for django/trunk/tests/regressiontests/dispatch/tests
- Timestamp:
- 08/06/08 10:32:46 (4 months ago)
- Files:
Legend:
- Unmodified
- Added
- Removed
- Modified
- Copied
- Moved
django/trunk/tests/regressiontests/dispatch/tests/__init__.py
r5876 r8223 3 3 """ 4 4 5 from test_saferef import * 5 6 from test_dispatcher import * 6 from test_robustapply import *7 from test_saferef import *django/trunk/tests/regressiontests/dispatch/tests/test_dispatcher.py
r8004 r8223 1 from django.dispatch.dispatcher import * 2 from django.dispatch import dispatcher, robust 1 from django.dispatch import Signal 3 2 import unittest 4 3 import copy … … 16 15 gc.collect() 17 16 18 def x(a): 19 return a 20 21 class Dummy(object): 22 pass 17 def receiver_1_arg(val, **kwargs): 18 return val 23 19 24 20 class Callable(object): 25 def __call__(self, a):26 return a21 def __call__(self, val, **kwargs): 22 return val 27 23 28 def a(self, a): 29 return a 24 def a(self, val, **kwargs): 25 return val 26 27 a_signal = Signal(providing_args=["val"]) 30 28 31 29 class DispatcherTests(unittest.TestCase): 32 30 """Test suite for dispatcher (barely started)""" 33 34 def setUp(self): 35 # track the initial state, since it's possible that others have bleed receivers in 36 garbage_collect() 37 self.sendersBack = copy.copy(dispatcher.sendersBack) 38 self.connections = copy.copy(dispatcher.connections) 39 self.senders = copy.copy(dispatcher.senders) 40 41 def _testIsClean(self): 31 32 def _testIsClean(self, signal): 42 33 """Assert that everything has been cleaned up automatically""" 43 self.assertEqual(dispatcher.sendersBack, self.sendersBack) 44 self.assertEqual(dispatcher.connections, self.connections) 45 self.assertEqual(dispatcher.senders, self.senders) 34 self.assertEqual(signal.receivers, []) 35 36 # force cleanup just in case 37 signal.receivers = [] 46 38 47 39 def testExact(self): 48 a = Dummy() 49 signal = 'this' 50 connect(x, signal, a) 51 expected = [(x,a)] 52 result = send('this',a, a=a) 40 a_signal.connect(receiver_1_arg, sender=self) 41 expected = [(receiver_1_arg,"test")] 42 result = a_signal.send(sender=self, val="test") 53 43 self.assertEqual(result, expected) 54 disconnect(x, signal, a) 55 self.assertEqual(list(getAllReceivers(a,signal)), []) 56 self._testIsClean() 57 58 def testAnonymousSend(self): 59 a = Dummy() 60 signal = 'this' 61 connect(x, signal) 62 expected = [(x,a)] 63 result = send(signal,None, a=a) 44 a_signal.disconnect(receiver_1_arg, sender=self) 45 self._testIsClean(a_signal) 46 47 def testIgnoredSender(self): 48 a_signal.connect(receiver_1_arg) 49 expected = [(receiver_1_arg,"test")] 50 result = a_signal.send(sender=self, val="test") 64 51 self.assertEqual(result, expected) 65 disconnect(x, signal) 66 self.assertEqual(list(getAllReceivers(None,signal)), []) 67 self._testIsClean() 68 69 def testAnyRegistration(self): 70 a = Dummy() 71 signal = 'this' 72 connect(x, signal, Any) 73 expected = [(x,a)] 74 result = send('this',object(), a=a) 75 self.assertEqual(result, expected) 76 disconnect(x, signal, Any) 77 expected = [] 78 result = send('this',object(), a=a) 79 self.assertEqual(result, expected) 80 self.assertEqual(list(getAllReceivers(Any,signal)), []) 81 82 self._testIsClean() 83 84 def testAnyRegistration2(self): 85 a = Dummy() 86 signal = 'this' 87 connect(x, Any, a) 88 expected = [(x,a)] 89 result = send('this',a, a=a) 90 self.assertEqual(result, expected) 91 disconnect(x, Any, a) 92 self.assertEqual(list(getAllReceivers(a,Any)), []) 93 self._testIsClean() 52 a_signal.disconnect(receiver_1_arg) 53 self._testIsClean(a_signal) 94 54 95 55 def testGarbageCollected(self): 96 56 a = Callable() 97 b = Dummy() 98 signal = 'this' 99 connect(a.a, signal, b) 57 a_signal.connect(a.a, sender=self) 100 58 expected = [] 101 59 del a 102 60 garbage_collect() 103 result = send('this',b, a=b)61 result = a_signal.send(sender=self, val="test") 104 62 self.assertEqual(result, expected) 105 self.assertEqual(list(getAllReceivers(b,signal)), []) 106 self._testIsClean() 107 108 def testGarbageCollectedObj(self): 109 class x: 110 def __call__(self, a): 111 return a 112 a = Callable() 113 b = Dummy() 114 signal = 'this' 115 connect(a, signal, b) 116 expected = [] 117 del a 118 garbage_collect() 119 result = send('this',b, a=b) 120 self.assertEqual(result, expected) 121 self.assertEqual(list(getAllReceivers(b,signal)), []) 122 self._testIsClean() 123 63 self._testIsClean(a_signal) 124 64 125 65 def testMultipleRegistration(self): 126 66 a = Callable() 127 b = Dummy() 128 signal = 'this' 129 connect(a, signal, b) 130 connect(a, signal, b) 131 connect(a, signal, b) 132 connect(a, signal, b) 133 connect(a, signal, b) 134 connect(a, signal, b) 135 result = send('this',b, a=b) 67 a_signal.connect(a) 68 a_signal.connect(a) 69 a_signal.connect(a) 70 a_signal.connect(a) 71 a_signal.connect(a) 72 a_signal.connect(a) 73 result = a_signal.send(sender=self, val="test") 136 74 self.assertEqual(len(result), 1) 137 self.assertEqual(len( list(getAllReceivers(b,signal))), 1)75 self.assertEqual(len(a_signal.receivers), 1) 138 76 del a 139 del b140 77 del result 141 78 garbage_collect() 142 self._testIsClean() 79 self._testIsClean(a_signal) 80 81 def testUidRegistration(self): 82 def uid_based_receiver_1(**kwargs): 83 pass 84 85 def uid_based_receiver_2(**kwargs): 86 pass 87 88 a_signal.connect(uid_based_receiver_1, dispatch_uid = "uid") 89 a_signal.connect(uid_based_receiver_2, dispatch_uid = "uid") 90 self.assertEqual(len(a_signal.receivers), 1) 91 a_signal.disconnect(dispatch_uid = "uid") 92 self._testIsClean(a_signal) 143 93 144 94 def testRobust(self): 145 95 """Test the sendRobust function""" 146 def fails( ):96 def fails(val, **kwargs): 147 97 raise ValueError('this') 148 a = object() 149 signal = 'this' 150 connect(fails, Any, a) 151 result = robust.sendRobust('this',a, a=a) 98 a_signal.connect(fails) 99 result = a_signal.send_robust(sender=self, val="test") 152 100 err = result[0][1] 153 101 self.assert_(isinstance(err, ValueError)) 154 102 self.assertEqual(err.args, ('this',)) 103 a_signal.disconnect(fails) 104 self._testIsClean(a_signal) 155 105 156 106 def getSuite():
