PYTHON

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