3a4461455ccc34b3d8b5dae3802dccfcc2f87266
1 # SPDX-License-Identifier: GPL-2.0-only
3 # Copyright (C) 2019 EfficiOS Inc.
9 from bt2
import port
as bt2_port
12 class PortTestCase(unittest
.TestCase
):
14 def _create_comp(comp_cls
, name
=None):
20 return graph
.add_component(comp_cls
, name
)
22 def test_src_add_output_port(self
):
24 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
26 def __init__(comp_self
, config
, params
, obj
):
27 port
= comp_self
._add
_output
_port
("out")
28 self
.assertEqual(port
.name
, "out")
30 comp
= self
._create
_comp
(MySource
)
31 self
.assertEqual(len(comp
.output_ports
), 1)
32 self
.assertIs(type(comp
.output_ports
["out"]), bt2_port
._OutputPortConst
)
34 # Test adding output port with duplicate name to source.
35 def test_src_add_output_port_dup_name_raises(self
):
37 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
39 def __init__(comp_self
, config
, params
, obj
):
40 comp_self
._add
_output
_port
("out")
42 with self
.assertRaisesRegex(
44 "source component `comp` already contains an output port named `out`",
46 comp_self
._add
_output
_port
("out")
52 self
._create
_comp
(MySource
)
55 def test_flt_add_output_port(self
):
57 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
59 def __init__(comp_self
, config
, params
, obj
):
60 port
= comp_self
._add
_output
_port
("out")
61 self
.assertEqual(port
.name
, "out")
63 comp
= self
._create
_comp
(MyFilter
)
64 self
.assertEqual(len(comp
.output_ports
), 1)
66 # Test adding output port with duplicate name to filter.
67 def test_flt_add_output_port_dup_name_raises(self
):
69 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
71 def __init__(comp_self
, config
, params
, obj
):
72 comp_self
._add
_output
_port
("out")
74 with self
.assertRaisesRegex(
76 "filter component `comp` already contains an output port named `out`",
78 comp_self
._add
_output
_port
("out")
84 self
._create
_comp
(MyFilter
)
87 def test_flt_add_input_port(self
):
89 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
91 def __init__(comp_self
, config
, params
, obj
):
92 port
= comp_self
._add
_input
_port
("in")
93 self
.assertEqual(port
.name
, "in")
95 comp
= self
._create
_comp
(MyFilter
)
96 self
.assertEqual(len(comp
.input_ports
), 1)
97 self
.assertIs(type(comp
.input_ports
["in"]), bt2_port
._InputPortConst
)
99 # Test adding input port with duplicate name to filter.
100 def test_flt_add_input_port_dup_name_raises(self
):
102 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
104 def __init__(comp_self
, config
, params
, obj
):
105 comp_self
._add
_input
_port
("in")
107 with self
.assertRaisesRegex(
109 "filter component `comp` already contains an input port named `in`",
111 comp_self
._add
_input
_port
("in")
117 self
._create
_comp
(MyFilter
)
118 self
.assertTrue(seen
)
120 def test_sink_add_input_port(self
):
121 class MySink(bt2
._UserSinkComponent
):
122 def __init__(comp_self
, config
, params
, obj
):
123 port
= comp_self
._add
_input
_port
("in")
124 self
.assertEqual(port
.name
, "in")
126 def _user_consume(self
):
129 comp
= self
._create
_comp
(MySink
)
130 self
.assertEqual(len(comp
.input_ports
), 1)
132 # Test adding input port with duplicate name to sink.
133 def test_sink_add_input_port_dup_name_raises(self
):
134 class MySink(bt2
._UserSinkComponent
):
135 def __init__(comp_self
, config
, params
, obj
):
136 comp_self
._add
_input
_port
("in")
138 with self
.assertRaisesRegex(
140 "sink component `comp` already contains an input port named `in`",
142 comp_self
._add
_input
_port
("in")
147 def _user_consume(self
):
151 self
._create
_comp
(MySink
)
152 self
.assertTrue(seen
)
154 def test_user_src_output_ports_getitem(self
):
156 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
158 def __init__(comp_self
, config
, params
, obj
):
159 port1
= comp_self
._add
_output
_port
("clear")
160 port2
= comp_self
._add
_output
_port
("print")
161 port3
= comp_self
._add
_output
_port
("insert")
162 self
.assertEqual(port3
.addr
, comp_self
._output
_ports
["insert"].addr
)
163 self
.assertEqual(port2
.addr
, comp_self
._output
_ports
["print"].addr
)
164 self
.assertEqual(port1
.addr
, comp_self
._output
_ports
["clear"].addr
)
166 self
._create
_comp
(MySource
)
168 def test_user_flt_output_ports_getitem(self
):
170 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
172 def __init__(comp_self
, config
, params
, obj
):
173 port1
= comp_self
._add
_output
_port
("clear")
174 port2
= comp_self
._add
_output
_port
("print")
175 port3
= comp_self
._add
_output
_port
("insert")
176 self
.assertEqual(port3
.addr
, comp_self
._output
_ports
["insert"].addr
)
177 self
.assertEqual(port2
.addr
, comp_self
._output
_ports
["print"].addr
)
178 self
.assertEqual(port1
.addr
, comp_self
._output
_ports
["clear"].addr
)
180 self
._create
_comp
(MyFilter
)
182 def test_user_flt_input_ports_getitem(self
):
184 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
186 def __init__(comp_self
, config
, params
, obj
):
187 port1
= comp_self
._add
_input
_port
("clear")
188 port2
= comp_self
._add
_input
_port
("print")
189 port3
= comp_self
._add
_input
_port
("insert")
190 self
.assertEqual(port3
.addr
, comp_self
._input
_ports
["insert"].addr
)
191 self
.assertEqual(port2
.addr
, comp_self
._input
_ports
["print"].addr
)
192 self
.assertEqual(port1
.addr
, comp_self
._input
_ports
["clear"].addr
)
194 self
._create
_comp
(MyFilter
)
196 def test_user_sink_input_ports_getitem(self
):
197 class MySink(bt2
._UserSinkComponent
):
198 def __init__(comp_self
, config
, params
, obj
):
199 port1
= comp_self
._add
_input
_port
("clear")
200 port2
= comp_self
._add
_input
_port
("print")
201 port3
= comp_self
._add
_input
_port
("insert")
202 self
.assertEqual(port3
.addr
, comp_self
._input
_ports
["insert"].addr
)
203 self
.assertEqual(port2
.addr
, comp_self
._input
_ports
["print"].addr
)
204 self
.assertEqual(port1
.addr
, comp_self
._input
_ports
["clear"].addr
)
206 def _user_consume(self
):
209 self
._create
_comp
(MySink
)
211 def test_user_src_output_ports_getitem_invalid_key(self
):
213 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
215 def __init__(comp_self
, config
, params
, obj
):
216 comp_self
._add
_output
_port
("clear")
217 comp_self
._add
_output
_port
("print")
218 comp_self
._add
_output
_port
("insert")
220 with self
.assertRaises(KeyError):
221 comp_self
._output
_ports
["hello"]
223 self
._create
_comp
(MySource
)
225 def test_user_flt_output_ports_getitem_invalid_key(self
):
227 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
229 def __init__(comp_self
, config
, params
, obj
):
230 comp_self
._add
_output
_port
("clear")
231 comp_self
._add
_output
_port
("print")
232 comp_self
._add
_output
_port
("insert")
234 with self
.assertRaises(KeyError):
235 comp_self
._output
_ports
["hello"]
237 self
._create
_comp
(MyFilter
)
239 def test_user_flt_input_ports_getitem_invalid_key(self
):
241 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
243 def __init__(comp_self
, config
, params
, obj
):
244 comp_self
._add
_input
_port
("clear")
245 comp_self
._add
_input
_port
("print")
246 comp_self
._add
_input
_port
("insert")
248 with self
.assertRaises(KeyError):
249 comp_self
._input
_ports
["hello"]
251 self
._create
_comp
(MyFilter
)
253 def test_user_sink_input_ports_getitem_invalid_key(self
):
254 class MySink(bt2
._UserSinkComponent
):
255 def __init__(comp_self
, config
, params
, obj
):
256 comp_self
._add
_input
_port
("clear")
257 comp_self
._add
_input
_port
("print")
258 comp_self
._add
_input
_port
("insert")
260 with self
.assertRaises(KeyError):
261 comp_self
._input
_ports
["hello"]
263 def _user_consume(self
):
266 self
._create
_comp
(MySink
)
268 def test_user_src_output_ports_len(self
):
270 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
272 def __init__(comp_self
, config
, params
, obj
):
273 comp_self
._add
_output
_port
("clear")
274 comp_self
._add
_output
_port
("print")
275 comp_self
._add
_output
_port
("insert")
276 self
.assertEqual(len(comp_self
._output
_ports
), 3)
278 self
._create
_comp
(MySource
)
280 def test_user_flt_output_ports_len(self
):
282 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
284 def __init__(comp_self
, config
, params
, obj
):
285 comp_self
._add
_output
_port
("clear")
286 comp_self
._add
_output
_port
("print")
287 comp_self
._add
_output
_port
("insert")
288 self
.assertEqual(len(comp_self
._output
_ports
), 3)
290 self
._create
_comp
(MyFilter
)
292 def test_user_flt_input_ports_len(self
):
294 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
296 def __init__(comp_self
, config
, params
, obj
):
297 comp_self
._add
_input
_port
("clear")
298 comp_self
._add
_input
_port
("print")
299 comp_self
._add
_input
_port
("insert")
300 self
.assertEqual(len(comp_self
._input
_ports
), 3)
302 self
._create
_comp
(MyFilter
)
304 def test_user_sink_input_ports_len(self
):
305 class MySink(bt2
._UserSinkComponent
):
306 def __init__(comp_self
, config
, params
, obj
):
307 comp_self
._add
_input
_port
("clear")
308 comp_self
._add
_input
_port
("print")
309 comp_self
._add
_input
_port
("insert")
310 self
.assertEqual(len(comp_self
._input
_ports
), 3)
312 def _user_consume(self
):
315 self
._create
_comp
(MySink
)
317 def test_user_src_output_ports_iter(self
):
319 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
321 def __init__(comp_self
, config
, params
, obj
):
322 port1
= comp_self
._add
_output
_port
("clear")
323 port2
= comp_self
._add
_output
_port
("print")
324 port3
= comp_self
._add
_output
_port
("insert")
327 for port_name
, port
in comp_self
._output
_ports
.items():
328 ports
.append((port_name
, port
))
330 self
.assertEqual(ports
[0][0], "clear")
331 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
332 self
.assertEqual(ports
[1][0], "print")
333 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
334 self
.assertEqual(ports
[2][0], "insert")
335 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
337 self
._create
_comp
(MySource
)
339 def test_user_flt_output_ports_iter(self
):
341 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
343 def __init__(comp_self
, config
, params
, obj
):
344 port1
= comp_self
._add
_output
_port
("clear")
345 port2
= comp_self
._add
_output
_port
("print")
346 port3
= comp_self
._add
_output
_port
("insert")
349 for port_name
, port
in comp_self
._output
_ports
.items():
350 ports
.append((port_name
, port
))
352 self
.assertEqual(ports
[0][0], "clear")
353 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
354 self
.assertEqual(ports
[1][0], "print")
355 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
356 self
.assertEqual(ports
[2][0], "insert")
357 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
359 self
._create
_comp
(MyFilter
)
361 def test_user_flt_input_ports_iter(self
):
363 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
365 def __init__(comp_self
, config
, params
, obj
):
366 port1
= comp_self
._add
_input
_port
("clear")
367 port2
= comp_self
._add
_input
_port
("print")
368 port3
= comp_self
._add
_input
_port
("insert")
371 for port_name
, port
in comp_self
._input
_ports
.items():
372 ports
.append((port_name
, port
))
374 self
.assertEqual(ports
[0][0], "clear")
375 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
376 self
.assertEqual(ports
[1][0], "print")
377 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
378 self
.assertEqual(ports
[2][0], "insert")
379 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
381 self
._create
_comp
(MyFilter
)
383 def test_user_sink_input_ports_iter(self
):
384 class MySink(bt2
._UserSinkComponent
):
385 def __init__(comp_self
, config
, params
, obj
):
386 port1
= comp_self
._add
_input
_port
("clear")
387 port2
= comp_self
._add
_input
_port
("print")
388 port3
= comp_self
._add
_input
_port
("insert")
391 for port_name
, port
in comp_self
._input
_ports
.items():
392 ports
.append((port_name
, port
))
394 self
.assertEqual(ports
[0][0], "clear")
395 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
396 self
.assertEqual(ports
[1][0], "print")
397 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
398 self
.assertEqual(ports
[2][0], "insert")
399 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
401 def _user_consume(self
):
404 self
._create
_comp
(MySink
)
406 def test_gen_src_output_ports_getitem(self
):
412 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
414 def __init__(comp_self
, config
, params
, obj
):
415 nonlocal port1
, port2
, port3
416 port1
= comp_self
._add
_output
_port
("clear")
417 port2
= comp_self
._add
_output
_port
("print")
418 port3
= comp_self
._add
_output
_port
("insert")
420 comp
= self
._create
_comp
(MySource
)
421 self
.assertEqual(port3
.addr
, comp
.output_ports
["insert"].addr
)
422 self
.assertEqual(port2
.addr
, comp
.output_ports
["print"].addr
)
423 self
.assertEqual(port1
.addr
, comp
.output_ports
["clear"].addr
)
428 def test_gen_flt_output_ports_getitem(self
):
434 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
436 def __init__(comp_self
, config
, params
, obj
):
437 nonlocal port1
, port2
, port3
438 port1
= comp_self
._add
_output
_port
("clear")
439 port2
= comp_self
._add
_output
_port
("print")
440 port3
= comp_self
._add
_output
_port
("insert")
442 comp
= self
._create
_comp
(MyFilter
)
443 self
.assertEqual(port3
.addr
, comp
.output_ports
["insert"].addr
)
444 self
.assertEqual(port2
.addr
, comp
.output_ports
["print"].addr
)
445 self
.assertEqual(port1
.addr
, comp
.output_ports
["clear"].addr
)
450 def test_gen_flt_input_ports_getitem(self
):
456 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
458 def __init__(comp_self
, config
, params
, obj
):
459 nonlocal port1
, port2
, port3
460 port1
= comp_self
._add
_input
_port
("clear")
461 port2
= comp_self
._add
_input
_port
("print")
462 port3
= comp_self
._add
_input
_port
("insert")
464 comp
= self
._create
_comp
(MyFilter
)
465 self
.assertEqual(port3
.addr
, comp
.input_ports
["insert"].addr
)
466 self
.assertEqual(port2
.addr
, comp
.input_ports
["print"].addr
)
467 self
.assertEqual(port1
.addr
, comp
.input_ports
["clear"].addr
)
472 def test_gen_sink_input_ports_getitem(self
):
477 class MySink(bt2
._UserSinkComponent
):
478 def __init__(comp_self
, config
, params
, obj
):
479 nonlocal port1
, port2
, port3
480 port1
= comp_self
._add
_input
_port
("clear")
481 port2
= comp_self
._add
_input
_port
("print")
482 port3
= comp_self
._add
_input
_port
("insert")
484 def _user_consume(self
):
487 comp
= self
._create
_comp
(MySink
)
488 self
.assertEqual(port3
.addr
, comp
.input_ports
["insert"].addr
)
489 self
.assertEqual(port2
.addr
, comp
.input_ports
["print"].addr
)
490 self
.assertEqual(port1
.addr
, comp
.input_ports
["clear"].addr
)
495 def test_gen_src_output_ports_getitem_invalid_key(self
):
497 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
499 def __init__(comp_self
, config
, params
, obj
):
500 comp_self
._add
_output
_port
("clear")
501 comp_self
._add
_output
_port
("print")
502 comp_self
._add
_output
_port
("insert")
504 comp
= self
._create
_comp
(MySource
)
506 with self
.assertRaises(KeyError):
507 comp
.output_ports
["hello"]
509 def test_gen_flt_output_ports_getitem_invalid_key(self
):
511 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
513 def __init__(comp_self
, config
, params
, obj
):
514 comp_self
._add
_output
_port
("clear")
515 comp_self
._add
_output
_port
("print")
516 comp_self
._add
_output
_port
("insert")
518 comp
= self
._create
_comp
(MyFilter
)
520 with self
.assertRaises(KeyError):
521 comp
.output_ports
["hello"]
523 def test_gen_flt_input_ports_getitem_invalid_key(self
):
525 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
527 def __init__(comp_self
, config
, params
, obj
):
528 comp_self
._add
_input
_port
("clear")
529 comp_self
._add
_input
_port
("print")
530 comp_self
._add
_input
_port
("insert")
532 comp
= self
._create
_comp
(MyFilter
)
534 with self
.assertRaises(KeyError):
535 comp
.input_ports
["hello"]
537 def test_gen_sink_input_ports_getitem_invalid_key(self
):
538 class MySink(bt2
._UserSinkComponent
):
539 def __init__(comp_self
, config
, params
, obj
):
540 comp_self
._add
_input
_port
("clear")
541 comp_self
._add
_input
_port
("print")
542 comp_self
._add
_input
_port
("insert")
544 with self
.assertRaises(KeyError):
545 comp_self
._input
_ports
["hello"]
547 def _user_consume(self
):
550 comp
= self
._create
_comp
(MySink
)
552 with self
.assertRaises(KeyError):
553 comp
.input_ports
["hello"]
555 def test_gen_src_output_ports_len(self
):
557 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
559 def __init__(comp_self
, config
, params
, obj
):
560 comp_self
._add
_output
_port
("clear")
561 comp_self
._add
_output
_port
("print")
562 comp_self
._add
_output
_port
("insert")
564 comp
= self
._create
_comp
(MySource
)
565 self
.assertEqual(len(comp
.output_ports
), 3)
567 def test_gen_flt_output_ports_len(self
):
569 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
571 def __init__(comp_self
, config
, params
, obj
):
572 comp_self
._add
_output
_port
("clear")
573 comp_self
._add
_output
_port
("print")
574 comp_self
._add
_output
_port
("insert")
576 comp
= self
._create
_comp
(MyFilter
)
577 self
.assertEqual(len(comp
.output_ports
), 3)
579 def test_gen_flt_input_ports_len(self
):
581 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
583 def __init__(comp_self
, config
, params
, obj
):
584 comp_self
._add
_input
_port
("clear")
585 comp_self
._add
_input
_port
("print")
586 comp_self
._add
_input
_port
("insert")
588 comp
= self
._create
_comp
(MyFilter
)
589 self
.assertEqual(len(comp
.input_ports
), 3)
591 def test_gen_sink_input_ports_len(self
):
592 class MySink(bt2
._UserSinkComponent
):
593 def __init__(comp_self
, config
, params
, obj
):
594 comp_self
._add
_input
_port
("clear")
595 comp_self
._add
_input
_port
("print")
596 comp_self
._add
_input
_port
("insert")
598 def _user_consume(self
):
601 comp
= self
._create
_comp
(MySink
)
602 self
.assertEqual(len(comp
.input_ports
), 3)
604 def test_gen_src_output_ports_iter(self
):
610 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
612 def __init__(comp_self
, config
, params
, obj
):
613 nonlocal port1
, port2
, port3
614 port1
= comp_self
._add
_output
_port
("clear")
615 port2
= comp_self
._add
_output
_port
("print")
616 port3
= comp_self
._add
_output
_port
("insert")
618 comp
= self
._create
_comp
(MySource
)
621 for port_name
, port
in comp
.output_ports
.items():
622 ports
.append((port_name
, port
))
624 self
.assertEqual(ports
[0][0], "clear")
625 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
626 self
.assertEqual(ports
[1][0], "print")
627 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
628 self
.assertEqual(ports
[2][0], "insert")
629 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
634 def test_gen_flt_output_ports_iter(self
):
640 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
642 def __init__(comp_self
, config
, params
, obj
):
643 nonlocal port1
, port2
, port3
644 port1
= comp_self
._add
_output
_port
("clear")
645 port2
= comp_self
._add
_output
_port
("print")
646 port3
= comp_self
._add
_output
_port
("insert")
648 comp
= self
._create
_comp
(MyFilter
)
651 for port_name
, port
in comp
.output_ports
.items():
652 ports
.append((port_name
, port
))
654 self
.assertEqual(ports
[0][0], "clear")
655 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
656 self
.assertEqual(ports
[1][0], "print")
657 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
658 self
.assertEqual(ports
[2][0], "insert")
659 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
664 def test_gen_flt_input_ports_iter(self
):
670 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
672 def __init__(comp_self
, config
, params
, obj
):
673 nonlocal port1
, port2
, port3
674 port1
= comp_self
._add
_input
_port
("clear")
675 port2
= comp_self
._add
_input
_port
("print")
676 port3
= comp_self
._add
_input
_port
("insert")
678 comp
= self
._create
_comp
(MyFilter
)
681 for port_name
, port
in comp
.input_ports
.items():
682 ports
.append((port_name
, port
))
684 self
.assertEqual(ports
[0][0], "clear")
685 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
686 self
.assertEqual(ports
[1][0], "print")
687 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
688 self
.assertEqual(ports
[2][0], "insert")
689 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
694 def test_gen_sink_input_ports_iter(self
):
699 class MySink(bt2
._UserSinkComponent
):
700 def __init__(comp_self
, config
, params
, obj
):
701 nonlocal port1
, port2
, port3
702 port1
= comp_self
._add
_input
_port
("clear")
703 port2
= comp_self
._add
_input
_port
("print")
704 port3
= comp_self
._add
_input
_port
("insert")
706 def _user_consume(self
):
709 comp
= self
._create
_comp
(MySink
)
712 for port_name
, port
in comp
.input_ports
.items():
713 ports
.append((port_name
, port
))
715 self
.assertEqual(ports
[0][0], "clear")
716 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
717 self
.assertEqual(ports
[1][0], "print")
718 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
719 self
.assertEqual(ports
[2][0], "insert")
720 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
726 class MySink(bt2
._UserSinkComponent
):
727 def __init__(comp_self
, config
, params
, obj
):
728 comp_self
._add
_input
_port
("clear")
730 def _user_consume(self
):
733 comp
= self
._create
_comp
(MySink
)
734 self
.assertEqual(comp
.input_ports
["clear"].name
, "clear")
736 def test_connection_none(self
):
737 class MySink(bt2
._UserSinkComponent
):
738 def __init__(comp_self
, config
, params
, obj
):
739 comp_self
._add
_input
_port
("clear")
741 def _user_consume(self
):
744 comp
= self
._create
_comp
(MySink
)
745 self
.assertIsNone(comp
.input_ports
["clear"].connection
)
747 def test_is_connected_false(self
):
748 class MySink(bt2
._UserSinkComponent
):
749 def __init__(comp_self
, config
, params
, obj
):
750 comp_self
._add
_input
_port
("clear")
752 def _user_consume(self
):
755 comp
= self
._create
_comp
(MySink
)
756 self
.assertFalse(comp
.input_ports
["clear"].is_connected
)
758 def test_self_name(self
):
759 class MySink(bt2
._UserSinkComponent
):
760 def __init__(comp_self
, config
, params
, obj
):
761 port
= comp_self
._add
_input
_port
("clear")
762 self
.assertEqual(port
.name
, "clear")
764 def _user_consume(self
):
767 self
._create
_comp
(MySink
)
769 def test_self_connection_none(self
):
770 class MySink(bt2
._UserSinkComponent
):
771 def __init__(comp_self
, config
, params
, obj
):
772 port
= comp_self
._add
_input
_port
("clear")
773 self
.assertIsNone(port
.connection
)
775 def _user_consume(self
):
778 self
._create
_comp
(MySink
)
780 def test_self_is_connected_false(self
):
781 class MySink(bt2
._UserSinkComponent
):
782 def __init__(comp_self
, config
, params
, obj
):
783 port
= comp_self
._add
_input
_port
("clear")
784 self
.assertFalse(port
.is_connected
)
786 def _user_consume(self
):
789 self
._create
_comp
(MySink
)
791 def test_source_self_port_user_data(self
):
794 nonlocal objects_deleted
798 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
800 def __init__(comp_self
, config
, params
, obj
):
803 p
= comp_self
._add
_output
_port
("port1")
804 user_datas
.append(p
.user_data
)
805 p
= comp_self
._add
_output
_port
("port2", 2)
806 user_datas
.append(p
.user_data
)
807 p
= comp_self
._add
_output
_port
("port3", MyUserData())
808 user_datas
.append(p
.user_data
)
813 comp
= self
._create
_comp
(MySource
)
814 self
.assertEqual(len(user_datas
), 3)
815 self
.assertIs(user_datas
[0], None)
816 self
.assertEqual(user_datas
[1], 2)
817 self
.assertIs(type(user_datas
[2]), MyUserData
)
819 # Verify that the user data gets freed.
820 self
.assertEqual(objects_deleted
, 0)
823 self
.assertEqual(objects_deleted
, 1)
825 def test_filter_self_port_user_data(self
):
828 nonlocal objects_deleted
832 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
834 def __init__(comp_self
, config
, params
, obj
):
837 p
= comp_self
._add
_output
_port
("port1")
838 user_datas
.append(p
.user_data
)
839 p
= comp_self
._add
_output
_port
("port2", "user data string")
840 user_datas
.append(p
.user_data
)
841 p
= comp_self
._add
_output
_port
("port3", MyUserData())
842 user_datas
.append(p
.user_data
)
844 p
= comp_self
._add
_input
_port
("port4")
845 user_datas
.append(p
.user_data
)
846 p
= comp_self
._add
_input
_port
("port5", user_data
={"user data": "dict"})
847 user_datas
.append(p
.user_data
)
848 p
= comp_self
._add
_input
_port
("port6", MyUserData())
849 user_datas
.append(p
.user_data
)
854 comp
= self
._create
_comp
(MyFilter
)
855 self
.assertEqual(len(user_datas
), 6)
856 self
.assertIs(user_datas
[0], None)
857 self
.assertEqual(user_datas
[1], "user data string")
858 self
.assertIs(type(user_datas
[2]), MyUserData
)
859 self
.assertIs(user_datas
[3], None)
860 self
.assertEqual(user_datas
[4], {"user data": "dict"})
861 self
.assertIs(type(user_datas
[5]), MyUserData
)
863 # Verify that the user data gets freed.
864 self
.assertEqual(objects_deleted
, 0)
867 self
.assertEqual(objects_deleted
, 2)
869 def test_sink_self_port_user_data(self
):
872 nonlocal objects_deleted
875 class MySink(bt2
._UserSinkComponent
):
876 def __init__(comp_self
, config
, params
, obj
):
879 p
= comp_self
._add
_input
_port
("port1")
880 user_datas
.append(p
.user_data
)
881 p
= comp_self
._add
_input
_port
("port2", MyUserData())
882 user_datas
.append(p
.user_data
)
884 def _user_consume(self
):
890 comp
= self
._create
_comp
(MySink
)
891 self
.assertEqual(len(user_datas
), 2)
892 self
.assertIs(user_datas
[0], None)
893 self
.assertIs(type(user_datas
[1]), MyUserData
)
895 # Verify that the user data gets freed.
896 self
.assertEqual(objects_deleted
, 0)
899 self
.assertEqual(objects_deleted
, 1)
902 if __name__
== "__main__":
This page took 0.048566 seconds and 4 git commands to generate.