Python Event Emitter
#Method 1
def test_twisted_emit():
"""Test that event_emitters can handle wrapping coroutines when using
twisted and ensureDeferred.
"""
ee = EventEmitter(scheduler=ensureDeferred)
should_call = Mock()
@ee.on('event')
async def event_handler():
should_call(True)
ee.emit('event')
should_call.assert_called_once()
#Method 2
def test_twisted_error():
"""Test that event_emitters can handle wrapping coroutines when using
twisted and ensureDeferred.
"""
ee = EventEmitter(scheduler=ensureDeferred)
should_call = Mock()
@ee.on('event')
async def event_handler():
raise PyeeTestException()
@ee.on('error')
def handle_error(e):
should_call(e)
ee.emit('event')
should_call.assert_called_once()
#Source file: test_async.py
Imaginathan