2 # Copyright (C) 2019 EfficiOS Inc.
4 # This program is free software; you can redistribute it and/or
5 # modify it under the terms of the GNU General Public License
6 # as published by the Free Software Foundation; only version 2
9 # This program is distributed in the hope that it will be useful,
10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 # GNU General Public License for more details.
14 # You should have received a copy of the GNU General Public License
15 # along with this program; if not, write to the Free Software
16 # Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
23 class PortTestCase(unittest
.TestCase
):
25 def _create_comp(comp_cls
, name
=None):
31 return graph
.add_component(comp_cls
, name
)
33 def test_src_add_output_port(self
):
34 class MyIter(bt2
._UserMessageIterator
):
38 class MySource(bt2
._UserSourceComponent
,
39 message_iterator_class
=MyIter
):
40 def __init__(comp_self
, params
):
41 port
= comp_self
._add
_output
_port
('out')
42 self
.assertEqual(port
.name
, 'out')
44 comp
= self
._create
_comp
(MySource
)
45 self
.assertEqual(len(comp
.output_ports
), 1)
47 def test_flt_add_output_port(self
):
48 class MyIter(bt2
._UserMessageIterator
):
52 class MyFilter(bt2
._UserFilterComponent
,
53 message_iterator_class
=MyIter
):
54 def __init__(comp_self
, params
):
55 port
= comp_self
._add
_output
_port
('out')
56 self
.assertEqual(port
.name
, 'out')
58 comp
= self
._create
_comp
(MyFilter
)
59 self
.assertEqual(len(comp
.output_ports
), 1)
61 def test_flt_add_input_port(self
):
62 class MyIter(bt2
._UserMessageIterator
):
66 class MyFilter(bt2
._UserFilterComponent
,
67 message_iterator_class
=MyIter
):
68 def __init__(comp_self
, params
):
69 port
= comp_self
._add
_input
_port
('in')
70 self
.assertEqual(port
.name
, 'in')
72 comp
= self
._create
_comp
(MyFilter
)
73 self
.assertEqual(len(comp
.input_ports
), 1)
75 def test_sink_add_input_port(self
):
76 class MySink(bt2
._UserSinkComponent
):
77 def __init__(comp_self
, params
):
78 port
= comp_self
._add
_input
_port
('in')
79 self
.assertEqual(port
.name
, 'in')
84 comp
= self
._create
_comp
(MySink
)
85 self
.assertEqual(len(comp
.input_ports
), 1)
87 def test_user_src_output_ports_getitem(self
):
88 class MyIter(bt2
._UserMessageIterator
):
92 class MySource(bt2
._UserSourceComponent
,
93 message_iterator_class
=MyIter
):
94 def __init__(comp_self
, params
):
95 port1
= comp_self
._add
_output
_port
('clear')
96 port2
= comp_self
._add
_output
_port
('print')
97 port3
= comp_self
._add
_output
_port
('insert')
98 self
.assertEqual(port3
.addr
, comp_self
._output
_ports
['insert'].addr
)
99 self
.assertEqual(port2
.addr
, comp_self
._output
_ports
['print'].addr
)
100 self
.assertEqual(port1
.addr
, comp_self
._output
_ports
['clear'].addr
)
102 self
._create
_comp
(MySource
)
104 def test_user_flt_output_ports_getitem(self
):
105 class MyIter(bt2
._UserMessageIterator
):
109 class MyFilter(bt2
._UserFilterComponent
,
110 message_iterator_class
=MyIter
):
111 def __init__(comp_self
, params
):
112 port1
= comp_self
._add
_output
_port
('clear')
113 port2
= comp_self
._add
_output
_port
('print')
114 port3
= comp_self
._add
_output
_port
('insert')
115 self
.assertEqual(port3
.addr
, comp_self
._output
_ports
['insert'].addr
)
116 self
.assertEqual(port2
.addr
, comp_self
._output
_ports
['print'].addr
)
117 self
.assertEqual(port1
.addr
, comp_self
._output
_ports
['clear'].addr
)
119 self
._create
_comp
(MyFilter
)
121 def test_user_flt_input_ports_getitem(self
):
122 class MyIter(bt2
._UserMessageIterator
):
126 class MyFilter(bt2
._UserFilterComponent
,
127 message_iterator_class
=MyIter
):
128 def __init__(comp_self
, params
):
129 port1
= comp_self
._add
_input
_port
('clear')
130 port2
= comp_self
._add
_input
_port
('print')
131 port3
= comp_self
._add
_input
_port
('insert')
132 self
.assertEqual(port3
.addr
, comp_self
._input
_ports
['insert'].addr
)
133 self
.assertEqual(port2
.addr
, comp_self
._input
_ports
['print'].addr
)
134 self
.assertEqual(port1
.addr
, comp_self
._input
_ports
['clear'].addr
)
136 self
._create
_comp
(MyFilter
)
138 def test_user_sink_input_ports_getitem(self
):
139 class MySink(bt2
._UserSinkComponent
):
140 def __init__(comp_self
, params
):
141 port1
= comp_self
._add
_input
_port
('clear')
142 port2
= comp_self
._add
_input
_port
('print')
143 port3
= comp_self
._add
_input
_port
('insert')
144 self
.assertEqual(port3
.addr
, comp_self
._input
_ports
['insert'].addr
)
145 self
.assertEqual(port2
.addr
, comp_self
._input
_ports
['print'].addr
)
146 self
.assertEqual(port1
.addr
, comp_self
._input
_ports
['clear'].addr
)
151 self
._create
_comp
(MySink
)
153 def test_user_src_output_ports_getitem_invalid_key(self
):
154 class MyIter(bt2
._UserMessageIterator
):
158 class MySource(bt2
._UserSourceComponent
,
159 message_iterator_class
=MyIter
):
160 def __init__(comp_self
, params
):
161 comp_self
._add
_output
_port
('clear')
162 comp_self
._add
_output
_port
('print')
163 comp_self
._add
_output
_port
('insert')
165 with self
.assertRaises(KeyError):
166 comp_self
._output
_ports
['hello']
168 self
._create
_comp
(MySource
)
170 def test_user_flt_output_ports_getitem_invalid_key(self
):
171 class MyIter(bt2
._UserMessageIterator
):
175 class MyFilter(bt2
._UserFilterComponent
,
176 message_iterator_class
=MyIter
):
177 def __init__(comp_self
, params
):
178 comp_self
._add
_output
_port
('clear')
179 comp_self
._add
_output
_port
('print')
180 comp_self
._add
_output
_port
('insert')
182 with self
.assertRaises(KeyError):
183 comp_self
._output
_ports
['hello']
185 self
._create
_comp
(MyFilter
)
187 def test_user_flt_input_ports_getitem_invalid_key(self
):
188 class MyIter(bt2
._UserMessageIterator
):
192 class MyFilter(bt2
._UserFilterComponent
,
193 message_iterator_class
=MyIter
):
194 def __init__(comp_self
, params
):
195 comp_self
._add
_input
_port
('clear')
196 comp_self
._add
_input
_port
('print')
197 comp_self
._add
_input
_port
('insert')
199 with self
.assertRaises(KeyError):
200 comp_self
._input
_ports
['hello']
202 self
._create
_comp
(MyFilter
)
204 def test_user_sink_input_ports_getitem_invalid_key(self
):
205 class MySink(bt2
._UserSinkComponent
):
206 def __init__(comp_self
, params
):
207 comp_self
._add
_input
_port
('clear')
208 comp_self
._add
_input
_port
('print')
209 comp_self
._add
_input
_port
('insert')
211 with self
.assertRaises(KeyError):
212 comp_self
._input
_ports
['hello']
217 self
._create
_comp
(MySink
)
219 def test_user_src_output_ports_len(self
):
220 class MyIter(bt2
._UserMessageIterator
):
224 class MySource(bt2
._UserSourceComponent
,
225 message_iterator_class
=MyIter
):
226 def __init__(comp_self
, params
):
227 comp_self
._add
_output
_port
('clear')
228 comp_self
._add
_output
_port
('print')
229 comp_self
._add
_output
_port
('insert')
230 self
.assertEqual(len(comp_self
._output
_ports
), 3)
232 self
._create
_comp
(MySource
)
234 def test_user_flt_output_ports_len(self
):
235 class MyIter(bt2
._UserMessageIterator
):
239 class MyFilter(bt2
._UserFilterComponent
,
240 message_iterator_class
=MyIter
):
241 def __init__(comp_self
, params
):
242 comp_self
._add
_output
_port
('clear')
243 comp_self
._add
_output
_port
('print')
244 comp_self
._add
_output
_port
('insert')
245 self
.assertEqual(len(comp_self
._output
_ports
), 3)
247 self
._create
_comp
(MyFilter
)
249 def test_user_flt_input_ports_len(self
):
250 class MyIter(bt2
._UserMessageIterator
):
254 class MyFilter(bt2
._UserFilterComponent
,
255 message_iterator_class
=MyIter
):
256 def __init__(comp_self
, params
):
257 comp_self
._add
_input
_port
('clear')
258 comp_self
._add
_input
_port
('print')
259 comp_self
._add
_input
_port
('insert')
260 self
.assertEqual(len(comp_self
._input
_ports
), 3)
262 self
._create
_comp
(MyFilter
)
264 def test_user_sink_input_ports_len(self
):
265 class MySink(bt2
._UserSinkComponent
):
266 def __init__(comp_self
, params
):
267 comp_self
._add
_input
_port
('clear')
268 comp_self
._add
_input
_port
('print')
269 comp_self
._add
_input
_port
('insert')
270 self
.assertEqual(len(comp_self
._input
_ports
), 3)
275 self
._create
_comp
(MySink
)
277 def test_user_src_output_ports_iter(self
):
278 class MyIter(bt2
._UserMessageIterator
):
282 class MySource(bt2
._UserSourceComponent
,
283 message_iterator_class
=MyIter
):
284 def __init__(comp_self
, params
):
285 port1
= comp_self
._add
_output
_port
('clear')
286 port2
= comp_self
._add
_output
_port
('print')
287 port3
= comp_self
._add
_output
_port
('insert')
290 for port_name
, port
in comp_self
._output
_ports
.items():
291 ports
.append((port_name
, port
))
293 self
.assertEqual(ports
[0][0], 'clear')
294 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
295 self
.assertEqual(ports
[1][0], 'print')
296 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
297 self
.assertEqual(ports
[2][0], 'insert')
298 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
300 self
._create
_comp
(MySource
)
302 def test_user_flt_output_ports_iter(self
):
303 class MyIter(bt2
._UserMessageIterator
):
307 class MyFilter(bt2
._UserFilterComponent
,
308 message_iterator_class
=MyIter
):
309 def __init__(comp_self
, params
):
310 port1
= comp_self
._add
_output
_port
('clear')
311 port2
= comp_self
._add
_output
_port
('print')
312 port3
= comp_self
._add
_output
_port
('insert')
315 for port_name
, port
in comp_self
._output
_ports
.items():
316 ports
.append((port_name
, port
))
318 self
.assertEqual(ports
[0][0], 'clear')
319 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
320 self
.assertEqual(ports
[1][0], 'print')
321 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
322 self
.assertEqual(ports
[2][0], 'insert')
323 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
325 self
._create
_comp
(MyFilter
)
327 def test_user_flt_input_ports_iter(self
):
328 class MyIter(bt2
._UserMessageIterator
):
332 class MyFilter(bt2
._UserFilterComponent
,
333 message_iterator_class
=MyIter
):
334 def __init__(comp_self
, params
):
335 port1
= comp_self
._add
_input
_port
('clear')
336 port2
= comp_self
._add
_input
_port
('print')
337 port3
= comp_self
._add
_input
_port
('insert')
340 for port_name
, port
in comp_self
._input
_ports
.items():
341 ports
.append((port_name
, port
))
343 self
.assertEqual(ports
[0][0], 'clear')
344 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
345 self
.assertEqual(ports
[1][0], 'print')
346 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
347 self
.assertEqual(ports
[2][0], 'insert')
348 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
350 self
._create
_comp
(MyFilter
)
352 def test_user_sink_input_ports_iter(self
):
353 class MySink(bt2
._UserSinkComponent
):
354 def __init__(comp_self
, params
):
355 port1
= comp_self
._add
_input
_port
('clear')
356 port2
= comp_self
._add
_input
_port
('print')
357 port3
= comp_self
._add
_input
_port
('insert')
360 for port_name
, port
in comp_self
._input
_ports
.items():
361 ports
.append((port_name
, port
))
363 self
.assertEqual(ports
[0][0], 'clear')
364 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
365 self
.assertEqual(ports
[1][0], 'print')
366 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
367 self
.assertEqual(ports
[2][0], 'insert')
368 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
373 self
._create
_comp
(MySink
)
375 def test_gen_src_output_ports_getitem(self
):
376 class MyIter(bt2
._UserMessageIterator
):
384 class MySource(bt2
._UserSourceComponent
,
385 message_iterator_class
=MyIter
):
386 def __init__(comp_self
, params
):
387 nonlocal port1
, port2
, port3
388 port1
= comp_self
._add
_output
_port
('clear')
389 port2
= comp_self
._add
_output
_port
('print')
390 port3
= comp_self
._add
_output
_port
('insert')
392 comp
= self
._create
_comp
(MySource
)
393 self
.assertEqual(port3
.addr
, comp
.output_ports
['insert'].addr
)
394 self
.assertEqual(port2
.addr
, comp
.output_ports
['print'].addr
)
395 self
.assertEqual(port1
.addr
, comp
.output_ports
['clear'].addr
)
400 def test_gen_flt_output_ports_getitem(self
):
401 class MyIter(bt2
._UserMessageIterator
):
409 class MyFilter(bt2
._UserFilterComponent
,
410 message_iterator_class
=MyIter
):
411 def __init__(comp_self
, params
):
412 nonlocal port1
, port2
, port3
413 port1
= comp_self
._add
_output
_port
('clear')
414 port2
= comp_self
._add
_output
_port
('print')
415 port3
= comp_self
._add
_output
_port
('insert')
417 comp
= self
._create
_comp
(MyFilter
)
418 self
.assertEqual(port3
.addr
, comp
.output_ports
['insert'].addr
)
419 self
.assertEqual(port2
.addr
, comp
.output_ports
['print'].addr
)
420 self
.assertEqual(port1
.addr
, comp
.output_ports
['clear'].addr
)
425 def test_gen_flt_input_ports_getitem(self
):
426 class MyIter(bt2
._UserMessageIterator
):
434 class MyFilter(bt2
._UserFilterComponent
,
435 message_iterator_class
=MyIter
):
436 def __init__(comp_self
, params
):
437 nonlocal port1
, port2
, port3
438 port1
= comp_self
._add
_input
_port
('clear')
439 port2
= comp_self
._add
_input
_port
('print')
440 port3
= comp_self
._add
_input
_port
('insert')
442 comp
= self
._create
_comp
(MyFilter
)
443 self
.assertEqual(port3
.addr
, comp
.input_ports
['insert'].addr
)
444 self
.assertEqual(port2
.addr
, comp
.input_ports
['print'].addr
)
445 self
.assertEqual(port1
.addr
, comp
.input_ports
['clear'].addr
)
450 def test_gen_sink_input_ports_getitem(self
):
455 class MySink(bt2
._UserSinkComponent
):
456 def __init__(comp_self
, params
):
457 nonlocal port1
, port2
, port3
458 port1
= comp_self
._add
_input
_port
('clear')
459 port2
= comp_self
._add
_input
_port
('print')
460 port3
= comp_self
._add
_input
_port
('insert')
465 comp
= self
._create
_comp
(MySink
)
466 self
.assertEqual(port3
.addr
, comp
.input_ports
['insert'].addr
)
467 self
.assertEqual(port2
.addr
, comp
.input_ports
['print'].addr
)
468 self
.assertEqual(port1
.addr
, comp
.input_ports
['clear'].addr
)
473 def test_gen_src_output_ports_getitem_invalid_key(self
):
474 class MyIter(bt2
._UserMessageIterator
):
478 class MySource(bt2
._UserSourceComponent
,
479 message_iterator_class
=MyIter
):
480 def __init__(comp_self
, params
):
481 comp_self
._add
_output
_port
('clear')
482 comp_self
._add
_output
_port
('print')
483 comp_self
._add
_output
_port
('insert')
485 comp
= self
._create
_comp
(MySource
)
487 with self
.assertRaises(KeyError):
488 comp
.output_ports
['hello']
490 def test_gen_flt_output_ports_getitem_invalid_key(self
):
491 class MyIter(bt2
._UserMessageIterator
):
495 class MyFilter(bt2
._UserFilterComponent
,
496 message_iterator_class
=MyIter
):
497 def __init__(comp_self
, params
):
498 comp_self
._add
_output
_port
('clear')
499 comp_self
._add
_output
_port
('print')
500 comp_self
._add
_output
_port
('insert')
502 comp
= self
._create
_comp
(MyFilter
)
504 with self
.assertRaises(KeyError):
505 comp
.output_ports
['hello']
507 def test_gen_flt_input_ports_getitem_invalid_key(self
):
508 class MyIter(bt2
._UserMessageIterator
):
512 class MyFilter(bt2
._UserFilterComponent
,
513 message_iterator_class
=MyIter
):
514 def __init__(comp_self
, params
):
515 comp_self
._add
_input
_port
('clear')
516 comp_self
._add
_input
_port
('print')
517 comp_self
._add
_input
_port
('insert')
519 comp
= self
._create
_comp
(MyFilter
)
521 with self
.assertRaises(KeyError):
522 comp
.input_ports
['hello']
524 def test_gen_sink_input_ports_getitem_invalid_key(self
):
525 class MySink(bt2
._UserSinkComponent
):
526 def __init__(comp_self
, params
):
527 comp_self
._add
_input
_port
('clear')
528 comp_self
._add
_input
_port
('print')
529 comp_self
._add
_input
_port
('insert')
531 with self
.assertRaises(KeyError):
532 comp_self
._input
_ports
['hello']
537 comp
= self
._create
_comp
(MySink
)
539 with self
.assertRaises(KeyError):
540 comp
.input_ports
['hello']
542 def test_gen_src_output_ports_len(self
):
543 class MyIter(bt2
._UserMessageIterator
):
547 class MySource(bt2
._UserSourceComponent
,
548 message_iterator_class
=MyIter
):
549 def __init__(comp_self
, params
):
550 comp_self
._add
_output
_port
('clear')
551 comp_self
._add
_output
_port
('print')
552 comp_self
._add
_output
_port
('insert')
554 comp
= self
._create
_comp
(MySource
)
555 self
.assertEqual(len(comp
.output_ports
), 3)
557 def test_gen_flt_output_ports_len(self
):
558 class MyIter(bt2
._UserMessageIterator
):
562 class MyFilter(bt2
._UserFilterComponent
,
563 message_iterator_class
=MyIter
):
564 def __init__(comp_self
, params
):
565 comp_self
._add
_output
_port
('clear')
566 comp_self
._add
_output
_port
('print')
567 comp_self
._add
_output
_port
('insert')
569 comp
= self
._create
_comp
(MyFilter
)
570 self
.assertEqual(len(comp
.output_ports
), 3)
572 def test_gen_flt_input_ports_len(self
):
573 class MyIter(bt2
._UserMessageIterator
):
577 class MyFilter(bt2
._UserFilterComponent
,
578 message_iterator_class
=MyIter
):
579 def __init__(comp_self
, params
):
580 comp_self
._add
_input
_port
('clear')
581 comp_self
._add
_input
_port
('print')
582 comp_self
._add
_input
_port
('insert')
584 comp
= self
._create
_comp
(MyFilter
)
585 self
.assertEqual(len(comp
.input_ports
), 3)
587 def test_gen_sink_input_ports_len(self
):
588 class MySink(bt2
._UserSinkComponent
):
589 def __init__(comp_self
, params
):
590 comp_self
._add
_input
_port
('clear')
591 comp_self
._add
_input
_port
('print')
592 comp_self
._add
_input
_port
('insert')
597 comp
= self
._create
_comp
(MySink
)
598 self
.assertEqual(len(comp
.input_ports
), 3)
600 def test_gen_src_output_ports_iter(self
):
601 class MyIter(bt2
._UserMessageIterator
):
609 class MySource(bt2
._UserSourceComponent
,
610 message_iterator_class
=MyIter
):
611 def __init__(comp_self
, params
):
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
):
634 class MyIter(bt2
._UserMessageIterator
):
642 class MyFilter(bt2
._UserFilterComponent
,
643 message_iterator_class
=MyIter
):
644 def __init__(comp_self
, params
):
645 nonlocal port1
, port2
, port3
646 port1
= comp_self
._add
_output
_port
('clear')
647 port2
= comp_self
._add
_output
_port
('print')
648 port3
= comp_self
._add
_output
_port
('insert')
650 comp
= self
._create
_comp
(MyFilter
)
653 for port_name
, port
in comp
.output_ports
.items():
654 ports
.append((port_name
, port
))
656 self
.assertEqual(ports
[0][0], 'clear')
657 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
658 self
.assertEqual(ports
[1][0], 'print')
659 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
660 self
.assertEqual(ports
[2][0], 'insert')
661 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
666 def test_gen_flt_input_ports_iter(self
):
667 class MyIter(bt2
._UserMessageIterator
):
675 class MyFilter(bt2
._UserFilterComponent
,
676 message_iterator_class
=MyIter
):
677 def __init__(comp_self
, params
):
678 nonlocal port1
, port2
, port3
679 port1
= comp_self
._add
_input
_port
('clear')
680 port2
= comp_self
._add
_input
_port
('print')
681 port3
= comp_self
._add
_input
_port
('insert')
683 comp
= self
._create
_comp
(MyFilter
)
686 for port_name
, port
in comp
.input_ports
.items():
687 ports
.append((port_name
, port
))
689 self
.assertEqual(ports
[0][0], 'clear')
690 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
691 self
.assertEqual(ports
[1][0], 'print')
692 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
693 self
.assertEqual(ports
[2][0], 'insert')
694 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
699 def test_gen_sink_input_ports_iter(self
):
704 class MySink(bt2
._UserSinkComponent
):
705 def __init__(comp_self
, params
):
706 nonlocal port1
, port2
, port3
707 port1
= comp_self
._add
_input
_port
('clear')
708 port2
= comp_self
._add
_input
_port
('print')
709 port3
= comp_self
._add
_input
_port
('insert')
714 comp
= self
._create
_comp
(MySink
)
717 for port_name
, port
in comp
.input_ports
.items():
718 ports
.append((port_name
, port
))
720 self
.assertEqual(ports
[0][0], 'clear')
721 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
722 self
.assertEqual(ports
[1][0], 'print')
723 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
724 self
.assertEqual(ports
[2][0], 'insert')
725 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
731 class MySink(bt2
._UserSinkComponent
):
732 def __init__(comp_self
, params
):
733 comp_self
._add
_input
_port
('clear')
738 comp
= self
._create
_comp
(MySink
)
739 self
.assertEqual(comp
.input_ports
['clear'].name
, 'clear')
741 def test_connection_none(self
):
742 class MySink(bt2
._UserSinkComponent
):
743 def __init__(comp_self
, params
):
744 comp_self
._add
_input
_port
('clear')
749 comp
= self
._create
_comp
(MySink
)
750 self
.assertIsNone(comp
.input_ports
['clear'].connection
)
752 def test_is_connected_false(self
):
753 class MySink(bt2
._UserSinkComponent
):
754 def __init__(comp_self
, params
):
755 comp_self
._add
_input
_port
('clear')
760 comp
= self
._create
_comp
(MySink
)
761 self
.assertFalse(comp
.input_ports
['clear'].is_connected
)
763 def test_self_name(self
):
764 class MySink(bt2
._UserSinkComponent
):
765 def __init__(comp_self
, params
):
766 port
= comp_self
._add
_input
_port
('clear')
767 self
.assertEqual(port
.name
, 'clear')
772 self
._create
_comp
(MySink
)
774 def test_self_connection_none(self
):
775 class MySink(bt2
._UserSinkComponent
):
776 def __init__(comp_self
, params
):
777 port
= comp_self
._add
_input
_port
('clear')
778 self
.assertIsNone(port
.connection
)
783 self
._create
_comp
(MySink
)
785 def test_self_is_connected_false(self
):
786 class MySink(bt2
._UserSinkComponent
):
787 def __init__(comp_self
, params
):
788 port
= comp_self
._add
_input
_port
('clear')
789 self
.assertFalse(port
.is_connected
)
794 self
._create
_comp
(MySink
)
796 def test_source_self_port_user_data(self
):
797 class MyIter(bt2
._UserMessageIterator
):
801 class MySource(bt2
._UserFilterComponent
,
802 message_iterator_class
=MyIter
):
803 def __init__(comp_self
, params
):
806 p
= comp_self
._add
_output
_port
('port1')
807 user_datas
.append(p
.user_data
)
808 p
= comp_self
._add
_output
_port
('port2', 2)
809 user_datas
.append(p
.user_data
)
813 comp
= self
._create
_comp
(MySource
)
814 self
.assertEqual(user_datas
, [None, 2])
816 def test_filter_self_port_user_data(self
):
817 class MyIter(bt2
._UserMessageIterator
):
821 class MyFilter(bt2
._UserFilterComponent
,
822 message_iterator_class
=MyIter
):
823 def __init__(comp_self
, params
):
826 p
= comp_self
._add
_output
_port
('port1')
827 user_datas
.append(p
.user_data
)
828 p
= comp_self
._add
_output
_port
('port2', 'user data string')
829 user_datas
.append(p
.user_data
)
831 p
= comp_self
._add
_input
_port
('port3')
832 user_datas
.append(p
.user_data
)
833 p
= comp_self
._add
_input
_port
('port4', user_data
={'user data': 'dict'})
834 user_datas
.append(p
.user_data
)
838 comp
= self
._create
_comp
(MyFilter
)
839 self
.assertEqual(user_datas
,
840 [None, 'user data string', None, {'user data': 'dict'}])
842 def test_sink_self_port_user_data(self
):
843 class MyIter(bt2
._UserMessageIterator
):
847 class MySink(bt2
._UserFilterComponent
,
848 message_iterator_class
=MyIter
):
849 def __init__(comp_self
, params
):
852 p
= comp_self
._add
_input
_port
('port1')
853 user_datas
.append(p
.user_data
)
854 p
= comp_self
._add
_input
_port
('port2', set())
855 user_datas
.append(p
.user_data
)
859 comp
= self
._create
_comp
(MySink
)
860 self
.assertEqual(user_datas
, [None, set()])
This page took 0.048213 seconds and 4 git commands to generate.