1 import threading
2 import time
3 import unittest
4
5 import cherrypy
6 from cherrypy._cpcompat import get_daemon, set
7 from cherrypy.process import wspbus
8
9
10 msg = "Listener %d on channel %s: %s."
11
12
14
16 def listener(arg=None):
17 self.responses.append(msg % (index, channel, arg))
18 return listener
19
21 b = wspbus.Bus()
22
23 self.responses, expected = [], []
24
25 for channel in b.listeners:
26 for index, priority in enumerate([100, 50, 0, 51]):
27 b.subscribe(channel, self.get_listener(channel, index), priority)
28
29 for channel in b.listeners:
30 b.publish(channel)
31 expected.extend([msg % (i, channel, None) for i in (2, 1, 3, 0)])
32 b.publish(channel, arg=79347)
33 expected.extend([msg % (i, channel, 79347) for i in (2, 1, 3, 0)])
34
35 self.assertEqual(self.responses, expected)
36
38 b = wspbus.Bus()
39
40 self.responses, expected = [], []
41
42 custom_listeners = ('hugh', 'louis', 'dewey')
43 for channel in custom_listeners:
44 for index, priority in enumerate([None, 10, 60, 40]):
45 b.subscribe(channel, self.get_listener(channel, index), priority)
46
47 for channel in custom_listeners:
48 b.publish(channel, 'ah so')
49 expected.extend([msg % (i, channel, 'ah so') for i in (1, 3, 0, 2)])
50 b.publish(channel)
51 expected.extend([msg % (i, channel, None) for i in (1, 3, 0, 2)])
52
53 self.assertEqual(self.responses, expected)
54
56 b = wspbus.Bus()
57
58 self.responses, expected = [], []
59 channels = [c for c in b.listeners if c != 'log']
60
61 for channel in channels:
62 b.subscribe(channel, self.get_listener(channel, 1))
63
64 b.subscribe(channel, lambda: None, priority=20)
65
66 for channel in channels:
67 self.assertRaises(wspbus.ChannelFailures, b.publish, channel, 123)
68 expected.append(msg % (1, channel, 123))
69
70 self.assertEqual(self.responses, expected)
71
72
74
76 self._log_entries = []
77 def logit(msg, level):
78 self._log_entries.append(msg)
79 bus.subscribe('log', logit)
80
82 self.assertEqual(self._log_entries, entries)
83
85 def listener(arg=None):
86 self.responses.append(msg % (index, channel, arg))
87 return listener
88
90 b = wspbus.Bus()
91 self.log(b)
92
93 self.responses = []
94 num = 3
95 for index in range(num):
96 b.subscribe('start', self.get_listener('start', index))
97
98 b.start()
99 try:
100
101 self.assertEqual(set(self.responses),
102 set([msg % (i, 'start', None) for i in range(num)]))
103
104
105 self.assertEqual(b.state, b.states.STARTED)
106
107 self.assertLog(['Bus STARTING', 'Bus STARTED'])
108 finally:
109
110 b.exit()
111
113 b = wspbus.Bus()
114 self.log(b)
115
116 self.responses = []
117 num = 3
118 for index in range(num):
119 b.subscribe('stop', self.get_listener('stop', index))
120
121 b.stop()
122
123
124 self.assertEqual(set(self.responses),
125 set([msg % (i, 'stop', None) for i in range(num)]))
126
127 self.assertEqual(b.state, b.states.STOPPED)
128
129 self.assertLog(['Bus STOPPING', 'Bus STOPPED'])
130
132 b = wspbus.Bus()
133 self.log(b)
134
135 self.responses = []
136 num = 3
137 for index in range(num):
138 b.subscribe('graceful', self.get_listener('graceful', index))
139
140 b.graceful()
141
142
143 self.assertEqual(set(self.responses),
144 set([msg % (i, 'graceful', None) for i in range(num)]))
145
146 self.assertLog(['Bus graceful'])
147
149 b = wspbus.Bus()
150 self.log(b)
151
152 self.responses = []
153 num = 3
154 for index in range(num):
155 b.subscribe('stop', self.get_listener('stop', index))
156 b.subscribe('exit', self.get_listener('exit', index))
157
158 b.exit()
159
160
161
162 self.assertEqual(set(self.responses),
163 set([msg % (i, 'stop', None) for i in range(num)] +
164 [msg % (i, 'exit', None) for i in range(num)]))
165
166 self.assertEqual(b.state, b.states.EXITING)
167
168 self.assertLog(['Bus STOPPING', 'Bus STOPPED', 'Bus EXITING', 'Bus EXITED'])
169
176
177 for method, states in [('start', [b.states.STARTED]),
178 ('stop', [b.states.STOPPED]),
179 ('start', [b.states.STARTING, b.states.STARTED]),
180 ('exit', [b.states.EXITING]),
181 ]:
182 threading.Thread(target=f, args=(method,)).start()
183 b.wait(states)
184
185
186 if b.state not in states:
187 self.fail("State %r not in %r" % (b.state, states))
188
196 def g():
197 time.sleep(0.4)
198 threading.Thread(target=f).start()
199 threading.Thread(target=g).start()
200 threads = [t for t in threading.enumerate() if not get_daemon(t)]
201 self.assertEqual(len(threads), 3)
202
203 b.block()
204
205
206 self.assertEqual(b.state, b.states.EXITING)
207
208 threads = [t for t in threading.enumerate() if not get_daemon(t)]
209 self.assertEqual(len(threads), 1)
210
211 self.assertEqual(self._log_entries[:-1],
212 ['Bus STOPPING', 'Bus STOPPED',
213 'Bus EXITING', 'Bus EXITED',
214 'Waiting for child threads to terminate...'])
215
217 b = wspbus.Bus()
218 self.log(b)
219 try:
220 events = []
221 def f(*args, **kwargs):
222 events.append(("f", args, kwargs))
223 def g():
224 events.append("g")
225 b.subscribe("start", g)
226 b.start_with_callback(f, (1, 3, 5), {"foo": "bar"})
227
228 time.sleep(0.2)
229
230
231 self.assertEqual(b.state, b.states.STARTED)
232
233 self.assertEqual(events, ["g", ("f", (1, 3, 5), {"foo": "bar"})])
234 finally:
235 b.exit()
236
238 b = wspbus.Bus()
239 self.log(b)
240 self.assertLog([])
241
242
243 expected = []
244 for msg in ["O mah darlin'"] * 3 + ["Clementiiiiiiiine"]:
245 b.log(msg)
246 expected.append(msg)
247 self.assertLog(expected)
248
249
250 try:
251 foo
252 except NameError:
253 b.log("You are lost and gone forever", traceback=True)
254 lastmsg = self._log_entries[-1]
255 if "Traceback" not in lastmsg or "NameError" not in lastmsg:
256 self.fail("Last log message %r did not contain "
257 "the expected traceback." % lastmsg)
258 else:
259 self.fail("NameError was not raised as expected.")
260
261
262 if __name__ == "__main__":
263 unittest.main()
264