1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
|
# -*- coding: utf-8 -*-
from typing import Any, Generator
from unittest.mock import Mock
from twisted.internet.defer import Deferred, inlineCallbacks, succeed
from twisted.python.failure import Failure
from pyee.twisted import TwistedEventEmitter
class PyeeTestError(Exception):
pass
def test_emit() -> None:
"""Test that TwistedEventEmitter can handle wrapping
coroutines
"""
ee = TwistedEventEmitter()
should_call = Mock()
@ee.on("event")
async def event_handler() -> None:
_ = await succeed("yes!")
should_call(True)
ee.emit("event")
should_call.assert_called_once()
def test_once() -> None:
"""Test that TwistedEventEmitter also wraps coroutines for
once
"""
ee = TwistedEventEmitter()
should_call = Mock()
@ee.once("event")
async def event_handler():
_ = await succeed("yes!")
should_call(True)
ee.emit("event")
should_call.assert_called_once()
def test_error() -> None:
"""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_propagates_failure():
"""Test that TwistedEventEmitters can propagate failures
from twisted Deferreds
"""
ee = TwistedEventEmitter()
should_call = Mock()
@ee.on("event")
@inlineCallbacks
def event_handler() -> Generator[Deferred[object], object, None]:
d: Deferred[Any] = Deferred()
d.callback(Failure(PyeeTestError()))
yield d
@ee.on("failure")
def handle_failure(f: Any) -> None:
assert isinstance(f, Failure)
should_call(f)
ee.emit("event")
should_call.assert_called_once()
def test_propagates_sync_failure():
"""Test that TwistedEventEmitters can propagate failures
from twisted Deferreds
"""
ee = TwistedEventEmitter()
should_call = Mock()
@ee.on("event")
def event_handler():
raise PyeeTestError()
@ee.on("failure")
def handle_failure(f):
assert isinstance(f, Failure)
should_call(f)
ee.emit("event")
should_call.assert_called_once()
def test_propagates_exception():
"""Test that TwistedEventEmitters propagate failures as exceptions to
the error event when no failure handler
"""
ee = TwistedEventEmitter()
should_call = Mock()
@ee.on("event")
@inlineCallbacks
def event_handler() -> Generator[Deferred[object], object, None]:
d: Deferred[Any] = Deferred()
d.callback(Failure(PyeeTestError()))
yield d
@ee.on("error")
def handle_error(exc):
assert isinstance(exc, Exception)
should_call(exc)
ee.emit("event")
should_call.assert_called_once()
|
