1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
|
from pydispatch.dispatcher import *
from pydispatch import dispatcher, robust
import unittest
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 _isclean( self ):
"""Assert that everything has been cleaned up automatically"""
assert len(dispatcher.sendersBack) == 0, dispatcher.sendersBack
assert len(dispatcher.connections) == 0, dispatcher.connections
assert len(dispatcher.senders) == 0, dispatcher.senders
def testExact (self):
a = Dummy()
signal = 'this'
connect( x, signal, a )
expected = [(x,a)]
result = send('this',a, a=a)
assert result == expected,"""Send didn't return expected result:\n\texpected:%s\n\tgot:%s"""% (expected, result)
disconnect( x, signal, a )
assert len(list(getAllReceivers(a,signal))) == 0
self._isclean()
def testAnonymousSend(self):
a = Dummy()
signal = 'this'
connect( x, signal )
expected = [(x,a)]
result = send(signal,None, a=a)
assert result == expected,"""Send didn't return expected result:\n\texpected:%s\n\tgot:%s"""% (expected, result)
disconnect( x, signal )
assert len(list(getAllReceivers(None,signal))) == 0
self._isclean()
def testAnyRegistration(self):
a = Dummy()
signal = 'this'
connect( x, signal, Any )
expected = [(x,a)]
result = send('this',object(), a=a)
assert result == expected,"""Send didn't return expected result:\n\texpected:%s\n\tgot:%s"""% (expected, result)
disconnect( x, signal, Any )
expected = []
result = send('this',object(), a=a)
assert result == expected,"""Send didn't return expected result:\n\texpected:%s\n\tgot:%s"""% (expected, result)
assert len(list(getAllReceivers(Any,signal))) == 0
self._isclean()
def testAnyRegistration2(self):
a = Dummy()
signal = 'this'
connect( x, Any, a )
expected = [(x,a)]
result = send(signal,a, a=a)
assert result == expected,"""Send didn't return expected result:\n\texpected:%s\n\tgot:%s"""% (expected, result)
disconnect( x, Any, a )
assert len(list(getAllReceivers(a,Any))) == 0
self._isclean()
def testGarbageCollected(self):
a = Callable()
b = Dummy()
signal = 'this'
connect( a.a, signal, b )
expected = []
del a
result = send('this',b, a=b)
assert result == expected,"""Send didn't return expected result:\n\texpected:%s\n\tgot:%s"""% (expected, result)
assert len(list(getAllReceivers(b,signal))) == 0, """Remaining handlers: %s"""%(getAllReceivers(b,signal),)
self._isclean()
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)
assert result == expected,"""Send didn't return expected result:\n\texpected:%s\n\tgot:%s"""% (expected, result)
assert len(list(getAllReceivers(b,signal))) == 0, """Remaining handlers: %s"""%(getAllReceivers(b,signal),)
self._isclean()
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)
assert len( result ) == 1, result
assert len(list(getAllReceivers(b,signal))) == 1, """Remaining handlers: %s"""%(getAllReceivers(b,signal),)
del a
del b
del result
self._isclean()
def testRobust( self ):
"""Test the sendRobust function"""
signal = 'this'
def fails( ):
raise ValueError( signal )
a = object()
connect( fails, Any, a )
result = robust.sendRobust(signal,a, a=a)
err = result[0][1]
assert isinstance( err, ValueError )
assert err.args == (signal,)
def testParameterRepr(self):
assert repr(dispatcher.Any) == '_Any', repr(dispatcher.Any)
assert str(dispatcher.Any) == '_Any', str(dispatcher.Any)
def testNoNoneSignal(self):
self.assertRaises(
errors.DispatcherTypeError,
dispatcher.connect, x, signal=None
)
self.assertRaises(
errors.DispatcherTypeError,
dispatcher.disconnect, x, signal=None
)
def testDisconnectUnconnected(self):
self.assertRaises(
errors.DispatcherKeyError,
dispatcher.disconnect, x, signal='not-registered'
)
def getSuite():
return unittest.makeSuite(DispatcherTests,'test')
if __name__ == "__main__":
unittest.main ()
|
