How to use pyee - 10 common examples

def test_once():
    """Test that `once()` method works propers.
    """

    # very similar to "test_emit" but also makes sure that the event
    # gets removed afterwards

    ee = EventEmitter()

    def once_handler(data, error=None):
        nt.assert_equals(data, 'emitter is emitted!')
        if (error):
            raise ItWorkedException

    # Tests to make sure that after event is emitted that it's gone.
    ee.once('event', once_handler)
    ee.emit('event', 'emitter is emitted!')
    nt.assert_equal(ee._events['event'], [])

    # Tests to make sure callback fires. "Hides" other exceptions.
    with nt.assert_raises(ItWorkedException) as it_worked:  # noqa
        ee.once('event', once_handler)
        ee.emit('event', 'emitter is emitted!', True)

    EventEmitter
])
def test_emit_error(cls):
    """Errors raise with no event handler, otherwise emit on handler"""

    call_me = Mock()
    ee = cls()

    test_exception = PyeeTestException('lololol')

    with raises(PyeeTestException):
        ee.emit('error', test_exception)

    @ee.on('error')
    def on_error(exc):
        call_me()

def test_once():
    """Test that `once()` method works propers.
    """

    # very similar to "test_emit" but also makes sure that the event
    # gets removed afterwards

    call_me = Mock()
    ee = BaseEventEmitter()

    def once_handler(data):
        assert data == 'emitter is emitted!'
        call_me()

    # Tests to make sure that after event is emitted that it's gone.
    ee.once('event', once_handler)

    ee.emit('event', 'emitter is emitted!')

    call_me.assert_called_once()

    assert ee._events['event'] == OrderedDict()

def test_listener_removal_on_emit():
    """Test that a listener removed during an emit is called inside the current
    emit cycle.
    """

    call_me = Mock()
    ee = BaseEventEmitter()

    def should_remove():
        ee.remove_listener('remove', call_me)

    ee.on('remove', should_remove)
    ee.on('remove', call_me)

    ee.emit('remove')

    call_me.assert_called_once()

    call_me.reset_mock()

    # Also test with the listeners added in the opposite order
    ee = BaseEventEmitter()
    ee.on('remove', call_me)

    BaseEventEmitter,
    EventEmitter
])
def test_emit_sync(cls):
    """Basic synchronous emission works"""

    call_me = Mock()
    ee = cls()

    @ee.on('event')
    def event_handler(data, **kwargs):
        call_me()
        assert data == 'emitter is emitted!'

    # Making sure data is passed propers
    ee.emit('event', 'emitter is emitted!', error=False)

    AsyncIOEventEmitter,
    EventEmitter
])
@pytest.mark.asyncio
async def test_asyncio_emit(cls, event_loop):
    """Test that asyncio-supporting event emitters can handle wrapping
    coroutines
    """

    ee = cls(loop=event_loop)

    should_call = Future(loop=event_loop)

    @ee.on('event')
    async def event_handler():
        should_call.set_result(True)

    (TwistedEventEmitter, dict()),
    (EventEmitter, dict(scheduler=ensureDeferred))
])
def test_twisted_emit(cls, kwargs):
    """Test that twisted-supporting event emitters can handle wrapping
    coroutines
    """
    ee = cls(**kwargs)

    should_call = Mock()

    @ee.on('event')
    async def event_handler():
        _ = await succeed('yes!')
        should_call(True)

    ee.emit('event')

def test_twisted_error():
    """Test that TwistedEventEmitters handle Failures when wrapping coroutines.
    """
    ee = TwistedEventEmitter()

    should_call = Mock()

    @ee.on('event')
    async def event_handler():
        raise PyeeTestError()

    @ee.on('failure')
    def handle_error(e):
        should_call(e)

    ee.emit('event')

    should_call.assert_called_once()

def test_properties_preserved():
    """Test that the properties of decorated functions are preserved."""

    call_me = Mock()
    call_me_also = Mock()
    ee = BaseEventEmitter()

    @ee.on('always')
    def always_event_handler():
        """An event handler."""
        call_me()

    @ee.once('once')
    def once_event_handler():
        """Another event handler."""
        call_me_also()

    assert always_event_handler.__doc__ == 'An event handler.'
    assert once_event_handler.__doc__ == 'Another event handler.'

    always_event_handler()
    call_me.assert_called_once()

def test_inheritance():
    """Test that inheritance is preserved from object"""
    assert object in getmro(BaseEventEmitter)

    class example(BaseEventEmitter):
        def __init__(self):
            super(example, self).__init__()

    assert BaseEventEmitter in getmro(example)
    assert object in getmro(example)