Add test capturing expectation that 'wait' will publish periodically to 'channel'.
import threading
import time
-import unittest
+import unittest.mock
from cherrypy.process import wspbus
@@ -194,6 +194,18 @@ class BusMethodTests(unittest.TestCase):
if b.state not in states:
self.fail('State %r not in %r' % (b.state, states))
+ def test_wait_publishes_periodically(self):
+ bus = wspbus.Bus()
+ callback = unittest.mock.MagicMock()
+ bus.subscribe('main', callback)
+
+ def set_start():
+ time.sleep(0.05)
+ bus.start()
+ threading.Thread(target=set_start).start()
+ bus.wait(bus.states.STARTED, interval=0.01, channel='main')
+ assert callback.call_count > 3
+
def test_block(self):
b = wspbus.Bus()
self.log(b)
