test_bus.py (7457B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\carbon\common\lib\cherrypy\test\test_bus.py 2 import threading 3 import time 4 import unittest 5 import cherrypy 6 from cherrypy._cpcompat import get_daemon, set 7 from cherrypy.process import wspbus 8 msg = 'Listener %d on channel %s: %s.' 9 10 class PublishSubscribeTests(unittest.TestCase): 11 12 def get_listener(self, channel, index): 13 14 def listener(arg = None): 15 self.responses.append(msg % (index, channel, arg)) 16 17 return listener 18 19 def test_builtin_channels(self): 20 b = wspbus.Bus() 21 self.responses, expected = [], [] 22 for channel in b.listeners: 23 for index, priority in enumerate([100, 24 50, 25 0, 26 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 37 def test_custom_channels(self): 38 b = wspbus.Bus() 39 self.responses, expected = [], [] 40 custom_listeners = ('hugh', 'louis', 'dewey') 41 for channel in custom_listeners: 42 for index, priority in enumerate([None, 43 10, 44 60, 45 40]): 46 b.subscribe(channel, self.get_listener(channel, index), priority) 47 48 for channel in custom_listeners: 49 b.publish(channel, 'ah so') 50 expected.extend([ msg % (i, channel, 'ah so') for i in (1, 3, 0, 2) ]) 51 b.publish(channel) 52 expected.extend([ msg % (i, channel, None) for i in (1, 3, 0, 2) ]) 53 54 self.assertEqual(self.responses, expected) 55 56 def test_listener_errors(self): 57 b = wspbus.Bus() 58 self.responses, expected = [], [] 59 channels = [ c for c in b.listeners if c != 'log' ] 60 for channel in channels: 61 b.subscribe(channel, self.get_listener(channel, 1)) 62 b.subscribe(channel, lambda : None, priority=20) 63 64 for channel in channels: 65 self.assertRaises(wspbus.ChannelFailures, b.publish, channel, 123) 66 expected.append(msg % (1, channel, 123)) 67 68 self.assertEqual(self.responses, expected) 69 70 71 class BusMethodTests(unittest.TestCase): 72 73 def log(self, bus): 74 self._log_entries = [] 75 76 def logit(msg, level): 77 self._log_entries.append(msg) 78 79 bus.subscribe('log', logit) 80 81 def assertLog(self, entries): 82 self.assertEqual(self._log_entries, entries) 83 84 def get_listener(self, channel, index): 85 86 def listener(arg = None): 87 self.responses.append(msg % (index, channel, arg)) 88 89 return listener 90 91 def test_start(self): 92 b = wspbus.Bus() 93 self.log(b) 94 self.responses = [] 95 num = 3 96 for index in range(num): 97 b.subscribe('start', self.get_listener('start', index)) 98 99 b.start() 100 try: 101 self.assertEqual(set(self.responses), set([ msg % (i, 'start', None) for i in range(num) ])) 102 self.assertEqual(b.state, b.states.STARTED) 103 self.assertLog(['Bus STARTING', 'Bus STARTED']) 104 finally: 105 b.exit() 106 107 def test_stop(self): 108 b = wspbus.Bus() 109 self.log(b) 110 self.responses = [] 111 num = 3 112 for index in range(num): 113 b.subscribe('stop', self.get_listener('stop', index)) 114 115 b.stop() 116 self.assertEqual(set(self.responses), set([ msg % (i, 'stop', None) for i in range(num) ])) 117 self.assertEqual(b.state, b.states.STOPPED) 118 self.assertLog(['Bus STOPPING', 'Bus STOPPED']) 119 120 def test_graceful(self): 121 b = wspbus.Bus() 122 self.log(b) 123 self.responses = [] 124 num = 3 125 for index in range(num): 126 b.subscribe('graceful', self.get_listener('graceful', index)) 127 128 b.graceful() 129 self.assertEqual(set(self.responses), set([ msg % (i, 'graceful', None) for i in range(num) ])) 130 self.assertLog(['Bus graceful']) 131 132 def test_exit(self): 133 b = wspbus.Bus() 134 self.log(b) 135 self.responses = [] 136 num = 3 137 for index in range(num): 138 b.subscribe('stop', self.get_listener('stop', index)) 139 b.subscribe('exit', self.get_listener('exit', index)) 140 141 b.exit() 142 self.assertEqual(set(self.responses), set([ msg % (i, 'stop', None) for i in range(num) ] + [ msg % (i, 'exit', None) for i in range(num) ])) 143 self.assertEqual(b.state, b.states.EXITING) 144 self.assertLog(['Bus STOPPING', 145 'Bus STOPPED', 146 'Bus EXITING', 147 'Bus EXITED']) 148 149 def test_wait(self): 150 b = wspbus.Bus() 151 152 def f(method): 153 time.sleep(0.2) 154 getattr(b, method)() 155 156 for method, states in [('start', [b.states.STARTED]), 157 ('stop', [b.states.STOPPED]), 158 ('start', [b.states.STARTING, b.states.STARTED]), 159 ('exit', [b.states.EXITING])]: 160 threading.Thread(target=f, args=(method,)).start() 161 b.wait(states) 162 if b.state not in states: 163 self.fail('State %r not in %r' % (b.state, states)) 164 165 def test_block(self): 166 b = wspbus.Bus() 167 self.log(b) 168 169 def f(): 170 time.sleep(0.2) 171 b.exit() 172 173 def g(): 174 time.sleep(0.4) 175 176 threading.Thread(target=f).start() 177 threading.Thread(target=g).start() 178 threads = [ t for t in threading.enumerate() if not get_daemon(t) ] 179 self.assertEqual(len(threads), 3) 180 b.block() 181 self.assertEqual(b.state, b.states.EXITING) 182 threads = [ t for t in threading.enumerate() if not get_daemon(t) ] 183 self.assertEqual(len(threads), 1) 184 self.assertEqual(self._log_entries[:-1], ['Bus STOPPING', 185 'Bus STOPPED', 186 'Bus EXITING', 187 'Bus EXITED', 188 'Waiting for child threads to terminate...']) 189 190 def test_start_with_callback(self): 191 b = wspbus.Bus() 192 self.log(b) 193 try: 194 events = [] 195 196 def f(*args, **kwargs): 197 events.append(('f', args, kwargs)) 198 199 def g(): 200 events.append('g') 201 202 b.subscribe('start', g) 203 b.start_with_callback(f, (1, 3, 5), {'foo': 'bar'}) 204 time.sleep(0.2) 205 self.assertEqual(b.state, b.states.STARTED) 206 self.assertEqual(events, ['g', ('f', (1, 3, 5), {'foo': 'bar'})]) 207 finally: 208 b.exit() 209 210 def test_log(self): 211 b = wspbus.Bus() 212 self.log(b) 213 self.assertLog([]) 214 expected = [] 215 for msg in ["O mah darlin'"] * 3 + ['Clementiiiiiiiine']: 216 b.log(msg) 217 expected.append(msg) 218 self.assertLog(expected) 219 220 try: 221 foo 222 except NameError: 223 b.log('You are lost and gone forever', traceback=True) 224 lastmsg = self._log_entries[-1] 225 if 'Traceback' not in lastmsg or 'NameError' not in lastmsg: 226 self.fail('Last log message %r did not contain the expected traceback.' % lastmsg) 227 else: 228 self.fail('NameError was not raised as expected.') 229 230 231 if __name__ == '__main__': 232 unittest.main()