1# -*- coding: utf-8 -*-
2from collections import OrderedDict
3
4from mock import Mock
5from pytest import raises
6
7from pyee import EventEmitter
8
9
10class PyeeTestException(Exception):
11    pass
12
13
14def test_emit_sync():
15    """Basic synchronous emission works"""
16
17    call_me = Mock()
18    ee = EventEmitter()
19
20    @ee.on("event")
21    def event_handler(data, **kwargs):
22        call_me()
23        assert data == "emitter is emitted!"
24
25    assert ee.event_names() == {"event"}
26
27    # Making sure data is passed propers
28    ee.emit("event", "emitter is emitted!", error=False)
29
30    call_me.assert_called_once()
31
32
33def test_emit_error():
34    """Errors raise with no event handler, otherwise emit on handler"""
35
36    call_me = Mock()
37    ee = EventEmitter()
38
39    test_exception = PyeeTestException("lololol")
40
41    with raises(PyeeTestException):
42        ee.emit("error", test_exception)
43
44    @ee.on("error")
45    def on_error(exc):
46        call_me()
47
48    assert ee.event_names() == {"error"}
49
50    # No longer raises and error instead return True indicating handled
51    assert ee.emit("error", test_exception) is True
52    call_me.assert_called_once()
53
54
55def test_emit_return():
56    """Emit returns True when handlers are registered on an event, and false
57    otherwise.
58    """
59
60    call_me = Mock()
61    ee = EventEmitter()
62
63    assert ee.event_names() == set()
64
65    # make sure emitting without a callback returns False
66    assert not ee.emit("data")
67
68    # add a callback
69    ee.on("data")(call_me)
70
71    # should return True now
72    assert ee.emit("data")
73
74
75def test_new_listener_event():
76    """The 'new_listener' event fires whenever a new listener is added."""
77
78    call_me = Mock()
79    ee = EventEmitter()
80
81    ee.on("new_listener", call_me)
82
83    # Should fire new_listener event
84    @ee.on("event")
85    def event_handler(data):
86        pass
87
88    assert ee.event_names() == {"new_listener", "event"}
89
90    call_me.assert_called_once_with("event", event_handler)
91
92
93def test_listener_removal():
94    """Removing listeners removes the correct listener from an event."""
95
96    ee = EventEmitter()
97
98    # Some functions to pass to the EE
99    def first():
100        return 1
101
102    ee.on("event", first)
103
104    @ee.on("event")
105    def second():
106        return 2
107
108    @ee.on("event")
109    def third():
110        return 3
111
112    def fourth():
113        return 4
114
115    ee.on("event", fourth)
116
117    assert ee.event_names() == {"event"}
118
119    assert ee._events["event"] == OrderedDict(
120        [(first, first), (second, second), (third, third), (fourth, fourth)]
121    )
122
123    ee.remove_listener("event", second)
124
125    assert ee._events["event"] == OrderedDict(
126        [(first, first), (third, third), (fourth, fourth)]
127    )
128
129    ee.remove_listener("event", first)
130    assert ee._events["event"] == OrderedDict([(third, third), (fourth, fourth)])
131
132    ee.remove_all_listeners("event")
133    assert "event" not in ee._events["event"]
134
135
136def test_listener_removal_on_emit():
137    """Test that a listener removed during an emit is called inside the current
138    emit cycle.
139    """
140
141    call_me = Mock()
142    ee = EventEmitter()
143
144    def should_remove():
145        ee.remove_listener("remove", call_me)
146
147    ee.on("remove", should_remove)
148    ee.on("remove", call_me)
149
150    assert ee.event_names() == {"remove"}
151
152    ee.emit("remove")
153
154    call_me.assert_called_once()
155
156    call_me.reset_mock()
157
158    # Also test with the listeners added in the opposite order
159    ee = EventEmitter()
160    ee.on("remove", call_me)
161    ee.on("remove", should_remove)
162
163    assert ee.event_names() == {"remove"}
164
165    ee.emit("remove")
166
167    call_me.assert_called_once()
168
169
170def test_once():
171    """Test that `once()` method works propers."""
172
173    # very similar to "test_emit" but also makes sure that the event
174    # gets removed afterwards
175
176    call_me = Mock()
177    ee = EventEmitter()
178
179    def once_handler(data):
180        assert data == "emitter is emitted!"
181        call_me()
182
183    # Tests to make sure that after event is emitted that it's gone.
184    ee.once("event", once_handler)
185
186    assert ee.event_names() == {"event"}
187
188    ee.emit("event", "emitter is emitted!")
189
190    call_me.assert_called_once()
191
192    assert ee.event_names() == set()
193
194    assert "event" not in ee._events
195
196
197def test_once_removal():
198    """Removal of once functions works"""
199
200    ee = EventEmitter()
201
202    def once_handler(data):
203        pass
204
205    handle = ee.once("event", once_handler)
206
207    assert handle == once_handler
208    assert ee.event_names() == {"event"}
209
210    ee.remove_listener("event", handle)
211
212    assert "event" not in ee._events
213    assert ee.event_names() == set()
214
215
216def test_listeners():
217    """`listeners()` returns a copied list of listeners."""
218
219    call_me = Mock()
220    ee = EventEmitter()
221
222    @ee.on("event")
223    def event_handler():
224        pass
225
226    @ee.once("event")
227    def once_handler():
228        pass
229
230    listeners = ee.listeners("event")
231
232    assert listeners[0] == event_handler
233    assert listeners[1] == once_handler
234
235    # listeners is a copy, you can't mutate the innards this way
236    listeners[0] = call_me
237
238    ee.emit("event")
239
240    call_me.assert_not_called()
241
242
243def test_listeners_does_work_with_unknown_listeners():
244    """`listeners()` should not throw."""
245    ee = EventEmitter()
246    listeners = ee.listeners("event")
247    assert listeners == []
248
249
250def test_properties_preserved():
251    """Test that the properties of decorated functions are preserved."""
252
253    call_me = Mock()
254    call_me_also = Mock()
255    ee = EventEmitter()
256
257    @ee.on("always")
258    def always_event_handler():
259        """An event handler."""
260        call_me()
261
262    @ee.once("once")
263    def once_event_handler():
264        """Another event handler."""
265        call_me_also()
266
267    assert always_event_handler.__doc__ == "An event handler."
268    assert once_event_handler.__doc__ == "Another event handler."
269
270    always_event_handler()
271    call_me.assert_called_once()
272
273    once_event_handler()
274    call_me_also.assert_called_once()
275
276    call_me_also.reset_mock()
277
278    # Calling the event handler directly doesn't clear the handler
279    ee.emit("once")
280    call_me_also.assert_called_once()
281