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 def test_flt_add_output_port(self
):
35 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
37 def __init__(comp_self
, config
, params
, obj
):
38 port
= comp_self
._add
_output
_port
('out')
39 self
.assertEqual(port
.name
, 'out')
41 comp
= self
._create
_comp
(MyFilter
)
42 self
.assertEqual(len(comp
.output_ports
), 1)
44 def test_flt_add_input_port(self
):
46 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
48 def __init__(comp_self
, config
, params
, obj
):
49 port
= comp_self
._add
_input
_port
('in')
50 self
.assertEqual(port
.name
, 'in')
52 comp
= self
._create
_comp
(MyFilter
)
53 self
.assertEqual(len(comp
.input_ports
), 1)
54 self
.assertIs(type(comp
.input_ports
['in']), bt2_port
._InputPortConst
)
56 def test_sink_add_input_port(self
):
57 class MySink(bt2
._UserSinkComponent
):
58 def __init__(comp_self
, config
, params
, obj
):
59 port
= comp_self
._add
_input
_port
('in')
60 self
.assertEqual(port
.name
, 'in')
62 def _user_consume(self
):
65 comp
= self
._create
_comp
(MySink
)
66 self
.assertEqual(len(comp
.input_ports
), 1)
68 def test_user_src_output_ports_getitem(self
):
70 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
72 def __init__(comp_self
, config
, params
, obj
):
73 port1
= comp_self
._add
_output
_port
('clear')
74 port2
= comp_self
._add
_output
_port
('print')
75 port3
= comp_self
._add
_output
_port
('insert')
76 self
.assertEqual(port3
.addr
, comp_self
._output
_ports
['insert'].addr
)
77 self
.assertEqual(port2
.addr
, comp_self
._output
_ports
['print'].addr
)
78 self
.assertEqual(port1
.addr
, comp_self
._output
_ports
['clear'].addr
)
80 self
._create
_comp
(MySource
)
82 def test_user_flt_output_ports_getitem(self
):
84 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
86 def __init__(comp_self
, config
, params
, obj
):
87 port1
= comp_self
._add
_output
_port
('clear')
88 port2
= comp_self
._add
_output
_port
('print')
89 port3
= comp_self
._add
_output
_port
('insert')
90 self
.assertEqual(port3
.addr
, comp_self
._output
_ports
['insert'].addr
)
91 self
.assertEqual(port2
.addr
, comp_self
._output
_ports
['print'].addr
)
92 self
.assertEqual(port1
.addr
, comp_self
._output
_ports
['clear'].addr
)
94 self
._create
_comp
(MyFilter
)
96 def test_user_flt_input_ports_getitem(self
):
98 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
100 def __init__(comp_self
, config
, params
, obj
):
101 port1
= comp_self
._add
_input
_port
('clear')
102 port2
= comp_self
._add
_input
_port
('print')
103 port3
= comp_self
._add
_input
_port
('insert')
104 self
.assertEqual(port3
.addr
, comp_self
._input
_ports
['insert'].addr
)
105 self
.assertEqual(port2
.addr
, comp_self
._input
_ports
['print'].addr
)
106 self
.assertEqual(port1
.addr
, comp_self
._input
_ports
['clear'].addr
)
108 self
._create
_comp
(MyFilter
)
110 def test_user_sink_input_ports_getitem(self
):
111 class MySink(bt2
._UserSinkComponent
):
112 def __init__(comp_self
, config
, params
, obj
):
113 port1
= comp_self
._add
_input
_port
('clear')
114 port2
= comp_self
._add
_input
_port
('print')
115 port3
= comp_self
._add
_input
_port
('insert')
116 self
.assertEqual(port3
.addr
, comp_self
._input
_ports
['insert'].addr
)
117 self
.assertEqual(port2
.addr
, comp_self
._input
_ports
['print'].addr
)
118 self
.assertEqual(port1
.addr
, comp_self
._input
_ports
['clear'].addr
)
120 def _user_consume(self
):
123 self
._create
_comp
(MySink
)
125 def test_user_src_output_ports_getitem_invalid_key(self
):
127 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
129 def __init__(comp_self
, config
, params
, obj
):
130 comp_self
._add
_output
_port
('clear')
131 comp_self
._add
_output
_port
('print')
132 comp_self
._add
_output
_port
('insert')
134 with self
.assertRaises(KeyError):
135 comp_self
._output
_ports
['hello']
137 self
._create
_comp
(MySource
)
139 def test_user_flt_output_ports_getitem_invalid_key(self
):
141 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
143 def __init__(comp_self
, config
, params
, obj
):
144 comp_self
._add
_output
_port
('clear')
145 comp_self
._add
_output
_port
('print')
146 comp_self
._add
_output
_port
('insert')
148 with self
.assertRaises(KeyError):
149 comp_self
._output
_ports
['hello']
151 self
._create
_comp
(MyFilter
)
153 def test_user_flt_input_ports_getitem_invalid_key(self
):
155 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
157 def __init__(comp_self
, config
, params
, obj
):
158 comp_self
._add
_input
_port
('clear')
159 comp_self
._add
_input
_port
('print')
160 comp_self
._add
_input
_port
('insert')
162 with self
.assertRaises(KeyError):
163 comp_self
._input
_ports
['hello']
165 self
._create
_comp
(MyFilter
)
167 def test_user_sink_input_ports_getitem_invalid_key(self
):
168 class MySink(bt2
._UserSinkComponent
):
169 def __init__(comp_self
, config
, params
, obj
):
170 comp_self
._add
_input
_port
('clear')
171 comp_self
._add
_input
_port
('print')
172 comp_self
._add
_input
_port
('insert')
174 with self
.assertRaises(KeyError):
175 comp_self
._input
_ports
['hello']
177 def _user_consume(self
):
180 self
._create
_comp
(MySink
)
182 def test_user_src_output_ports_len(self
):
184 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
186 def __init__(comp_self
, config
, params
, obj
):
187 comp_self
._add
_output
_port
('clear')
188 comp_self
._add
_output
_port
('print')
189 comp_self
._add
_output
_port
('insert')
190 self
.assertEqual(len(comp_self
._output
_ports
), 3)
192 self
._create
_comp
(MySource
)
194 def test_user_flt_output_ports_len(self
):
196 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
198 def __init__(comp_self
, config
, params
, obj
):
199 comp_self
._add
_output
_port
('clear')
200 comp_self
._add
_output
_port
('print')
201 comp_self
._add
_output
_port
('insert')
202 self
.assertEqual(len(comp_self
._output
_ports
), 3)
204 self
._create
_comp
(MyFilter
)
206 def test_user_flt_input_ports_len(self
):
208 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
210 def __init__(comp_self
, config
, params
, obj
):
211 comp_self
._add
_input
_port
('clear')
212 comp_self
._add
_input
_port
('print')
213 comp_self
._add
_input
_port
('insert')
214 self
.assertEqual(len(comp_self
._input
_ports
), 3)
216 self
._create
_comp
(MyFilter
)
218 def test_user_sink_input_ports_len(self
):
219 class MySink(bt2
._UserSinkComponent
):
220 def __init__(comp_self
, config
, params
, obj
):
221 comp_self
._add
_input
_port
('clear')
222 comp_self
._add
_input
_port
('print')
223 comp_self
._add
_input
_port
('insert')
224 self
.assertEqual(len(comp_self
._input
_ports
), 3)
226 def _user_consume(self
):
229 self
._create
_comp
(MySink
)
231 def test_user_src_output_ports_iter(self
):
233 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
235 def __init__(comp_self
, config
, params
, obj
):
236 port1
= comp_self
._add
_output
_port
('clear')
237 port2
= comp_self
._add
_output
_port
('print')
238 port3
= comp_self
._add
_output
_port
('insert')
241 for port_name
, port
in comp_self
._output
_ports
.items():
242 ports
.append((port_name
, port
))
244 self
.assertEqual(ports
[0][0], 'clear')
245 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
246 self
.assertEqual(ports
[1][0], 'print')
247 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
248 self
.assertEqual(ports
[2][0], 'insert')
249 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
251 self
._create
_comp
(MySource
)
253 def test_user_flt_output_ports_iter(self
):
255 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
257 def __init__(comp_self
, config
, params
, obj
):
258 port1
= comp_self
._add
_output
_port
('clear')
259 port2
= comp_self
._add
_output
_port
('print')
260 port3
= comp_self
._add
_output
_port
('insert')
263 for port_name
, port
in comp_self
._output
_ports
.items():
264 ports
.append((port_name
, port
))
266 self
.assertEqual(ports
[0][0], 'clear')
267 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
268 self
.assertEqual(ports
[1][0], 'print')
269 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
270 self
.assertEqual(ports
[2][0], 'insert')
271 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
273 self
._create
_comp
(MyFilter
)
275 def test_user_flt_input_ports_iter(self
):
277 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
279 def __init__(comp_self
, config
, params
, obj
):
280 port1
= comp_self
._add
_input
_port
('clear')
281 port2
= comp_self
._add
_input
_port
('print')
282 port3
= comp_self
._add
_input
_port
('insert')
285 for port_name
, port
in comp_self
._input
_ports
.items():
286 ports
.append((port_name
, port
))
288 self
.assertEqual(ports
[0][0], 'clear')
289 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
290 self
.assertEqual(ports
[1][0], 'print')
291 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
292 self
.assertEqual(ports
[2][0], 'insert')
293 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
295 self
._create
_comp
(MyFilter
)
297 def test_user_sink_input_ports_iter(self
):
298 class MySink(bt2
._UserSinkComponent
):
299 def __init__(comp_self
, config
, params
, obj
):
300 port1
= comp_self
._add
_input
_port
('clear')
301 port2
= comp_self
._add
_input
_port
('print')
302 port3
= comp_self
._add
_input
_port
('insert')
305 for port_name
, port
in comp_self
._input
_ports
.items():
306 ports
.append((port_name
, port
))
308 self
.assertEqual(ports
[0][0], 'clear')
309 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
310 self
.assertEqual(ports
[1][0], 'print')
311 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
312 self
.assertEqual(ports
[2][0], 'insert')
313 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
315 def _user_consume(self
):
318 self
._create
_comp
(MySink
)
320 def test_gen_src_output_ports_getitem(self
):
326 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
328 def __init__(comp_self
, config
, params
, obj
):
329 nonlocal port1
, port2
, port3
330 port1
= comp_self
._add
_output
_port
('clear')
331 port2
= comp_self
._add
_output
_port
('print')
332 port3
= comp_self
._add
_output
_port
('insert')
334 comp
= self
._create
_comp
(MySource
)
335 self
.assertEqual(port3
.addr
, comp
.output_ports
['insert'].addr
)
336 self
.assertEqual(port2
.addr
, comp
.output_ports
['print'].addr
)
337 self
.assertEqual(port1
.addr
, comp
.output_ports
['clear'].addr
)
342 def test_gen_flt_output_ports_getitem(self
):
348 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
350 def __init__(comp_self
, config
, params
, obj
):
351 nonlocal port1
, port2
, port3
352 port1
= comp_self
._add
_output
_port
('clear')
353 port2
= comp_self
._add
_output
_port
('print')
354 port3
= comp_self
._add
_output
_port
('insert')
356 comp
= self
._create
_comp
(MyFilter
)
357 self
.assertEqual(port3
.addr
, comp
.output_ports
['insert'].addr
)
358 self
.assertEqual(port2
.addr
, comp
.output_ports
['print'].addr
)
359 self
.assertEqual(port1
.addr
, comp
.output_ports
['clear'].addr
)
364 def test_gen_flt_input_ports_getitem(self
):
370 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
372 def __init__(comp_self
, config
, params
, obj
):
373 nonlocal port1
, port2
, port3
374 port1
= comp_self
._add
_input
_port
('clear')
375 port2
= comp_self
._add
_input
_port
('print')
376 port3
= comp_self
._add
_input
_port
('insert')
378 comp
= self
._create
_comp
(MyFilter
)
379 self
.assertEqual(port3
.addr
, comp
.input_ports
['insert'].addr
)
380 self
.assertEqual(port2
.addr
, comp
.input_ports
['print'].addr
)
381 self
.assertEqual(port1
.addr
, comp
.input_ports
['clear'].addr
)
386 def test_gen_sink_input_ports_getitem(self
):
391 class MySink(bt2
._UserSinkComponent
):
392 def __init__(comp_self
, config
, params
, obj
):
393 nonlocal port1
, port2
, port3
394 port1
= comp_self
._add
_input
_port
('clear')
395 port2
= comp_self
._add
_input
_port
('print')
396 port3
= comp_self
._add
_input
_port
('insert')
398 def _user_consume(self
):
401 comp
= self
._create
_comp
(MySink
)
402 self
.assertEqual(port3
.addr
, comp
.input_ports
['insert'].addr
)
403 self
.assertEqual(port2
.addr
, comp
.input_ports
['print'].addr
)
404 self
.assertEqual(port1
.addr
, comp
.input_ports
['clear'].addr
)
409 def test_gen_src_output_ports_getitem_invalid_key(self
):
411 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
413 def __init__(comp_self
, config
, params
, obj
):
414 comp_self
._add
_output
_port
('clear')
415 comp_self
._add
_output
_port
('print')
416 comp_self
._add
_output
_port
('insert')
418 comp
= self
._create
_comp
(MySource
)
420 with self
.assertRaises(KeyError):
421 comp
.output_ports
['hello']
423 def test_gen_flt_output_ports_getitem_invalid_key(self
):
425 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
427 def __init__(comp_self
, config
, params
, obj
):
428 comp_self
._add
_output
_port
('clear')
429 comp_self
._add
_output
_port
('print')
430 comp_self
._add
_output
_port
('insert')
432 comp
= self
._create
_comp
(MyFilter
)
434 with self
.assertRaises(KeyError):
435 comp
.output_ports
['hello']
437 def test_gen_flt_input_ports_getitem_invalid_key(self
):
439 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
441 def __init__(comp_self
, config
, params
, obj
):
442 comp_self
._add
_input
_port
('clear')
443 comp_self
._add
_input
_port
('print')
444 comp_self
._add
_input
_port
('insert')
446 comp
= self
._create
_comp
(MyFilter
)
448 with self
.assertRaises(KeyError):
449 comp
.input_ports
['hello']
451 def test_gen_sink_input_ports_getitem_invalid_key(self
):
452 class MySink(bt2
._UserSinkComponent
):
453 def __init__(comp_self
, config
, params
, obj
):
454 comp_self
._add
_input
_port
('clear')
455 comp_self
._add
_input
_port
('print')
456 comp_self
._add
_input
_port
('insert')
458 with self
.assertRaises(KeyError):
459 comp_self
._input
_ports
['hello']
461 def _user_consume(self
):
464 comp
= self
._create
_comp
(MySink
)
466 with self
.assertRaises(KeyError):
467 comp
.input_ports
['hello']
469 def test_gen_src_output_ports_len(self
):
471 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
473 def __init__(comp_self
, config
, params
, obj
):
474 comp_self
._add
_output
_port
('clear')
475 comp_self
._add
_output
_port
('print')
476 comp_self
._add
_output
_port
('insert')
478 comp
= self
._create
_comp
(MySource
)
479 self
.assertEqual(len(comp
.output_ports
), 3)
481 def test_gen_flt_output_ports_len(self
):
483 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
485 def __init__(comp_self
, config
, params
, obj
):
486 comp_self
._add
_output
_port
('clear')
487 comp_self
._add
_output
_port
('print')
488 comp_self
._add
_output
_port
('insert')
490 comp
= self
._create
_comp
(MyFilter
)
491 self
.assertEqual(len(comp
.output_ports
), 3)
493 def test_gen_flt_input_ports_len(self
):
495 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
497 def __init__(comp_self
, config
, params
, obj
):
498 comp_self
._add
_input
_port
('clear')
499 comp_self
._add
_input
_port
('print')
500 comp_self
._add
_input
_port
('insert')
502 comp
= self
._create
_comp
(MyFilter
)
503 self
.assertEqual(len(comp
.input_ports
), 3)
505 def test_gen_sink_input_ports_len(self
):
506 class MySink(bt2
._UserSinkComponent
):
507 def __init__(comp_self
, config
, params
, obj
):
508 comp_self
._add
_input
_port
('clear')
509 comp_self
._add
_input
_port
('print')
510 comp_self
._add
_input
_port
('insert')
512 def _user_consume(self
):
515 comp
= self
._create
_comp
(MySink
)
516 self
.assertEqual(len(comp
.input_ports
), 3)
518 def test_gen_src_output_ports_iter(self
):
524 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
526 def __init__(comp_self
, config
, params
, obj
):
527 nonlocal port1
, port2
, port3
528 port1
= comp_self
._add
_output
_port
('clear')
529 port2
= comp_self
._add
_output
_port
('print')
530 port3
= comp_self
._add
_output
_port
('insert')
532 comp
= self
._create
_comp
(MySource
)
535 for port_name
, port
in comp
.output_ports
.items():
536 ports
.append((port_name
, port
))
538 self
.assertEqual(ports
[0][0], 'clear')
539 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
540 self
.assertEqual(ports
[1][0], 'print')
541 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
542 self
.assertEqual(ports
[2][0], 'insert')
543 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
548 def test_gen_flt_output_ports_iter(self
):
554 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
556 def __init__(comp_self
, config
, params
, obj
):
557 nonlocal port1
, port2
, port3
558 port1
= comp_self
._add
_output
_port
('clear')
559 port2
= comp_self
._add
_output
_port
('print')
560 port3
= comp_self
._add
_output
_port
('insert')
562 comp
= self
._create
_comp
(MyFilter
)
565 for port_name
, port
in comp
.output_ports
.items():
566 ports
.append((port_name
, port
))
568 self
.assertEqual(ports
[0][0], 'clear')
569 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
570 self
.assertEqual(ports
[1][0], 'print')
571 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
572 self
.assertEqual(ports
[2][0], 'insert')
573 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
578 def test_gen_flt_input_ports_iter(self
):
584 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
586 def __init__(comp_self
, config
, params
, obj
):
587 nonlocal port1
, port2
, port3
588 port1
= comp_self
._add
_input
_port
('clear')
589 port2
= comp_self
._add
_input
_port
('print')
590 port3
= comp_self
._add
_input
_port
('insert')
592 comp
= self
._create
_comp
(MyFilter
)
595 for port_name
, port
in comp
.input_ports
.items():
596 ports
.append((port_name
, port
))
598 self
.assertEqual(ports
[0][0], 'clear')
599 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
600 self
.assertEqual(ports
[1][0], 'print')
601 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
602 self
.assertEqual(ports
[2][0], 'insert')
603 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
608 def test_gen_sink_input_ports_iter(self
):
613 class MySink(bt2
._UserSinkComponent
):
614 def __init__(comp_self
, config
, params
, obj
):
615 nonlocal port1
, port2
, port3
616 port1
= comp_self
._add
_input
_port
('clear')
617 port2
= comp_self
._add
_input
_port
('print')
618 port3
= comp_self
._add
_input
_port
('insert')
620 def _user_consume(self
):
623 comp
= self
._create
_comp
(MySink
)
626 for port_name
, port
in comp
.input_ports
.items():
627 ports
.append((port_name
, port
))
629 self
.assertEqual(ports
[0][0], 'clear')
630 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
631 self
.assertEqual(ports
[1][0], 'print')
632 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
633 self
.assertEqual(ports
[2][0], 'insert')
634 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
640 class MySink(bt2
._UserSinkComponent
):
641 def __init__(comp_self
, config
, params
, obj
):
642 comp_self
._add
_input
_port
('clear')
644 def _user_consume(self
):
647 comp
= self
._create
_comp
(MySink
)
648 self
.assertEqual(comp
.input_ports
['clear'].name
, 'clear')
650 def test_connection_none(self
):
651 class MySink(bt2
._UserSinkComponent
):
652 def __init__(comp_self
, config
, params
, obj
):
653 comp_self
._add
_input
_port
('clear')
655 def _user_consume(self
):
658 comp
= self
._create
_comp
(MySink
)
659 self
.assertIsNone(comp
.input_ports
['clear'].connection
)
661 def test_is_connected_false(self
):
662 class MySink(bt2
._UserSinkComponent
):
663 def __init__(comp_self
, config
, params
, obj
):
664 comp_self
._add
_input
_port
('clear')
666 def _user_consume(self
):
669 comp
= self
._create
_comp
(MySink
)
670 self
.assertFalse(comp
.input_ports
['clear'].is_connected
)
672 def test_self_name(self
):
673 class MySink(bt2
._UserSinkComponent
):
674 def __init__(comp_self
, config
, params
, obj
):
675 port
= comp_self
._add
_input
_port
('clear')
676 self
.assertEqual(port
.name
, 'clear')
678 def _user_consume(self
):
681 self
._create
_comp
(MySink
)
683 def test_self_connection_none(self
):
684 class MySink(bt2
._UserSinkComponent
):
685 def __init__(comp_self
, config
, params
, obj
):
686 port
= comp_self
._add
_input
_port
('clear')
687 self
.assertIsNone(port
.connection
)
689 def _user_consume(self
):
692 self
._create
_comp
(MySink
)
694 def test_self_is_connected_false(self
):
695 class MySink(bt2
._UserSinkComponent
):
696 def __init__(comp_self
, config
, params
, obj
):
697 port
= comp_self
._add
_input
_port
('clear')
698 self
.assertFalse(port
.is_connected
)
700 def _user_consume(self
):
703 self
._create
_comp
(MySink
)
705 def test_source_self_port_user_data(self
):
708 nonlocal objects_deleted
712 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
714 def __init__(comp_self
, config
, params
, obj
):
717 p
= comp_self
._add
_output
_port
('port1')
718 user_datas
.append(p
.user_data
)
719 p
= comp_self
._add
_output
_port
('port2', 2)
720 user_datas
.append(p
.user_data
)
721 p
= comp_self
._add
_output
_port
('port3', MyUserData())
722 user_datas
.append(p
.user_data
)
727 comp
= self
._create
_comp
(MySource
)
728 self
.assertEqual(len(user_datas
), 3)
729 self
.assertIs(user_datas
[0], None)
730 self
.assertEqual(user_datas
[1], 2)
731 self
.assertIs(type(user_datas
[2]), MyUserData
)
733 # Verify that the user data gets freed.
734 self
.assertEqual(objects_deleted
, 0)
737 self
.assertEqual(objects_deleted
, 1)
739 def test_filter_self_port_user_data(self
):
742 nonlocal objects_deleted
746 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
748 def __init__(comp_self
, config
, params
, obj
):
751 p
= comp_self
._add
_output
_port
('port1')
752 user_datas
.append(p
.user_data
)
753 p
= comp_self
._add
_output
_port
('port2', 'user data string')
754 user_datas
.append(p
.user_data
)
755 p
= comp_self
._add
_output
_port
('port3', MyUserData())
756 user_datas
.append(p
.user_data
)
758 p
= comp_self
._add
_input
_port
('port4')
759 user_datas
.append(p
.user_data
)
760 p
= comp_self
._add
_input
_port
('port5', user_data
={'user data': 'dict'})
761 user_datas
.append(p
.user_data
)
762 p
= comp_self
._add
_input
_port
('port6', MyUserData())
763 user_datas
.append(p
.user_data
)
768 comp
= self
._create
_comp
(MyFilter
)
769 self
.assertEqual(len(user_datas
), 6)
770 self
.assertIs(user_datas
[0], None)
771 self
.assertEqual(user_datas
[1], 'user data string')
772 self
.assertIs(type(user_datas
[2]), MyUserData
)
773 self
.assertIs(user_datas
[3], None)
774 self
.assertEqual(user_datas
[4], {'user data': 'dict'})
775 self
.assertIs(type(user_datas
[5]), MyUserData
)
777 # Verify that the user data gets freed.
778 self
.assertEqual(objects_deleted
, 0)
781 self
.assertEqual(objects_deleted
, 2)
783 def test_sink_self_port_user_data(self
):
786 nonlocal objects_deleted
789 class MySink(bt2
._UserSinkComponent
):
790 def __init__(comp_self
, config
, params
, obj
):
793 p
= comp_self
._add
_input
_port
('port1')
794 user_datas
.append(p
.user_data
)
795 p
= comp_self
._add
_input
_port
('port2', MyUserData())
796 user_datas
.append(p
.user_data
)
798 def _user_consume(self
):
804 comp
= self
._create
_comp
(MySink
)
805 self
.assertEqual(len(user_datas
), 2)
806 self
.assertIs(user_datas
[0], None)
807 self
.assertIs(type(user_datas
[1]), MyUserData
)
809 # Verify that the user data gets freed.
810 self
.assertEqual(objects_deleted
, 0)
813 self
.assertEqual(objects_deleted
, 1)
816 if __name__
== '__main__':
This page took 0.050936 seconds and 4 git commands to generate.