hmz007
36ed224bac
Signed-off-by: hmz007 <hmz007@gmail.com> Change-Id: Ib87fdbe7a0be7dc961af3500fe9ea4589a127f9f
113 lines
2.4 KiB
Python
113 lines
2.4 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
import pytest
|
|
import pytest_trio.plugin # noqa
|
|
import trio
|
|
|
|
from pyee import TrioEventEmitter
|
|
|
|
|
|
class PyeeTestError(Exception):
|
|
pass
|
|
|
|
|
|
@pytest.mark.trio
|
|
async def test_trio_emit():
|
|
"""Test that the trio event emitter can handle wrapping
|
|
coroutines
|
|
"""
|
|
|
|
async with TrioEventEmitter() as ee:
|
|
|
|
should_call = trio.Event()
|
|
|
|
@ee.on("event")
|
|
async def event_handler():
|
|
should_call.set()
|
|
|
|
ee.emit("event")
|
|
|
|
result = False
|
|
with trio.move_on_after(0.1):
|
|
await should_call.wait()
|
|
result = True
|
|
|
|
assert result
|
|
|
|
|
|
@pytest.mark.trio
|
|
async def test_trio_once_emit():
|
|
"""Test that trio event emitters also wrap coroutines when
|
|
using once
|
|
"""
|
|
|
|
async with TrioEventEmitter() as ee:
|
|
should_call = trio.Event()
|
|
|
|
@ee.once("event")
|
|
async def event_handler():
|
|
should_call.set()
|
|
|
|
ee.emit("event")
|
|
|
|
result = False
|
|
with trio.move_on_after(0.1):
|
|
await should_call.wait()
|
|
result = True
|
|
|
|
assert result
|
|
|
|
|
|
@pytest.mark.trio
|
|
async def test_trio_error():
|
|
"""Test that trio event emitters can handle errors when
|
|
wrapping coroutines
|
|
"""
|
|
|
|
async with TrioEventEmitter() as ee:
|
|
send, rcv = trio.open_memory_channel(1)
|
|
|
|
@ee.on("event")
|
|
async def event_handler():
|
|
raise PyeeTestError()
|
|
|
|
@ee.on("error")
|
|
async def handle_error(exc):
|
|
async with send:
|
|
await send.send(exc)
|
|
|
|
ee.emit("event")
|
|
|
|
result = None
|
|
with trio.move_on_after(0.1):
|
|
async with rcv:
|
|
result = await rcv.__anext__()
|
|
|
|
assert isinstance(result, PyeeTestError)
|
|
|
|
|
|
@pytest.mark.trio
|
|
async def test_sync_error(event_loop):
|
|
"""Test that regular functions have the same error handling as coroutines"""
|
|
|
|
async with TrioEventEmitter() as ee:
|
|
send, rcv = trio.open_memory_channel(1)
|
|
|
|
@ee.on("event")
|
|
def sync_handler():
|
|
raise PyeeTestError()
|
|
|
|
@ee.on("error")
|
|
async def handle_error(exc):
|
|
async with send:
|
|
await send.send(exc)
|
|
|
|
ee.emit("event")
|
|
|
|
result = None
|
|
with trio.move_on_after(0.1):
|
|
async with rcv:
|
|
result = await rcv.__anext__()
|
|
|
|
assert isinstance(result, PyeeTestError)
|