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.084514 seconds and 4 git commands to generate.