-
Notifications
You must be signed in to change notification settings - Fork 116
Expand file tree
/
Copy pathtest_dispatcher.py
More file actions
246 lines (197 loc) · 8.32 KB
/
Copy pathtest_dispatcher.py
File metadata and controls
246 lines (197 loc) · 8.32 KB
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
import unittest
from pythonosc.dispatcher import Dispatcher, Handler
class TestDispatcher(unittest.IsolatedAsyncioTestCase):
def setUp(self):
super().setUp()
self.dispatcher = Dispatcher()
def sortAndAssertSequenceEqual(self, expected, result):
def sort(lst):
return sorted(lst, key=lambda x: x.callback)
return self.assertSequenceEqual(sort(expected), sort(result))
def test_empty_by_default(self):
self.sortAndAssertSequenceEqual(
[], self.dispatcher.handlers_for_address("/test")
)
def test_use_default_handler_when_set_and_no_match(self):
handler = object()
self.dispatcher.set_default_handler(handler)
self.sortAndAssertSequenceEqual(
[Handler(handler, [])], self.dispatcher.handlers_for_address("/test")
)
def test_simple_map_and_match(self):
handler = object()
self.dispatcher.map("/test", handler, 1, 2, 3)
self.dispatcher.map("/test2", handler)
self.sortAndAssertSequenceEqual(
[Handler(handler, [1, 2, 3])], self.dispatcher.handlers_for_address("/test")
)
self.sortAndAssertSequenceEqual(
[Handler(handler, [])], self.dispatcher.handlers_for_address("/test2")
)
def test_example_from_spec(self):
addresses = [
"/first/this/one",
"/second/1",
"/second/2",
"/third/a",
"/third/b",
"/third/c",
]
for index, address in enumerate(addresses):
self.dispatcher.map(address, index)
for index, address in enumerate(addresses):
self.sortAndAssertSequenceEqual(
[Handler(index, [])], self.dispatcher.handlers_for_address(address)
)
self.sortAndAssertSequenceEqual(
[Handler(1, []), Handler(2, [])],
self.dispatcher.handlers_for_address("/second/?"),
)
self.sortAndAssertSequenceEqual(
[Handler(3, []), Handler(4, []), Handler(5, [])],
self.dispatcher.handlers_for_address("/third/*"),
)
def test_do_not_match_over_slash(self):
self.dispatcher.map("/foo/bar/1", 1)
self.dispatcher.map("/foo/bar/2", 2)
self.sortAndAssertSequenceEqual([], self.dispatcher.handlers_for_address("/*"))
def test_match_middle_star(self):
self.dispatcher.map("/foo/bar/1", 1)
self.dispatcher.map("/foo/bar/2", 2)
self.sortAndAssertSequenceEqual(
[Handler(2, [])], self.dispatcher.handlers_for_address("/foo/*/2")
)
def test_match_multiple_stars(self):
self.dispatcher.map("/foo/bar/1", 1)
self.dispatcher.map("/foo/bar/2", 2)
self.sortAndAssertSequenceEqual(
[Handler(1, []), Handler(2, [])],
self.dispatcher.handlers_for_address("/*/*/*"),
)
def test_match_address_contains_plus_as_character(self):
self.dispatcher.map("/footest/bar+tender/1", 1)
self.sortAndAssertSequenceEqual(
[Handler(1, [])], self.dispatcher.handlers_for_address("/foo*/bar+*/*")
)
self.sortAndAssertSequenceEqual(
[Handler(1, [])], self.dispatcher.handlers_for_address("/foo*/bar*/*")
)
def test_call_correct_dispatcher_on_star(self):
self.dispatcher.map("/a+b", 1)
self.dispatcher.map("/aaab", 2)
self.sortAndAssertSequenceEqual(
[Handler(2, [])], self.dispatcher.handlers_for_address("/aaab")
)
self.sortAndAssertSequenceEqual(
[Handler(1, [])], self.dispatcher.handlers_for_address("/a+b")
)
def test_map_star(self):
self.dispatcher.map("/starbase/*", 1)
self.sortAndAssertSequenceEqual(
[Handler(1, [])], self.dispatcher.handlers_for_address("/starbase/bar")
)
def test_map_root_star(self):
self.dispatcher.map("/*", 1)
self.sortAndAssertSequenceEqual(
[Handler(1, [])], self.dispatcher.handlers_for_address("/anything/matches")
)
def test_map_double_stars(self):
self.dispatcher.map("/foo/*/bar/*", 1)
self.sortAndAssertSequenceEqual(
[Handler(1, [])], self.dispatcher.handlers_for_address("/foo/wild/bar/wild")
)
self.sortAndAssertSequenceEqual(
[], self.dispatcher.handlers_for_address("/foo/wild/nomatch/wild")
)
def test_multiple_handlers(self):
self.dispatcher.map("/foo/bar", 1)
self.dispatcher.map("/foo/bar", 2)
self.sortAndAssertSequenceEqual(
[Handler(1, []), Handler(2, [])],
self.dispatcher.handlers_for_address("/foo/bar"),
)
def test_multiple_handlers_with_wildcard_map(self):
self.dispatcher.map("/foo/bar", 1)
self.dispatcher.map("/*", 2)
self.sortAndAssertSequenceEqual(
[Handler(1, []), Handler(2, [])],
self.dispatcher.handlers_for_address("/foo/bar"),
)
def test_unmap(self):
def dummyhandler():
pass
# Test with handler returned by map
returnedhandler = self.dispatcher.map("/map/me", dummyhandler)
self.sortAndAssertSequenceEqual(
[Handler(dummyhandler, [])], self.dispatcher.handlers_for_address("/map/me")
)
self.dispatcher.unmap("/map/me", returnedhandler)
self.sortAndAssertSequenceEqual(
[], self.dispatcher.handlers_for_address("/map/me")
)
# Test with reconstructing handler
self.dispatcher.map("/map/me/too", dummyhandler)
self.sortAndAssertSequenceEqual(
[Handler(dummyhandler, [])],
self.dispatcher.handlers_for_address("/map/me/too"),
)
self.dispatcher.unmap("/map/me/too", dummyhandler)
self.sortAndAssertSequenceEqual(
[], self.dispatcher.handlers_for_address("/map/me/too")
)
def test_unmap_exception(self):
def dummyhandler():
pass
with self.assertRaises(ValueError):
self.dispatcher.unmap("/unmap/exception", dummyhandler)
handlerobj = self.dispatcher.map("/unmap/somethingelse", dummyhandler())
with self.assertRaises(ValueError):
self.dispatcher.unmap("/unmap/exception", handlerobj)
def test_handlers_for_address_wildcard_no_partial_match(self):
self.dispatcher.map("/qwer/*/zxcv", 1)
# Should not match
handlers = list(
self.dispatcher.handlers_for_address("/qwer/whatever/zxcvsomethingmore")
)
self.assertEqual(len(handlers), 0)
# Should match
handlers = list(self.dispatcher.handlers_for_address("/qwer/whatever/zxcv"))
self.assertEqual(len(handlers), 1)
def test_strict_timing_disabled(self):
# Disable strict timing
dispatcher = Dispatcher(strict_timing=False)
callback_called = False
def handler(address, *args):
nonlocal callback_called
callback_called = True
dispatcher.map("/test", handler)
# Create a message with a future timestamp (1 hour from now)
# We'll use OscPacket to simulate a bundle with a future timestamp
# But for simple unit test, we can just check if it sleeps
# Since we can't easily mock time.sleep across the dispatcher without more effort,
# we'll just verify the logic exists.
self.assertFalse(dispatcher._strict_timing)
async def test_async_call_handlers_for_packet(self):
dispatcher = Dispatcher()
sync_called = False
def sync_handler(address, *args):
nonlocal sync_called
sync_called = True
async_called = False
async def async_handler(address, *args):
nonlocal async_called
async_called = True
dispatcher.map("/sync", sync_handler)
dispatcher.map("/async", async_handler)
# Dispatch sync handler
dgram_sync = b"/sync\x00\x00\x00,\x00\x00\x00"
await dispatcher.async_call_handlers_for_packet(dgram_sync, ("127.0.0.1", 1234))
self.assertTrue(sync_called)
# Dispatch async handler
dgram_async = b"/async\x00\x00,\x00\x00\x00"
await dispatcher.async_call_handlers_for_packet(
dgram_async, ("127.0.0.1", 1234)
)
self.assertTrue(async_called)
if __name__ == "__main__":
unittest.main()