1# -*- coding: utf-8 -*-
2
3from asyncio import Future, wait_for
4
5import pytest
6import pytest_asyncio.plugin  # noqa
7
8try:
9    from asyncio.exceptions import TimeoutError  # type: ignore
10except ImportError:
11    from concurrent.futures import TimeoutError  # type: ignore
12
13from mock import Mock
14from twisted.internet.defer import succeed
15
16from pyee import AsyncIOEventEmitter, TwistedEventEmitter
17
18
19class PyeeTestError(Exception):
20    pass
21
22
23@pytest.mark.asyncio
24async def test_asyncio_emit(event_loop):
25    """Test that AsyncIOEventEmitter can handle wrapping
26    coroutines
27    """
28
29    ee = AsyncIOEventEmitter(loop=event_loop)
30
31    should_call = Future(loop=event_loop)
32
33    @ee.on("event")
34    async def event_handler():
35        should_call.set_result(True)
36
37    ee.emit("event")
38
39    result = await wait_for(should_call, 0.1)
40
41    assert result is True
42
43
44@pytest.mark.asyncio
45async def test_asyncio_once_emit(event_loop):
46    """Test that AsyncIOEventEmitter also wrap coroutines when
47    using once
48    """
49
50    ee = AsyncIOEventEmitter(loop=event_loop)
51
52    should_call = Future(loop=event_loop)
53
54    @ee.once("event")
55    async def event_handler():
56        should_call.set_result(True)
57
58    ee.emit("event")
59
60    result = await wait_for(should_call, 0.1)
61
62    assert result is True
63
64
65@pytest.mark.asyncio
66async def test_asyncio_error(event_loop):
67    """Test that AsyncIOEventEmitter can handle errors when
68    wrapping coroutines
69    """
70    ee = AsyncIOEventEmitter(loop=event_loop)
71
72    should_call = Future(loop=event_loop)
73
74    @ee.on("event")
75    async def event_handler():
76        raise PyeeTestError()
77
78    @ee.on("error")
79    def handle_error(exc):
80        should_call.set_result(exc)
81
82    ee.emit("event")
83
84    result = await wait_for(should_call, 0.1)
85
86    assert isinstance(result, PyeeTestError)
87
88
89@pytest.mark.asyncio
90async def test_asyncio_cancellation(event_loop):
91    """Test that AsyncIOEventEmitter can handle Future cancellations"""
92
93    cancel_me = Future(loop=event_loop)
94    should_not_call = Future(loop=event_loop)
95
96    ee = AsyncIOEventEmitter(loop=event_loop)
97
98    @ee.on("event")
99    async def event_handler():
100        cancel_me.cancel()
101
102    @ee.on("error")
103    def handle_error(exc):
104        should_not_call.set_result(None)
105
106    ee.emit("event")
107
108    try:
109        await wait_for(should_not_call, 0.1)
110    except TimeoutError:
111        pass
112    else:
113        raise PyeeTestError()
114
115
116@pytest.mark.asyncio
117async def test_sync_error(event_loop):
118    """Test that regular functions have the same error handling as coroutines"""
119    ee = AsyncIOEventEmitter(loop=event_loop)
120
121    should_call = Future(loop=event_loop)
122
123    @ee.on("event")
124    def sync_handler():
125        raise PyeeTestError()
126
127    @ee.on("error")
128    def handle_error(exc):
129        should_call.set_result(exc)
130
131    ee.emit("event")
132
133    result = await wait_for(should_call, 0.1)
134
135    assert isinstance(result, PyeeTestError)
136
137
138def test_twisted_emit():
139    """Test that TwistedEventEmitter can handle wrapping
140    coroutines
141    """
142    ee = TwistedEventEmitter()
143
144    should_call = Mock()
145
146    @ee.on("event")
147    async def event_handler():
148        _ = await succeed("yes!")
149        should_call(True)
150
151    ee.emit("event")
152
153    should_call.assert_called_once()
154
155
156def test_twisted_once():
157    """Test that TwistedEventEmitter also wraps coroutines for
158    once
159    """
160    ee = TwistedEventEmitter()
161
162    should_call = Mock()
163
164    @ee.once("event")
165    async def event_handler():
166        _ = await succeed("yes!")
167        should_call(True)
168
169    ee.emit("event")
170
171    should_call.assert_called_once()
172
173
174def test_twisted_error():
175    """Test that TwistedEventEmitters handle Failures when wrapping coroutines."""
176    ee = TwistedEventEmitter()
177
178    should_call = Mock()
179
180    @ee.on("event")
181    async def event_handler():
182        raise PyeeTestError()
183
184    @ee.on("failure")
185    def handle_error(e):
186        should_call(e)
187
188    ee.emit("event")
189
190    should_call.assert_called_once()
191