Commit | Line | Data |
---|---|---|
0235b0db | 1 | # SPDX-License-Identifier: GPL-2.0-only |
d2d857a8 MJ |
2 | # |
3 | # Copyright (C) 2019 EfficiOS Inc. | |
4 | # | |
d2d857a8 | 5 | |
811644b8 | 6 | import unittest |
811644b8 | 7 | import bt2 |
5813b3a3 | 8 | from bt2 import port as bt2_port |
811644b8 PP |
9 | |
10 | ||
11 | class PortTestCase(unittest.TestCase): | |
12 | @staticmethod | |
13 | def _create_comp(comp_cls, name=None): | |
14 | graph = bt2.Graph() | |
15 | ||
16 | if name is None: | |
17 | name = 'comp' | |
18 | ||
19 | return graph.add_component(comp_cls, name) | |
20 | ||
21 | def test_src_add_output_port(self): | |
5c61fb9d SM |
22 | class MySource( |
23 | bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator | |
24 | ): | |
59225a3e | 25 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
26 | port = comp_self._add_output_port('out') |
27 | self.assertEqual(port.name, 'out') | |
28 | ||
29 | comp = self._create_comp(MySource) | |
30 | self.assertEqual(len(comp.output_ports), 1) | |
5813b3a3 | 31 | self.assertIs(type(comp.output_ports['out']), bt2_port._OutputPortConst) |
811644b8 | 32 | |
157a98ed SM |
33 | # Test adding output port with duplicate name to source. |
34 | def test_src_add_output_port_dup_name_raises(self): | |
35 | class MySource( | |
36 | bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator | |
37 | ): | |
38 | def __init__(comp_self, config, params, obj): | |
39 | comp_self._add_output_port('out') | |
40 | ||
41 | with self.assertRaisesRegex( | |
42 | ValueError, | |
43 | "source component `comp` already contains an output port named `out`", | |
44 | ): | |
45 | comp_self._add_output_port('out') | |
46 | ||
47 | nonlocal seen | |
48 | seen = True | |
49 | ||
50 | seen = False | |
51 | self._create_comp(MySource) | |
52 | self.assertTrue(seen) | |
53 | ||
811644b8 | 54 | def test_flt_add_output_port(self): |
5c61fb9d SM |
55 | class MyFilter( |
56 | bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator | |
57 | ): | |
59225a3e | 58 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
59 | port = comp_self._add_output_port('out') |
60 | self.assertEqual(port.name, 'out') | |
61 | ||
62 | comp = self._create_comp(MyFilter) | |
63 | self.assertEqual(len(comp.output_ports), 1) | |
64 | ||
157a98ed SM |
65 | # Test adding output port with duplicate name to filter. |
66 | def test_flt_add_output_port_dup_name_raises(self): | |
67 | class MyFilter( | |
68 | bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator | |
69 | ): | |
70 | def __init__(comp_self, config, params, obj): | |
71 | comp_self._add_output_port('out') | |
72 | ||
73 | with self.assertRaisesRegex( | |
74 | ValueError, | |
75 | "filter component `comp` already contains an output port named `out`", | |
76 | ): | |
77 | comp_self._add_output_port('out') | |
78 | ||
79 | nonlocal seen | |
80 | seen = True | |
81 | ||
82 | seen = False | |
83 | self._create_comp(MyFilter) | |
84 | self.assertTrue(seen) | |
85 | ||
811644b8 | 86 | def test_flt_add_input_port(self): |
5c61fb9d SM |
87 | class MyFilter( |
88 | bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator | |
89 | ): | |
59225a3e | 90 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
91 | port = comp_self._add_input_port('in') |
92 | self.assertEqual(port.name, 'in') | |
93 | ||
94 | comp = self._create_comp(MyFilter) | |
95 | self.assertEqual(len(comp.input_ports), 1) | |
5813b3a3 | 96 | self.assertIs(type(comp.input_ports['in']), bt2_port._InputPortConst) |
811644b8 | 97 | |
157a98ed SM |
98 | # Test adding input port with duplicate name to filter. |
99 | def test_flt_add_input_port_dup_name_raises(self): | |
100 | class MyFilter( | |
101 | bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator | |
102 | ): | |
103 | def __init__(comp_self, config, params, obj): | |
104 | comp_self._add_input_port('in') | |
105 | ||
106 | with self.assertRaisesRegex( | |
107 | ValueError, | |
108 | "filter component `comp` already contains an input port named `in`", | |
109 | ): | |
110 | comp_self._add_input_port('in') | |
111 | ||
112 | nonlocal seen | |
113 | seen = True | |
114 | ||
115 | seen = False | |
116 | self._create_comp(MyFilter) | |
117 | self.assertTrue(seen) | |
118 | ||
811644b8 PP |
119 | def test_sink_add_input_port(self): |
120 | class MySink(bt2._UserSinkComponent): | |
59225a3e | 121 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
122 | port = comp_self._add_input_port('in') |
123 | self.assertEqual(port.name, 'in') | |
124 | ||
6a91742b | 125 | def _user_consume(self): |
a01b452b SM |
126 | pass |
127 | ||
811644b8 PP |
128 | comp = self._create_comp(MySink) |
129 | self.assertEqual(len(comp.input_ports), 1) | |
130 | ||
157a98ed SM |
131 | # Test adding input port with duplicate name to sink. |
132 | def test_sink_add_input_port_dup_name_raises(self): | |
133 | class MySink(bt2._UserSinkComponent): | |
134 | def __init__(comp_self, config, params, obj): | |
135 | comp_self._add_input_port('in') | |
136 | ||
137 | with self.assertRaisesRegex( | |
138 | ValueError, | |
139 | "sink component `comp` already contains an input port named `in`", | |
140 | ): | |
141 | comp_self._add_input_port('in') | |
142 | ||
143 | nonlocal seen | |
144 | seen = True | |
145 | ||
146 | def _user_consume(self): | |
147 | pass | |
148 | ||
149 | seen = False | |
150 | self._create_comp(MySink) | |
151 | self.assertTrue(seen) | |
152 | ||
811644b8 | 153 | def test_user_src_output_ports_getitem(self): |
5c61fb9d SM |
154 | class MySource( |
155 | bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator | |
156 | ): | |
59225a3e | 157 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
158 | port1 = comp_self._add_output_port('clear') |
159 | port2 = comp_self._add_output_port('print') | |
160 | port3 = comp_self._add_output_port('insert') | |
161 | self.assertEqual(port3.addr, comp_self._output_ports['insert'].addr) | |
162 | self.assertEqual(port2.addr, comp_self._output_ports['print'].addr) | |
163 | self.assertEqual(port1.addr, comp_self._output_ports['clear'].addr) | |
164 | ||
894a8df5 | 165 | self._create_comp(MySource) |
811644b8 PP |
166 | |
167 | def test_user_flt_output_ports_getitem(self): | |
5c61fb9d SM |
168 | class MyFilter( |
169 | bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator | |
170 | ): | |
59225a3e | 171 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
172 | port1 = comp_self._add_output_port('clear') |
173 | port2 = comp_self._add_output_port('print') | |
174 | port3 = comp_self._add_output_port('insert') | |
175 | self.assertEqual(port3.addr, comp_self._output_ports['insert'].addr) | |
176 | self.assertEqual(port2.addr, comp_self._output_ports['print'].addr) | |
177 | self.assertEqual(port1.addr, comp_self._output_ports['clear'].addr) | |
178 | ||
894a8df5 | 179 | self._create_comp(MyFilter) |
811644b8 PP |
180 | |
181 | def test_user_flt_input_ports_getitem(self): | |
5c61fb9d SM |
182 | class MyFilter( |
183 | bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator | |
184 | ): | |
59225a3e | 185 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
186 | port1 = comp_self._add_input_port('clear') |
187 | port2 = comp_self._add_input_port('print') | |
188 | port3 = comp_self._add_input_port('insert') | |
189 | self.assertEqual(port3.addr, comp_self._input_ports['insert'].addr) | |
190 | self.assertEqual(port2.addr, comp_self._input_ports['print'].addr) | |
191 | self.assertEqual(port1.addr, comp_self._input_ports['clear'].addr) | |
192 | ||
894a8df5 | 193 | self._create_comp(MyFilter) |
811644b8 PP |
194 | |
195 | def test_user_sink_input_ports_getitem(self): | |
196 | class MySink(bt2._UserSinkComponent): | |
59225a3e | 197 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
198 | port1 = comp_self._add_input_port('clear') |
199 | port2 = comp_self._add_input_port('print') | |
200 | port3 = comp_self._add_input_port('insert') | |
201 | self.assertEqual(port3.addr, comp_self._input_ports['insert'].addr) | |
202 | self.assertEqual(port2.addr, comp_self._input_ports['print'].addr) | |
203 | self.assertEqual(port1.addr, comp_self._input_ports['clear'].addr) | |
204 | ||
6a91742b | 205 | def _user_consume(self): |
a01b452b SM |
206 | pass |
207 | ||
894a8df5 | 208 | self._create_comp(MySink) |
811644b8 PP |
209 | |
210 | def test_user_src_output_ports_getitem_invalid_key(self): | |
5c61fb9d SM |
211 | class MySource( |
212 | bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator | |
213 | ): | |
59225a3e | 214 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
215 | comp_self._add_output_port('clear') |
216 | comp_self._add_output_port('print') | |
217 | comp_self._add_output_port('insert') | |
218 | ||
219 | with self.assertRaises(KeyError): | |
220 | comp_self._output_ports['hello'] | |
221 | ||
894a8df5 | 222 | self._create_comp(MySource) |
811644b8 PP |
223 | |
224 | def test_user_flt_output_ports_getitem_invalid_key(self): | |
5c61fb9d SM |
225 | class MyFilter( |
226 | bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator | |
227 | ): | |
59225a3e | 228 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
229 | comp_self._add_output_port('clear') |
230 | comp_self._add_output_port('print') | |
231 | comp_self._add_output_port('insert') | |
232 | ||
233 | with self.assertRaises(KeyError): | |
234 | comp_self._output_ports['hello'] | |
235 | ||
894a8df5 | 236 | self._create_comp(MyFilter) |
811644b8 PP |
237 | |
238 | def test_user_flt_input_ports_getitem_invalid_key(self): | |
5c61fb9d SM |
239 | class MyFilter( |
240 | bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator | |
241 | ): | |
59225a3e | 242 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
243 | comp_self._add_input_port('clear') |
244 | comp_self._add_input_port('print') | |
245 | comp_self._add_input_port('insert') | |
246 | ||
247 | with self.assertRaises(KeyError): | |
248 | comp_self._input_ports['hello'] | |
249 | ||
894a8df5 | 250 | self._create_comp(MyFilter) |
811644b8 PP |
251 | |
252 | def test_user_sink_input_ports_getitem_invalid_key(self): | |
253 | class MySink(bt2._UserSinkComponent): | |
59225a3e | 254 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
255 | comp_self._add_input_port('clear') |
256 | comp_self._add_input_port('print') | |
257 | comp_self._add_input_port('insert') | |
258 | ||
259 | with self.assertRaises(KeyError): | |
260 | comp_self._input_ports['hello'] | |
261 | ||
6a91742b | 262 | def _user_consume(self): |
a01b452b SM |
263 | pass |
264 | ||
894a8df5 | 265 | self._create_comp(MySink) |
811644b8 PP |
266 | |
267 | def test_user_src_output_ports_len(self): | |
5c61fb9d SM |
268 | class MySource( |
269 | bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator | |
270 | ): | |
59225a3e | 271 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
272 | comp_self._add_output_port('clear') |
273 | comp_self._add_output_port('print') | |
274 | comp_self._add_output_port('insert') | |
275 | self.assertEqual(len(comp_self._output_ports), 3) | |
276 | ||
894a8df5 | 277 | self._create_comp(MySource) |
811644b8 PP |
278 | |
279 | def test_user_flt_output_ports_len(self): | |
5c61fb9d SM |
280 | class MyFilter( |
281 | bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator | |
282 | ): | |
59225a3e | 283 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
284 | comp_self._add_output_port('clear') |
285 | comp_self._add_output_port('print') | |
286 | comp_self._add_output_port('insert') | |
287 | self.assertEqual(len(comp_self._output_ports), 3) | |
288 | ||
894a8df5 | 289 | self._create_comp(MyFilter) |
811644b8 PP |
290 | |
291 | def test_user_flt_input_ports_len(self): | |
5c61fb9d SM |
292 | class MyFilter( |
293 | bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator | |
294 | ): | |
59225a3e | 295 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
296 | comp_self._add_input_port('clear') |
297 | comp_self._add_input_port('print') | |
298 | comp_self._add_input_port('insert') | |
299 | self.assertEqual(len(comp_self._input_ports), 3) | |
300 | ||
894a8df5 | 301 | self._create_comp(MyFilter) |
811644b8 PP |
302 | |
303 | def test_user_sink_input_ports_len(self): | |
304 | class MySink(bt2._UserSinkComponent): | |
59225a3e | 305 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
306 | comp_self._add_input_port('clear') |
307 | comp_self._add_input_port('print') | |
308 | comp_self._add_input_port('insert') | |
309 | self.assertEqual(len(comp_self._input_ports), 3) | |
310 | ||
6a91742b | 311 | def _user_consume(self): |
a01b452b SM |
312 | pass |
313 | ||
894a8df5 | 314 | self._create_comp(MySink) |
811644b8 PP |
315 | |
316 | def test_user_src_output_ports_iter(self): | |
5c61fb9d SM |
317 | class MySource( |
318 | bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator | |
319 | ): | |
59225a3e | 320 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
321 | port1 = comp_self._add_output_port('clear') |
322 | port2 = comp_self._add_output_port('print') | |
323 | port3 = comp_self._add_output_port('insert') | |
324 | ports = [] | |
325 | ||
326 | for port_name, port in comp_self._output_ports.items(): | |
327 | ports.append((port_name, port)) | |
328 | ||
329 | self.assertEqual(ports[0][0], 'clear') | |
330 | self.assertEqual(ports[0][1].addr, port1.addr) | |
331 | self.assertEqual(ports[1][0], 'print') | |
332 | self.assertEqual(ports[1][1].addr, port2.addr) | |
333 | self.assertEqual(ports[2][0], 'insert') | |
334 | self.assertEqual(ports[2][1].addr, port3.addr) | |
335 | ||
894a8df5 | 336 | self._create_comp(MySource) |
811644b8 PP |
337 | |
338 | def test_user_flt_output_ports_iter(self): | |
5c61fb9d SM |
339 | class MyFilter( |
340 | bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator | |
341 | ): | |
59225a3e | 342 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
343 | port1 = comp_self._add_output_port('clear') |
344 | port2 = comp_self._add_output_port('print') | |
345 | port3 = comp_self._add_output_port('insert') | |
346 | ports = [] | |
347 | ||
348 | for port_name, port in comp_self._output_ports.items(): | |
349 | ports.append((port_name, port)) | |
350 | ||
351 | self.assertEqual(ports[0][0], 'clear') | |
352 | self.assertEqual(ports[0][1].addr, port1.addr) | |
353 | self.assertEqual(ports[1][0], 'print') | |
354 | self.assertEqual(ports[1][1].addr, port2.addr) | |
355 | self.assertEqual(ports[2][0], 'insert') | |
356 | self.assertEqual(ports[2][1].addr, port3.addr) | |
357 | ||
894a8df5 | 358 | self._create_comp(MyFilter) |
811644b8 PP |
359 | |
360 | def test_user_flt_input_ports_iter(self): | |
5c61fb9d SM |
361 | class MyFilter( |
362 | bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator | |
363 | ): | |
59225a3e | 364 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
365 | port1 = comp_self._add_input_port('clear') |
366 | port2 = comp_self._add_input_port('print') | |
367 | port3 = comp_self._add_input_port('insert') | |
368 | ports = [] | |
369 | ||
370 | for port_name, port in comp_self._input_ports.items(): | |
371 | ports.append((port_name, port)) | |
372 | ||
373 | self.assertEqual(ports[0][0], 'clear') | |
374 | self.assertEqual(ports[0][1].addr, port1.addr) | |
375 | self.assertEqual(ports[1][0], 'print') | |
376 | self.assertEqual(ports[1][1].addr, port2.addr) | |
377 | self.assertEqual(ports[2][0], 'insert') | |
378 | self.assertEqual(ports[2][1].addr, port3.addr) | |
379 | ||
894a8df5 | 380 | self._create_comp(MyFilter) |
811644b8 PP |
381 | |
382 | def test_user_sink_input_ports_iter(self): | |
383 | class MySink(bt2._UserSinkComponent): | |
59225a3e | 384 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
385 | port1 = comp_self._add_input_port('clear') |
386 | port2 = comp_self._add_input_port('print') | |
387 | port3 = comp_self._add_input_port('insert') | |
388 | ports = [] | |
389 | ||
390 | for port_name, port in comp_self._input_ports.items(): | |
391 | ports.append((port_name, port)) | |
392 | ||
393 | self.assertEqual(ports[0][0], 'clear') | |
394 | self.assertEqual(ports[0][1].addr, port1.addr) | |
395 | self.assertEqual(ports[1][0], 'print') | |
396 | self.assertEqual(ports[1][1].addr, port2.addr) | |
397 | self.assertEqual(ports[2][0], 'insert') | |
398 | self.assertEqual(ports[2][1].addr, port3.addr) | |
399 | ||
6a91742b | 400 | def _user_consume(self): |
a01b452b SM |
401 | pass |
402 | ||
894a8df5 | 403 | self._create_comp(MySink) |
811644b8 PP |
404 | |
405 | def test_gen_src_output_ports_getitem(self): | |
811644b8 PP |
406 | port1 = None |
407 | port2 = None | |
408 | port3 = None | |
409 | ||
5c61fb9d SM |
410 | class MySource( |
411 | bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator | |
412 | ): | |
59225a3e | 413 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
414 | nonlocal port1, port2, port3 |
415 | port1 = comp_self._add_output_port('clear') | |
416 | port2 = comp_self._add_output_port('print') | |
417 | port3 = comp_self._add_output_port('insert') | |
418 | ||
419 | comp = self._create_comp(MySource) | |
420 | self.assertEqual(port3.addr, comp.output_ports['insert'].addr) | |
421 | self.assertEqual(port2.addr, comp.output_ports['print'].addr) | |
422 | self.assertEqual(port1.addr, comp.output_ports['clear'].addr) | |
423 | del port1 | |
424 | del port2 | |
425 | del port3 | |
426 | ||
427 | def test_gen_flt_output_ports_getitem(self): | |
811644b8 PP |
428 | port1 = None |
429 | port2 = None | |
430 | port3 = None | |
431 | ||
5c61fb9d SM |
432 | class MyFilter( |
433 | bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator | |
434 | ): | |
59225a3e | 435 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
436 | nonlocal port1, port2, port3 |
437 | port1 = comp_self._add_output_port('clear') | |
438 | port2 = comp_self._add_output_port('print') | |
439 | port3 = comp_self._add_output_port('insert') | |
440 | ||
441 | comp = self._create_comp(MyFilter) | |
442 | self.assertEqual(port3.addr, comp.output_ports['insert'].addr) | |
443 | self.assertEqual(port2.addr, comp.output_ports['print'].addr) | |
444 | self.assertEqual(port1.addr, comp.output_ports['clear'].addr) | |
445 | del port1 | |
446 | del port2 | |
447 | del port3 | |
448 | ||
449 | def test_gen_flt_input_ports_getitem(self): | |
811644b8 PP |
450 | port1 = None |
451 | port2 = None | |
452 | port3 = None | |
453 | ||
5c61fb9d SM |
454 | class MyFilter( |
455 | bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator | |
456 | ): | |
59225a3e | 457 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
458 | nonlocal port1, port2, port3 |
459 | port1 = comp_self._add_input_port('clear') | |
460 | port2 = comp_self._add_input_port('print') | |
461 | port3 = comp_self._add_input_port('insert') | |
462 | ||
463 | comp = self._create_comp(MyFilter) | |
464 | self.assertEqual(port3.addr, comp.input_ports['insert'].addr) | |
465 | self.assertEqual(port2.addr, comp.input_ports['print'].addr) | |
466 | self.assertEqual(port1.addr, comp.input_ports['clear'].addr) | |
467 | del port1 | |
468 | del port2 | |
469 | del port3 | |
470 | ||
471 | def test_gen_sink_input_ports_getitem(self): | |
472 | port1 = None | |
473 | port2 = None | |
474 | port3 = None | |
475 | ||
476 | class MySink(bt2._UserSinkComponent): | |
59225a3e | 477 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
478 | nonlocal port1, port2, port3 |
479 | port1 = comp_self._add_input_port('clear') | |
480 | port2 = comp_self._add_input_port('print') | |
481 | port3 = comp_self._add_input_port('insert') | |
482 | ||
6a91742b | 483 | def _user_consume(self): |
a01b452b SM |
484 | pass |
485 | ||
811644b8 PP |
486 | comp = self._create_comp(MySink) |
487 | self.assertEqual(port3.addr, comp.input_ports['insert'].addr) | |
488 | self.assertEqual(port2.addr, comp.input_ports['print'].addr) | |
489 | self.assertEqual(port1.addr, comp.input_ports['clear'].addr) | |
490 | del port1 | |
491 | del port2 | |
492 | del port3 | |
493 | ||
494 | def test_gen_src_output_ports_getitem_invalid_key(self): | |
5c61fb9d SM |
495 | class MySource( |
496 | bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator | |
497 | ): | |
59225a3e | 498 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
499 | comp_self._add_output_port('clear') |
500 | comp_self._add_output_port('print') | |
501 | comp_self._add_output_port('insert') | |
502 | ||
503 | comp = self._create_comp(MySource) | |
504 | ||
505 | with self.assertRaises(KeyError): | |
506 | comp.output_ports['hello'] | |
507 | ||
508 | def test_gen_flt_output_ports_getitem_invalid_key(self): | |
5c61fb9d SM |
509 | class MyFilter( |
510 | bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator | |
511 | ): | |
59225a3e | 512 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
513 | comp_self._add_output_port('clear') |
514 | comp_self._add_output_port('print') | |
515 | comp_self._add_output_port('insert') | |
516 | ||
517 | comp = self._create_comp(MyFilter) | |
518 | ||
519 | with self.assertRaises(KeyError): | |
520 | comp.output_ports['hello'] | |
521 | ||
522 | def test_gen_flt_input_ports_getitem_invalid_key(self): | |
5c61fb9d SM |
523 | class MyFilter( |
524 | bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator | |
525 | ): | |
59225a3e | 526 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
527 | comp_self._add_input_port('clear') |
528 | comp_self._add_input_port('print') | |
529 | comp_self._add_input_port('insert') | |
530 | ||
531 | comp = self._create_comp(MyFilter) | |
532 | ||
533 | with self.assertRaises(KeyError): | |
534 | comp.input_ports['hello'] | |
535 | ||
536 | def test_gen_sink_input_ports_getitem_invalid_key(self): | |
537 | class MySink(bt2._UserSinkComponent): | |
59225a3e | 538 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
539 | comp_self._add_input_port('clear') |
540 | comp_self._add_input_port('print') | |
541 | comp_self._add_input_port('insert') | |
542 | ||
543 | with self.assertRaises(KeyError): | |
544 | comp_self._input_ports['hello'] | |
545 | ||
6a91742b | 546 | def _user_consume(self): |
a01b452b SM |
547 | pass |
548 | ||
811644b8 PP |
549 | comp = self._create_comp(MySink) |
550 | ||
551 | with self.assertRaises(KeyError): | |
552 | comp.input_ports['hello'] | |
553 | ||
554 | def test_gen_src_output_ports_len(self): | |
5c61fb9d SM |
555 | class MySource( |
556 | bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator | |
557 | ): | |
59225a3e | 558 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
559 | comp_self._add_output_port('clear') |
560 | comp_self._add_output_port('print') | |
561 | comp_self._add_output_port('insert') | |
562 | ||
563 | comp = self._create_comp(MySource) | |
564 | self.assertEqual(len(comp.output_ports), 3) | |
565 | ||
566 | def test_gen_flt_output_ports_len(self): | |
5c61fb9d SM |
567 | class MyFilter( |
568 | bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator | |
569 | ): | |
59225a3e | 570 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
571 | comp_self._add_output_port('clear') |
572 | comp_self._add_output_port('print') | |
573 | comp_self._add_output_port('insert') | |
574 | ||
575 | comp = self._create_comp(MyFilter) | |
576 | self.assertEqual(len(comp.output_ports), 3) | |
577 | ||
578 | def test_gen_flt_input_ports_len(self): | |
5c61fb9d SM |
579 | class MyFilter( |
580 | bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator | |
581 | ): | |
59225a3e | 582 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
583 | comp_self._add_input_port('clear') |
584 | comp_self._add_input_port('print') | |
585 | comp_self._add_input_port('insert') | |
586 | ||
587 | comp = self._create_comp(MyFilter) | |
588 | self.assertEqual(len(comp.input_ports), 3) | |
589 | ||
590 | def test_gen_sink_input_ports_len(self): | |
591 | class MySink(bt2._UserSinkComponent): | |
59225a3e | 592 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
593 | comp_self._add_input_port('clear') |
594 | comp_self._add_input_port('print') | |
595 | comp_self._add_input_port('insert') | |
596 | ||
6a91742b | 597 | def _user_consume(self): |
a01b452b SM |
598 | pass |
599 | ||
811644b8 PP |
600 | comp = self._create_comp(MySink) |
601 | self.assertEqual(len(comp.input_ports), 3) | |
602 | ||
603 | def test_gen_src_output_ports_iter(self): | |
811644b8 PP |
604 | port1 = None |
605 | port2 = None | |
606 | port3 = None | |
607 | ||
5c61fb9d SM |
608 | class MySource( |
609 | bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator | |
610 | ): | |
59225a3e | 611 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
612 | nonlocal port1, port2, port3 |
613 | port1 = comp_self._add_output_port('clear') | |
614 | port2 = comp_self._add_output_port('print') | |
615 | port3 = comp_self._add_output_port('insert') | |
616 | ||
617 | comp = self._create_comp(MySource) | |
618 | ports = [] | |
619 | ||
620 | for port_name, port in comp.output_ports.items(): | |
621 | ports.append((port_name, port)) | |
622 | ||
623 | self.assertEqual(ports[0][0], 'clear') | |
624 | self.assertEqual(ports[0][1].addr, port1.addr) | |
625 | self.assertEqual(ports[1][0], 'print') | |
626 | self.assertEqual(ports[1][1].addr, port2.addr) | |
627 | self.assertEqual(ports[2][0], 'insert') | |
628 | self.assertEqual(ports[2][1].addr, port3.addr) | |
629 | del port1 | |
630 | del port2 | |
631 | del port3 | |
632 | ||
633 | def test_gen_flt_output_ports_iter(self): | |
811644b8 PP |
634 | port1 = None |
635 | port2 = None | |
636 | port3 = None | |
637 | ||
5c61fb9d SM |
638 | class MyFilter( |
639 | bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator | |
640 | ): | |
59225a3e | 641 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
642 | nonlocal port1, port2, port3 |
643 | port1 = comp_self._add_output_port('clear') | |
644 | port2 = comp_self._add_output_port('print') | |
645 | port3 = comp_self._add_output_port('insert') | |
646 | ||
647 | comp = self._create_comp(MyFilter) | |
648 | ports = [] | |
649 | ||
650 | for port_name, port in comp.output_ports.items(): | |
651 | ports.append((port_name, port)) | |
652 | ||
653 | self.assertEqual(ports[0][0], 'clear') | |
654 | self.assertEqual(ports[0][1].addr, port1.addr) | |
655 | self.assertEqual(ports[1][0], 'print') | |
656 | self.assertEqual(ports[1][1].addr, port2.addr) | |
657 | self.assertEqual(ports[2][0], 'insert') | |
658 | self.assertEqual(ports[2][1].addr, port3.addr) | |
659 | del port1 | |
660 | del port2 | |
661 | del port3 | |
662 | ||
663 | def test_gen_flt_input_ports_iter(self): | |
811644b8 PP |
664 | port1 = None |
665 | port2 = None | |
666 | port3 = None | |
667 | ||
5c61fb9d SM |
668 | class MyFilter( |
669 | bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator | |
670 | ): | |
59225a3e | 671 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
672 | nonlocal port1, port2, port3 |
673 | port1 = comp_self._add_input_port('clear') | |
674 | port2 = comp_self._add_input_port('print') | |
675 | port3 = comp_self._add_input_port('insert') | |
676 | ||
677 | comp = self._create_comp(MyFilter) | |
678 | ports = [] | |
679 | ||
680 | for port_name, port in comp.input_ports.items(): | |
681 | ports.append((port_name, port)) | |
682 | ||
683 | self.assertEqual(ports[0][0], 'clear') | |
684 | self.assertEqual(ports[0][1].addr, port1.addr) | |
685 | self.assertEqual(ports[1][0], 'print') | |
686 | self.assertEqual(ports[1][1].addr, port2.addr) | |
687 | self.assertEqual(ports[2][0], 'insert') | |
688 | self.assertEqual(ports[2][1].addr, port3.addr) | |
689 | del port1 | |
690 | del port2 | |
691 | del port3 | |
692 | ||
693 | def test_gen_sink_input_ports_iter(self): | |
694 | port1 = None | |
695 | port2 = None | |
696 | port3 = None | |
697 | ||
698 | class MySink(bt2._UserSinkComponent): | |
59225a3e | 699 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
700 | nonlocal port1, port2, port3 |
701 | port1 = comp_self._add_input_port('clear') | |
702 | port2 = comp_self._add_input_port('print') | |
703 | port3 = comp_self._add_input_port('insert') | |
704 | ||
6a91742b | 705 | def _user_consume(self): |
a01b452b SM |
706 | pass |
707 | ||
811644b8 PP |
708 | comp = self._create_comp(MySink) |
709 | ports = [] | |
710 | ||
711 | for port_name, port in comp.input_ports.items(): | |
712 | ports.append((port_name, port)) | |
713 | ||
714 | self.assertEqual(ports[0][0], 'clear') | |
715 | self.assertEqual(ports[0][1].addr, port1.addr) | |
716 | self.assertEqual(ports[1][0], 'print') | |
717 | self.assertEqual(ports[1][1].addr, port2.addr) | |
718 | self.assertEqual(ports[2][0], 'insert') | |
719 | self.assertEqual(ports[2][1].addr, port3.addr) | |
720 | del port1 | |
721 | del port2 | |
722 | del port3 | |
723 | ||
724 | def test_name(self): | |
725 | class MySink(bt2._UserSinkComponent): | |
59225a3e | 726 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
727 | comp_self._add_input_port('clear') |
728 | ||
6a91742b | 729 | def _user_consume(self): |
a01b452b SM |
730 | pass |
731 | ||
811644b8 PP |
732 | comp = self._create_comp(MySink) |
733 | self.assertEqual(comp.input_ports['clear'].name, 'clear') | |
734 | ||
811644b8 PP |
735 | def test_connection_none(self): |
736 | class MySink(bt2._UserSinkComponent): | |
59225a3e | 737 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
738 | comp_self._add_input_port('clear') |
739 | ||
6a91742b | 740 | def _user_consume(self): |
a01b452b SM |
741 | pass |
742 | ||
811644b8 PP |
743 | comp = self._create_comp(MySink) |
744 | self.assertIsNone(comp.input_ports['clear'].connection) | |
745 | ||
746 | def test_is_connected_false(self): | |
747 | class MySink(bt2._UserSinkComponent): | |
59225a3e | 748 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
749 | comp_self._add_input_port('clear') |
750 | ||
6a91742b | 751 | def _user_consume(self): |
a01b452b SM |
752 | pass |
753 | ||
811644b8 PP |
754 | comp = self._create_comp(MySink) |
755 | self.assertFalse(comp.input_ports['clear'].is_connected) | |
756 | ||
894a8df5 | 757 | def test_self_name(self): |
811644b8 | 758 | class MySink(bt2._UserSinkComponent): |
59225a3e | 759 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
760 | port = comp_self._add_input_port('clear') |
761 | self.assertEqual(port.name, 'clear') | |
762 | ||
6a91742b | 763 | def _user_consume(self): |
a01b452b SM |
764 | pass |
765 | ||
894a8df5 | 766 | self._create_comp(MySink) |
811644b8 | 767 | |
894a8df5 | 768 | def test_self_connection_none(self): |
811644b8 | 769 | class MySink(bt2._UserSinkComponent): |
59225a3e | 770 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
771 | port = comp_self._add_input_port('clear') |
772 | self.assertIsNone(port.connection) | |
773 | ||
6a91742b | 774 | def _user_consume(self): |
a01b452b SM |
775 | pass |
776 | ||
894a8df5 | 777 | self._create_comp(MySink) |
811644b8 | 778 | |
894a8df5 | 779 | def test_self_is_connected_false(self): |
811644b8 | 780 | class MySink(bt2._UserSinkComponent): |
59225a3e | 781 | def __init__(comp_self, config, params, obj): |
811644b8 PP |
782 | port = comp_self._add_input_port('clear') |
783 | self.assertFalse(port.is_connected) | |
784 | ||
6a91742b | 785 | def _user_consume(self): |
a01b452b SM |
786 | pass |
787 | ||
894a8df5 | 788 | self._create_comp(MySink) |
2e00bc76 SM |
789 | |
790 | def test_source_self_port_user_data(self): | |
9675ad63 SM |
791 | class MyUserData: |
792 | def __del__(self): | |
793 | nonlocal objects_deleted | |
794 | objects_deleted += 1 | |
795 | ||
5c61fb9d SM |
796 | class MySource( |
797 | bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator | |
798 | ): | |
59225a3e | 799 | def __init__(comp_self, config, params, obj): |
2e00bc76 SM |
800 | nonlocal user_datas |
801 | ||
802 | p = comp_self._add_output_port('port1') | |
803 | user_datas.append(p.user_data) | |
804 | p = comp_self._add_output_port('port2', 2) | |
805 | user_datas.append(p.user_data) | |
9675ad63 SM |
806 | p = comp_self._add_output_port('port3', MyUserData()) |
807 | user_datas.append(p.user_data) | |
2e00bc76 SM |
808 | |
809 | user_datas = [] | |
9675ad63 | 810 | objects_deleted = 0 |
2e00bc76 | 811 | |
9675ad63 SM |
812 | comp = self._create_comp(MySource) |
813 | self.assertEqual(len(user_datas), 3) | |
814 | self.assertIs(user_datas[0], None) | |
815 | self.assertEqual(user_datas[1], 2) | |
816 | self.assertIs(type(user_datas[2]), MyUserData) | |
817 | ||
818 | # Verify that the user data gets freed. | |
819 | self.assertEqual(objects_deleted, 0) | |
820 | del user_datas | |
821 | del comp | |
822 | self.assertEqual(objects_deleted, 1) | |
2e00bc76 SM |
823 | |
824 | def test_filter_self_port_user_data(self): | |
9675ad63 SM |
825 | class MyUserData: |
826 | def __del__(self): | |
827 | nonlocal objects_deleted | |
828 | objects_deleted += 1 | |
829 | ||
5c61fb9d SM |
830 | class MyFilter( |
831 | bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator | |
832 | ): | |
59225a3e | 833 | def __init__(comp_self, config, params, obj): |
2e00bc76 SM |
834 | nonlocal user_datas |
835 | ||
836 | p = comp_self._add_output_port('port1') | |
837 | user_datas.append(p.user_data) | |
838 | p = comp_self._add_output_port('port2', 'user data string') | |
839 | user_datas.append(p.user_data) | |
9675ad63 SM |
840 | p = comp_self._add_output_port('port3', MyUserData()) |
841 | user_datas.append(p.user_data) | |
2e00bc76 | 842 | |
9675ad63 | 843 | p = comp_self._add_input_port('port4') |
2e00bc76 | 844 | user_datas.append(p.user_data) |
9675ad63 SM |
845 | p = comp_self._add_input_port('port5', user_data={'user data': 'dict'}) |
846 | user_datas.append(p.user_data) | |
847 | p = comp_self._add_input_port('port6', MyUserData()) | |
2e00bc76 SM |
848 | user_datas.append(p.user_data) |
849 | ||
850 | user_datas = [] | |
9675ad63 | 851 | objects_deleted = 0 |
2e00bc76 | 852 | |
9675ad63 SM |
853 | comp = self._create_comp(MyFilter) |
854 | self.assertEqual(len(user_datas), 6) | |
855 | self.assertIs(user_datas[0], None) | |
856 | self.assertEqual(user_datas[1], 'user data string') | |
857 | self.assertIs(type(user_datas[2]), MyUserData) | |
858 | self.assertIs(user_datas[3], None) | |
859 | self.assertEqual(user_datas[4], {'user data': 'dict'}) | |
860 | self.assertIs(type(user_datas[5]), MyUserData) | |
861 | ||
862 | # Verify that the user data gets freed. | |
863 | self.assertEqual(objects_deleted, 0) | |
864 | del user_datas | |
865 | del comp | |
866 | self.assertEqual(objects_deleted, 2) | |
2e00bc76 SM |
867 | |
868 | def test_sink_self_port_user_data(self): | |
9675ad63 SM |
869 | class MyUserData: |
870 | def __del__(self): | |
871 | nonlocal objects_deleted | |
872 | objects_deleted += 1 | |
873 | ||
1606ab4c | 874 | class MySink(bt2._UserSinkComponent): |
59225a3e | 875 | def __init__(comp_self, config, params, obj): |
2e00bc76 SM |
876 | nonlocal user_datas |
877 | ||
878 | p = comp_self._add_input_port('port1') | |
879 | user_datas.append(p.user_data) | |
9675ad63 | 880 | p = comp_self._add_input_port('port2', MyUserData()) |
2e00bc76 SM |
881 | user_datas.append(p.user_data) |
882 | ||
1606ab4c SM |
883 | def _user_consume(self): |
884 | pass | |
885 | ||
2e00bc76 | 886 | user_datas = [] |
9675ad63 | 887 | objects_deleted = 0 |
2e00bc76 | 888 | |
9675ad63 SM |
889 | comp = self._create_comp(MySink) |
890 | self.assertEqual(len(user_datas), 2) | |
891 | self.assertIs(user_datas[0], None) | |
892 | self.assertIs(type(user_datas[1]), MyUserData) | |
893 | ||
894 | # Verify that the user data gets freed. | |
895 | self.assertEqual(objects_deleted, 0) | |
896 | del user_datas | |
897 | del comp | |
898 | self.assertEqual(objects_deleted, 1) | |
d14ddbba SM |
899 | |
900 | ||
901 | if __name__ == '__main__': | |
902 | unittest.main() |