1 # SPDX-License-Identifier: GPL-2.0-only
3 # Copyright (C) 2019 EfficiOS Inc.
8 from bt2
import port
as bt2_port
11 class PortTestCase(unittest
.TestCase
):
13 def _create_comp(comp_cls
, name
=None):
19 return graph
.add_component(comp_cls
, name
)
21 def test_src_add_output_port(self
):
23 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
25 def __init__(comp_self
, config
, params
, obj
):
26 port
= comp_self
._add
_output
_port
("out")
27 self
.assertEqual(port
.name
, "out")
29 comp
= self
._create
_comp
(MySource
)
30 self
.assertEqual(len(comp
.output_ports
), 1)
31 self
.assertIs(type(comp
.output_ports
["out"]), bt2_port
._OutputPortConst
)
33 # Test adding output port with duplicate name to source.
34 def test_src_add_output_port_dup_name_raises(self
):
36 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
38 def __init__(comp_self
, config
, params
, obj
):
39 comp_self
._add
_output
_port
("out")
41 with self
.assertRaisesRegex(
43 "source component `comp` already contains an output port named `out`",
45 comp_self
._add
_output
_port
("out")
51 self
._create
_comp
(MySource
)
54 def test_flt_add_output_port(self
):
56 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
58 def __init__(comp_self
, config
, params
, obj
):
59 port
= comp_self
._add
_output
_port
("out")
60 self
.assertEqual(port
.name
, "out")
62 comp
= self
._create
_comp
(MyFilter
)
63 self
.assertEqual(len(comp
.output_ports
), 1)
65 # Test adding output port with duplicate name to filter.
66 def test_flt_add_output_port_dup_name_raises(self
):
68 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
70 def __init__(comp_self
, config
, params
, obj
):
71 comp_self
._add
_output
_port
("out")
73 with self
.assertRaisesRegex(
75 "filter component `comp` already contains an output port named `out`",
77 comp_self
._add
_output
_port
("out")
83 self
._create
_comp
(MyFilter
)
86 def test_flt_add_input_port(self
):
88 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
90 def __init__(comp_self
, config
, params
, obj
):
91 port
= comp_self
._add
_input
_port
("in")
92 self
.assertEqual(port
.name
, "in")
94 comp
= self
._create
_comp
(MyFilter
)
95 self
.assertEqual(len(comp
.input_ports
), 1)
96 self
.assertIs(type(comp
.input_ports
["in"]), bt2_port
._InputPortConst
)
98 # Test adding input port with duplicate name to filter.
99 def test_flt_add_input_port_dup_name_raises(self
):
101 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
103 def __init__(comp_self
, config
, params
, obj
):
104 comp_self
._add
_input
_port
("in")
106 with self
.assertRaisesRegex(
108 "filter component `comp` already contains an input port named `in`",
110 comp_self
._add
_input
_port
("in")
116 self
._create
_comp
(MyFilter
)
117 self
.assertTrue(seen
)
119 def test_sink_add_input_port(self
):
120 class MySink(bt2
._UserSinkComponent
):
121 def __init__(comp_self
, config
, params
, obj
):
122 port
= comp_self
._add
_input
_port
("in")
123 self
.assertEqual(port
.name
, "in")
125 def _user_consume(self
):
128 comp
= self
._create
_comp
(MySink
)
129 self
.assertEqual(len(comp
.input_ports
), 1)
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")
137 with self
.assertRaisesRegex(
139 "sink component `comp` already contains an input port named `in`",
141 comp_self
._add
_input
_port
("in")
146 def _user_consume(self
):
150 self
._create
_comp
(MySink
)
151 self
.assertTrue(seen
)
153 def test_user_src_output_ports_getitem(self
):
155 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
157 def __init__(comp_self
, config
, params
, obj
):
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
)
165 self
._create
_comp
(MySource
)
167 def test_user_flt_output_ports_getitem(self
):
169 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
171 def __init__(comp_self
, config
, params
, obj
):
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
)
179 self
._create
_comp
(MyFilter
)
181 def test_user_flt_input_ports_getitem(self
):
183 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
185 def __init__(comp_self
, config
, params
, obj
):
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
)
193 self
._create
_comp
(MyFilter
)
195 def test_user_sink_input_ports_getitem(self
):
196 class MySink(bt2
._UserSinkComponent
):
197 def __init__(comp_self
, config
, params
, obj
):
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
)
205 def _user_consume(self
):
208 self
._create
_comp
(MySink
)
210 def test_user_src_output_ports_getitem_invalid_key(self
):
212 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
214 def __init__(comp_self
, config
, params
, obj
):
215 comp_self
._add
_output
_port
("clear")
216 comp_self
._add
_output
_port
("print")
217 comp_self
._add
_output
_port
("insert")
219 with self
.assertRaises(KeyError):
220 comp_self
._output
_ports
["hello"]
222 self
._create
_comp
(MySource
)
224 def test_user_flt_output_ports_getitem_invalid_key(self
):
226 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
228 def __init__(comp_self
, config
, params
, obj
):
229 comp_self
._add
_output
_port
("clear")
230 comp_self
._add
_output
_port
("print")
231 comp_self
._add
_output
_port
("insert")
233 with self
.assertRaises(KeyError):
234 comp_self
._output
_ports
["hello"]
236 self
._create
_comp
(MyFilter
)
238 def test_user_flt_input_ports_getitem_invalid_key(self
):
240 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
242 def __init__(comp_self
, config
, params
, obj
):
243 comp_self
._add
_input
_port
("clear")
244 comp_self
._add
_input
_port
("print")
245 comp_self
._add
_input
_port
("insert")
247 with self
.assertRaises(KeyError):
248 comp_self
._input
_ports
["hello"]
250 self
._create
_comp
(MyFilter
)
252 def test_user_sink_input_ports_getitem_invalid_key(self
):
253 class MySink(bt2
._UserSinkComponent
):
254 def __init__(comp_self
, config
, params
, obj
):
255 comp_self
._add
_input
_port
("clear")
256 comp_self
._add
_input
_port
("print")
257 comp_self
._add
_input
_port
("insert")
259 with self
.assertRaises(KeyError):
260 comp_self
._input
_ports
["hello"]
262 def _user_consume(self
):
265 self
._create
_comp
(MySink
)
267 def test_user_src_output_ports_len(self
):
269 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
271 def __init__(comp_self
, config
, params
, obj
):
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)
277 self
._create
_comp
(MySource
)
279 def test_user_flt_output_ports_len(self
):
281 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
283 def __init__(comp_self
, config
, params
, obj
):
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)
289 self
._create
_comp
(MyFilter
)
291 def test_user_flt_input_ports_len(self
):
293 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
295 def __init__(comp_self
, config
, params
, obj
):
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)
301 self
._create
_comp
(MyFilter
)
303 def test_user_sink_input_ports_len(self
):
304 class MySink(bt2
._UserSinkComponent
):
305 def __init__(comp_self
, config
, params
, obj
):
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)
311 def _user_consume(self
):
314 self
._create
_comp
(MySink
)
316 def test_user_src_output_ports_iter(self
):
318 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
320 def __init__(comp_self
, config
, params
, obj
):
321 port1
= comp_self
._add
_output
_port
("clear")
322 port2
= comp_self
._add
_output
_port
("print")
323 port3
= comp_self
._add
_output
_port
("insert")
326 for port_name
, port
in comp_self
._output
_ports
.items():
327 ports
.append((port_name
, port
))
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
)
336 self
._create
_comp
(MySource
)
338 def test_user_flt_output_ports_iter(self
):
340 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
342 def __init__(comp_self
, config
, params
, obj
):
343 port1
= comp_self
._add
_output
_port
("clear")
344 port2
= comp_self
._add
_output
_port
("print")
345 port3
= comp_self
._add
_output
_port
("insert")
348 for port_name
, port
in comp_self
._output
_ports
.items():
349 ports
.append((port_name
, port
))
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
)
358 self
._create
_comp
(MyFilter
)
360 def test_user_flt_input_ports_iter(self
):
362 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
364 def __init__(comp_self
, config
, params
, obj
):
365 port1
= comp_self
._add
_input
_port
("clear")
366 port2
= comp_self
._add
_input
_port
("print")
367 port3
= comp_self
._add
_input
_port
("insert")
370 for port_name
, port
in comp_self
._input
_ports
.items():
371 ports
.append((port_name
, port
))
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
)
380 self
._create
_comp
(MyFilter
)
382 def test_user_sink_input_ports_iter(self
):
383 class MySink(bt2
._UserSinkComponent
):
384 def __init__(comp_self
, config
, params
, obj
):
385 port1
= comp_self
._add
_input
_port
("clear")
386 port2
= comp_self
._add
_input
_port
("print")
387 port3
= comp_self
._add
_input
_port
("insert")
390 for port_name
, port
in comp_self
._input
_ports
.items():
391 ports
.append((port_name
, port
))
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
)
400 def _user_consume(self
):
403 self
._create
_comp
(MySink
)
405 def test_gen_src_output_ports_getitem(self
):
411 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
413 def __init__(comp_self
, config
, params
, obj
):
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")
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
)
427 def test_gen_flt_output_ports_getitem(self
):
433 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
435 def __init__(comp_self
, config
, params
, obj
):
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")
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
)
449 def test_gen_flt_input_ports_getitem(self
):
455 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
457 def __init__(comp_self
, config
, params
, obj
):
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")
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
)
471 def test_gen_sink_input_ports_getitem(self
):
476 class MySink(bt2
._UserSinkComponent
):
477 def __init__(comp_self
, config
, params
, obj
):
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")
483 def _user_consume(self
):
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
)
494 def test_gen_src_output_ports_getitem_invalid_key(self
):
496 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
498 def __init__(comp_self
, config
, params
, obj
):
499 comp_self
._add
_output
_port
("clear")
500 comp_self
._add
_output
_port
("print")
501 comp_self
._add
_output
_port
("insert")
503 comp
= self
._create
_comp
(MySource
)
505 with self
.assertRaises(KeyError):
506 comp
.output_ports
["hello"]
508 def test_gen_flt_output_ports_getitem_invalid_key(self
):
510 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
512 def __init__(comp_self
, config
, params
, obj
):
513 comp_self
._add
_output
_port
("clear")
514 comp_self
._add
_output
_port
("print")
515 comp_self
._add
_output
_port
("insert")
517 comp
= self
._create
_comp
(MyFilter
)
519 with self
.assertRaises(KeyError):
520 comp
.output_ports
["hello"]
522 def test_gen_flt_input_ports_getitem_invalid_key(self
):
524 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
526 def __init__(comp_self
, config
, params
, obj
):
527 comp_self
._add
_input
_port
("clear")
528 comp_self
._add
_input
_port
("print")
529 comp_self
._add
_input
_port
("insert")
531 comp
= self
._create
_comp
(MyFilter
)
533 with self
.assertRaises(KeyError):
534 comp
.input_ports
["hello"]
536 def test_gen_sink_input_ports_getitem_invalid_key(self
):
537 class MySink(bt2
._UserSinkComponent
):
538 def __init__(comp_self
, config
, params
, obj
):
539 comp_self
._add
_input
_port
("clear")
540 comp_self
._add
_input
_port
("print")
541 comp_self
._add
_input
_port
("insert")
543 with self
.assertRaises(KeyError):
544 comp_self
._input
_ports
["hello"]
546 def _user_consume(self
):
549 comp
= self
._create
_comp
(MySink
)
551 with self
.assertRaises(KeyError):
552 comp
.input_ports
["hello"]
554 def test_gen_src_output_ports_len(self
):
556 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
558 def __init__(comp_self
, config
, params
, obj
):
559 comp_self
._add
_output
_port
("clear")
560 comp_self
._add
_output
_port
("print")
561 comp_self
._add
_output
_port
("insert")
563 comp
= self
._create
_comp
(MySource
)
564 self
.assertEqual(len(comp
.output_ports
), 3)
566 def test_gen_flt_output_ports_len(self
):
568 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
570 def __init__(comp_self
, config
, params
, obj
):
571 comp_self
._add
_output
_port
("clear")
572 comp_self
._add
_output
_port
("print")
573 comp_self
._add
_output
_port
("insert")
575 comp
= self
._create
_comp
(MyFilter
)
576 self
.assertEqual(len(comp
.output_ports
), 3)
578 def test_gen_flt_input_ports_len(self
):
580 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
582 def __init__(comp_self
, config
, params
, obj
):
583 comp_self
._add
_input
_port
("clear")
584 comp_self
._add
_input
_port
("print")
585 comp_self
._add
_input
_port
("insert")
587 comp
= self
._create
_comp
(MyFilter
)
588 self
.assertEqual(len(comp
.input_ports
), 3)
590 def test_gen_sink_input_ports_len(self
):
591 class MySink(bt2
._UserSinkComponent
):
592 def __init__(comp_self
, config
, params
, obj
):
593 comp_self
._add
_input
_port
("clear")
594 comp_self
._add
_input
_port
("print")
595 comp_self
._add
_input
_port
("insert")
597 def _user_consume(self
):
600 comp
= self
._create
_comp
(MySink
)
601 self
.assertEqual(len(comp
.input_ports
), 3)
603 def test_gen_src_output_ports_iter(self
):
609 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
611 def __init__(comp_self
, config
, params
, obj
):
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")
617 comp
= self
._create
_comp
(MySource
)
620 for port_name
, port
in comp
.output_ports
.items():
621 ports
.append((port_name
, port
))
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
)
633 def test_gen_flt_output_ports_iter(self
):
639 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
641 def __init__(comp_self
, config
, params
, obj
):
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")
647 comp
= self
._create
_comp
(MyFilter
)
650 for port_name
, port
in comp
.output_ports
.items():
651 ports
.append((port_name
, port
))
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
)
663 def test_gen_flt_input_ports_iter(self
):
669 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
671 def __init__(comp_self
, config
, params
, obj
):
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")
677 comp
= self
._create
_comp
(MyFilter
)
680 for port_name
, port
in comp
.input_ports
.items():
681 ports
.append((port_name
, port
))
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
)
693 def test_gen_sink_input_ports_iter(self
):
698 class MySink(bt2
._UserSinkComponent
):
699 def __init__(comp_self
, config
, params
, obj
):
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")
705 def _user_consume(self
):
708 comp
= self
._create
_comp
(MySink
)
711 for port_name
, port
in comp
.input_ports
.items():
712 ports
.append((port_name
, port
))
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
)
725 class MySink(bt2
._UserSinkComponent
):
726 def __init__(comp_self
, config
, params
, obj
):
727 comp_self
._add
_input
_port
("clear")
729 def _user_consume(self
):
732 comp
= self
._create
_comp
(MySink
)
733 self
.assertEqual(comp
.input_ports
["clear"].name
, "clear")
735 def test_connection_none(self
):
736 class MySink(bt2
._UserSinkComponent
):
737 def __init__(comp_self
, config
, params
, obj
):
738 comp_self
._add
_input
_port
("clear")
740 def _user_consume(self
):
743 comp
= self
._create
_comp
(MySink
)
744 self
.assertIsNone(comp
.input_ports
["clear"].connection
)
746 def test_is_connected_false(self
):
747 class MySink(bt2
._UserSinkComponent
):
748 def __init__(comp_self
, config
, params
, obj
):
749 comp_self
._add
_input
_port
("clear")
751 def _user_consume(self
):
754 comp
= self
._create
_comp
(MySink
)
755 self
.assertFalse(comp
.input_ports
["clear"].is_connected
)
757 def test_self_name(self
):
758 class MySink(bt2
._UserSinkComponent
):
759 def __init__(comp_self
, config
, params
, obj
):
760 port
= comp_self
._add
_input
_port
("clear")
761 self
.assertEqual(port
.name
, "clear")
763 def _user_consume(self
):
766 self
._create
_comp
(MySink
)
768 def test_self_connection_none(self
):
769 class MySink(bt2
._UserSinkComponent
):
770 def __init__(comp_self
, config
, params
, obj
):
771 port
= comp_self
._add
_input
_port
("clear")
772 self
.assertIsNone(port
.connection
)
774 def _user_consume(self
):
777 self
._create
_comp
(MySink
)
779 def test_self_is_connected_false(self
):
780 class MySink(bt2
._UserSinkComponent
):
781 def __init__(comp_self
, config
, params
, obj
):
782 port
= comp_self
._add
_input
_port
("clear")
783 self
.assertFalse(port
.is_connected
)
785 def _user_consume(self
):
788 self
._create
_comp
(MySink
)
790 def test_source_self_port_user_data(self
):
793 nonlocal objects_deleted
797 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
799 def __init__(comp_self
, config
, params
, obj
):
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
)
806 p
= comp_self
._add
_output
_port
("port3", MyUserData())
807 user_datas
.append(p
.user_data
)
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
)
818 # Verify that the user data gets freed.
819 self
.assertEqual(objects_deleted
, 0)
822 self
.assertEqual(objects_deleted
, 1)
824 def test_filter_self_port_user_data(self
):
827 nonlocal objects_deleted
831 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
833 def __init__(comp_self
, config
, params
, obj
):
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
)
840 p
= comp_self
._add
_output
_port
("port3", MyUserData())
841 user_datas
.append(p
.user_data
)
843 p
= comp_self
._add
_input
_port
("port4")
844 user_datas
.append(p
.user_data
)
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())
848 user_datas
.append(p
.user_data
)
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
)
862 # Verify that the user data gets freed.
863 self
.assertEqual(objects_deleted
, 0)
866 self
.assertEqual(objects_deleted
, 2)
868 def test_sink_self_port_user_data(self
):
871 nonlocal objects_deleted
874 class MySink(bt2
._UserSinkComponent
):
875 def __init__(comp_self
, config
, params
, obj
):
878 p
= comp_self
._add
_input
_port
("port1")
879 user_datas
.append(p
.user_data
)
880 p
= comp_self
._add
_input
_port
("port2", MyUserData())
881 user_datas
.append(p
.user_data
)
883 def _user_consume(self
):
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
)
894 # Verify that the user data gets freed.
895 self
.assertEqual(objects_deleted
, 0)
898 self
.assertEqual(objects_deleted
, 1)
901 if __name__
== "__main__":
This page took 0.051775 seconds and 5 git commands to generate.