1# -*- coding: utf-8 -*-
2
3import pytest
4import pytest_trio.plugin  # noqa
5import trio
6
7from pyee import TrioEventEmitter
8
9
10class PyeeTestError(Exception):
11    pass
12
13
14@pytest.mark.trio
15async def test_trio_emit():
16    """Test that the trio event emitter can handle wrapping
17    coroutines
18    """
19
20    async with TrioEventEmitter() as ee:
21
22        should_call = trio.Event()
23
24        @ee.on("event")
25        async def event_handler():
26            should_call.set()
27
28        ee.emit("event")
29
30        result = False
31        with trio.move_on_after(0.1):
32            await should_call.wait()
33            result = True
34
35        assert result
36
37
38@pytest.mark.trio
39async def test_trio_once_emit():
40    """Test that trio event emitters also wrap coroutines when
41    using once
42    """
43
44    async with TrioEventEmitter() as ee:
45        should_call = trio.Event()
46
47        @ee.once("event")
48        async def event_handler():
49            should_call.set()
50
51        ee.emit("event")
52
53        result = False
54        with trio.move_on_after(0.1):
55            await should_call.wait()
56            result = True
57
58        assert result
59
60
61@pytest.mark.trio
62async def test_trio_error():
63    """Test that trio event emitters can handle errors when
64    wrapping coroutines
65    """
66
67    async with TrioEventEmitter() as ee:
68        send, rcv = trio.open_memory_channel(1)
69
70        @ee.on("event")
71        async def event_handler():
72            raise PyeeTestError()
73
74        @ee.on("error")
75        async def handle_error(exc):
76            async with send:
77                await send.send(exc)
78
79        ee.emit("event")
80
81        result = None
82        with trio.move_on_after(0.1):
83            async with rcv:
84                result = await rcv.__anext__()
85
86        assert isinstance(result, PyeeTestError)
87
88
89@pytest.mark.trio
90async def test_sync_error(event_loop):
91    """Test that regular functions have the same error handling as coroutines"""
92
93    async with TrioEventEmitter() as ee:
94        send, rcv = trio.open_memory_channel(1)
95
96        @ee.on("event")
97        def sync_handler():
98            raise PyeeTestError()
99
100        @ee.on("error")
101        async def handle_error(exc):
102            async with send:
103                await send.send(exc)
104
105        ee.emit("event")
106
107        result = None
108        with trio.move_on_after(0.1):
109            async with rcv:
110                result = await rcv.__anext__()
111
112        assert isinstance(result, PyeeTestError)
113