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
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
|
# Authors: The MNE-Python contributors.
# License: BSD-3-Clause
# Copyright the MNE-Python contributors.
import matplotlib.pyplot as plt
import pytest
from mne.datasets import testing
from mne.viz import ui_events
subjects_dir = testing.data_path(download=False) / "subjects"
@pytest.fixture
def event_channels():
"""Fixture that makes sure each test starts with a fresh UI event chans dict."""
ui_events._event_channels.clear()
return ui_events._event_channels
@pytest.fixture
def event_channel_links():
"""Fixture that makes sure each test starts with a fresh channel links dict."""
ui_events._event_channel_links.clear()
return ui_events._event_channel_links
@pytest.fixture
def disabled_event_channels():
"""Fixture that makes sure each test starts with a fresh disabled channels set."""
ui_events._disabled_event_channels.clear()
return ui_events._disabled_event_channels
@testing.requires_testing_data
def test_get_event_channel(event_channels):
"""Test creating and obtaining a figure's UI event channel."""
# At first, no event channels exist
assert len(event_channels) == 0
# Open a figure and get the event channel. This should create it.
fig = plt.figure()
ui_events._get_event_channel(fig)
assert len(event_channels) == 1
assert fig in event_channels
# Closing a figure should delete the event channel.
# During tests, matplotlib does not open an actual window so we need to force the
# close event.
fig.canvas.callbacks.process("close_event", None)
assert len(event_channels) == 0
def test_publish(event_channels):
"""Test publishing UI events."""
fig = plt.figure()
ui_events.publish(fig, ui_events.TimeChange(time=10.2))
# Publishing the event should have created the needed channel.
assert len(event_channels) == 1
assert fig in event_channels
def test_subscribe(event_channels):
"""Test subscribing to UI events."""
callback_calls = list()
def callback(event):
"""Respond to time change event."""
callback_calls.append(event)
assert isinstance(event, ui_events.TimeChange)
assert event.time == 10.2
fig = plt.figure()
ui_events.subscribe(fig, "time_change", callback)
# Subscribing to the event should have created the needed channel.
assert "time_change" in ui_events._get_event_channel(fig)
# Publishing the time change event should call the callback function.
ui_events.publish(fig, ui_events.TimeChange(time=10.2))
assert callback_calls
# Publishing a different event should not call the callback function.
callback_calls.clear() # Reset
ui_events.publish(fig, ui_events.FigureClosing())
assert not callback_calls
# Test disposing of the event channel, even with subscribers.
# During tests, matplotlib does not open an actual window so we need to force the
# close event.
fig.canvas.callbacks.process("close_event", None)
assert len(event_channels) == 0
def test_unsubscribe(event_channels):
"""Test unsubscribing from UI events."""
callback1_calls = list()
callback2_calls = list()
def callback1(event):
"""Respond to time change event."""
callback1_calls.append(event)
def callback2(event):
"""Respond to time change event."""
callback2_calls.append(event)
fig = plt.figure()
def setup_events():
"""Reset UI event scenario."""
callback1_calls.clear()
callback2_calls.clear()
ui_events.unsubscribe(fig, "all")
ui_events.subscribe(fig, "figure_closing", callback1)
ui_events.subscribe(fig, "time_change", callback1)
ui_events.subscribe(fig, "time_change", callback2)
# Test unsubscribing from a single event
setup_events()
with pytest.warns(RuntimeWarning, match="Cannot unsubscribe"):
ui_events.unsubscribe(fig, "nonexisting_event")
ui_events.unsubscribe(fig, "time_change")
assert "time_change" not in ui_events._get_event_channel(fig)
assert "figure_closing" in ui_events._get_event_channel(fig)
ui_events.publish(fig, ui_events.TimeChange(time=10.2))
assert not callback1_calls
assert not callback2_calls
ui_events.publish(fig, ui_events.FigureClosing())
assert callback1_calls
# Test unsubscribing from all events
setup_events()
ui_events.unsubscribe(fig, "all")
assert "time_change" not in ui_events._get_event_channel(fig)
assert "figure_closing" not in ui_events._get_event_channel(fig)
ui_events.publish(fig, ui_events.TimeChange(time=10.2))
ui_events.publish(fig, ui_events.FigureClosing())
assert not callback1_calls
assert not callback2_calls
# Test unsubscribing from a list of events
setup_events()
ui_events.unsubscribe(fig, ["time_change", "figure_closing"])
assert "time_change" not in ui_events._get_event_channel(fig)
assert "figure_closing" not in ui_events._get_event_channel(fig)
ui_events.publish(fig, ui_events.TimeChange(time=10.2))
ui_events.publish(fig, ui_events.FigureClosing())
assert not callback1_calls
assert not callback2_calls
# Test unsubscribing a specific callback function from a single event
setup_events()
with pytest.warns(RuntimeWarning, match="Cannot unsubscribe"):
ui_events.unsubscribe(fig, "figure_closing", callback2)
ui_events.unsubscribe(fig, "time_change", callback2)
ui_events.publish(fig, ui_events.TimeChange(time=10.2))
assert callback1_calls
assert not callback2_calls
# Test unsubscribing a specific callback function from all events
setup_events()
ui_events.unsubscribe(fig, "all", callback2)
ui_events.publish(fig, ui_events.TimeChange(time=10.2))
assert callback1_calls
assert not callback2_calls
# Test unsubscribing a specific callback function from a list of events
setup_events()
ui_events.unsubscribe(fig, ["time_change", "figure_closing"], callback1)
ui_events.publish(fig, ui_events.TimeChange(time=10.2))
ui_events.publish(fig, ui_events.FigureClosing())
assert not callback1_calls
def test_link(event_channels, event_channel_links):
"""Test linking the event channels of two functions."""
fig1 = plt.figure()
fig2 = plt.figure()
fig3 = plt.figure()
callback_calls = list()
def callback(event):
"""Respond to time change event."""
callback_calls.append(event)
# Both figures are subscribed to the time change events.
ui_events.subscribe(fig1, "time_change", callback)
ui_events.subscribe(fig2, "time_change", callback)
ui_events.subscribe(fig3, "time_change", callback)
# Linking the event channels causes events to be published on both channels.
ui_events.link(fig1, fig2, fig3)
assert len(event_channel_links) == 3
assert fig2 in event_channel_links[fig1]
assert fig3 in event_channel_links[fig1]
assert fig1 in event_channel_links[fig2]
assert fig3 in event_channel_links[fig2]
assert fig1 in event_channel_links[fig3]
assert fig2 in event_channel_links[fig3]
ui_events.publish(fig1, ui_events.TimeChange(time=10.2))
assert len(callback_calls) == 3
callback_calls.clear()
ui_events.publish(fig2, ui_events.TimeChange(time=10.2))
assert len(callback_calls) == 3
callback_calls.clear()
ui_events.publish(fig3, ui_events.TimeChange(time=10.2))
assert len(callback_calls) == 3
# Closing this should remove the links as well
fig3.canvas.callbacks.process("close_event", None)
# Test linking only specific events
ui_events.link(fig1, fig2, include_events=["time_change"])
callback_calls.clear()
ui_events.publish(fig1, ui_events.TimeChange(time=10.2))
ui_events.publish(fig2, ui_events.TimeChange(time=10.2))
assert len(callback_calls) == 4 # Called for both figures two times
ui_events.link(fig1, fig2, include_events=["some_other_event"])
callback_calls.clear()
ui_events.publish(fig1, ui_events.TimeChange(time=10.2))
ui_events.publish(fig2, ui_events.TimeChange(time=10.2))
assert len(callback_calls) == 2 # Only called for both figures once
ui_events.link(fig1, fig2, exclude_events=["time_change"])
callback_calls.clear()
ui_events.publish(fig1, ui_events.TimeChange(time=10.2))
ui_events.publish(fig2, ui_events.TimeChange(time=10.2))
assert len(callback_calls) == 2 # Only called for both figures once
# Test cleanup
fig1.canvas.callbacks.process("close_event", None)
fig2.canvas.callbacks.process("close_event", None)
assert len(event_channels) == 0
assert len(event_channel_links) == 0
def test_unlink(event_channel_links):
"""Test unlinking event channels."""
fig1 = plt.figure()
fig2 = plt.figure()
fig3 = plt.figure()
ui_events.link(fig1, fig2)
ui_events.link(fig2, fig3)
assert len(event_channel_links) == 3
# Fig1 is involved in two of the 4 links.
ui_events.unlink(fig1)
assert len(event_channel_links) == 2
assert fig1 not in event_channel_links[fig2]
assert fig1 not in event_channel_links[fig3]
ui_events.link(fig1, fig2) # Relink for the next test.
# Fig2 is involved in all links, unlinking it should clear them all.
ui_events.unlink(fig2)
assert len(event_channel_links) == 0
def test_disable_ui_events(event_channels, disabled_event_channels):
"""Test disable_ui_events context manager."""
callback_calls = list()
def callback(event):
"""Respond to time change event."""
callback_calls.append(event)
fig = plt.figure()
ui_events.subscribe(fig, "time_change", callback)
with ui_events.disable_ui_events(fig):
ui_events.publish(fig, ui_events.TimeChange(time=10.2))
assert not callback_calls
ui_events.publish(fig, ui_events.TimeChange(time=10.2))
assert callback_calls
|
