8be3f38cb2868ce3d691568e2a8e008797bec5c7
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
, message_iterator_class
=MyIter
):
39 def __init__(comp_self
, params
):
40 port
= comp_self
._add
_output
_port
('out')
41 self
.assertEqual(port
.name
, 'out')
43 comp
= self
._create
_comp
(MySource
)
44 self
.assertEqual(len(comp
.output_ports
), 1)
46 def test_flt_add_output_port(self
):
47 class MyIter(bt2
._UserMessageIterator
):
51 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
52 def __init__(comp_self
, params
):
53 port
= comp_self
._add
_output
_port
('out')
54 self
.assertEqual(port
.name
, 'out')
56 comp
= self
._create
_comp
(MyFilter
)
57 self
.assertEqual(len(comp
.output_ports
), 1)
59 def test_flt_add_input_port(self
):
60 class MyIter(bt2
._UserMessageIterator
):
64 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
65 def __init__(comp_self
, params
):
66 port
= comp_self
._add
_input
_port
('in')
67 self
.assertEqual(port
.name
, 'in')
69 comp
= self
._create
_comp
(MyFilter
)
70 self
.assertEqual(len(comp
.input_ports
), 1)
72 def test_sink_add_input_port(self
):
73 class MySink(bt2
._UserSinkComponent
):
74 def __init__(comp_self
, params
):
75 port
= comp_self
._add
_input
_port
('in')
76 self
.assertEqual(port
.name
, 'in')
78 def _user_consume(self
):
81 comp
= self
._create
_comp
(MySink
)
82 self
.assertEqual(len(comp
.input_ports
), 1)
84 def test_user_src_output_ports_getitem(self
):
85 class MyIter(bt2
._UserMessageIterator
):
89 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
90 def __init__(comp_self
, params
):
91 port1
= comp_self
._add
_output
_port
('clear')
92 port2
= comp_self
._add
_output
_port
('print')
93 port3
= comp_self
._add
_output
_port
('insert')
94 self
.assertEqual(port3
.addr
, comp_self
._output
_ports
['insert'].addr
)
95 self
.assertEqual(port2
.addr
, comp_self
._output
_ports
['print'].addr
)
96 self
.assertEqual(port1
.addr
, comp_self
._output
_ports
['clear'].addr
)
98 self
._create
_comp
(MySource
)
100 def test_user_flt_output_ports_getitem(self
):
101 class MyIter(bt2
._UserMessageIterator
):
105 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
106 def __init__(comp_self
, params
):
107 port1
= comp_self
._add
_output
_port
('clear')
108 port2
= comp_self
._add
_output
_port
('print')
109 port3
= comp_self
._add
_output
_port
('insert')
110 self
.assertEqual(port3
.addr
, comp_self
._output
_ports
['insert'].addr
)
111 self
.assertEqual(port2
.addr
, comp_self
._output
_ports
['print'].addr
)
112 self
.assertEqual(port1
.addr
, comp_self
._output
_ports
['clear'].addr
)
114 self
._create
_comp
(MyFilter
)
116 def test_user_flt_input_ports_getitem(self
):
117 class MyIter(bt2
._UserMessageIterator
):
121 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
122 def __init__(comp_self
, params
):
123 port1
= comp_self
._add
_input
_port
('clear')
124 port2
= comp_self
._add
_input
_port
('print')
125 port3
= comp_self
._add
_input
_port
('insert')
126 self
.assertEqual(port3
.addr
, comp_self
._input
_ports
['insert'].addr
)
127 self
.assertEqual(port2
.addr
, comp_self
._input
_ports
['print'].addr
)
128 self
.assertEqual(port1
.addr
, comp_self
._input
_ports
['clear'].addr
)
130 self
._create
_comp
(MyFilter
)
132 def test_user_sink_input_ports_getitem(self
):
133 class MySink(bt2
._UserSinkComponent
):
134 def __init__(comp_self
, params
):
135 port1
= comp_self
._add
_input
_port
('clear')
136 port2
= comp_self
._add
_input
_port
('print')
137 port3
= comp_self
._add
_input
_port
('insert')
138 self
.assertEqual(port3
.addr
, comp_self
._input
_ports
['insert'].addr
)
139 self
.assertEqual(port2
.addr
, comp_self
._input
_ports
['print'].addr
)
140 self
.assertEqual(port1
.addr
, comp_self
._input
_ports
['clear'].addr
)
142 def _user_consume(self
):
145 self
._create
_comp
(MySink
)
147 def test_user_src_output_ports_getitem_invalid_key(self
):
148 class MyIter(bt2
._UserMessageIterator
):
152 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
153 def __init__(comp_self
, params
):
154 comp_self
._add
_output
_port
('clear')
155 comp_self
._add
_output
_port
('print')
156 comp_self
._add
_output
_port
('insert')
158 with self
.assertRaises(KeyError):
159 comp_self
._output
_ports
['hello']
161 self
._create
_comp
(MySource
)
163 def test_user_flt_output_ports_getitem_invalid_key(self
):
164 class MyIter(bt2
._UserMessageIterator
):
168 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
169 def __init__(comp_self
, params
):
170 comp_self
._add
_output
_port
('clear')
171 comp_self
._add
_output
_port
('print')
172 comp_self
._add
_output
_port
('insert')
174 with self
.assertRaises(KeyError):
175 comp_self
._output
_ports
['hello']
177 self
._create
_comp
(MyFilter
)
179 def test_user_flt_input_ports_getitem_invalid_key(self
):
180 class MyIter(bt2
._UserMessageIterator
):
184 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
185 def __init__(comp_self
, params
):
186 comp_self
._add
_input
_port
('clear')
187 comp_self
._add
_input
_port
('print')
188 comp_self
._add
_input
_port
('insert')
190 with self
.assertRaises(KeyError):
191 comp_self
._input
_ports
['hello']
193 self
._create
_comp
(MyFilter
)
195 def test_user_sink_input_ports_getitem_invalid_key(self
):
196 class MySink(bt2
._UserSinkComponent
):
197 def __init__(comp_self
, params
):
198 comp_self
._add
_input
_port
('clear')
199 comp_self
._add
_input
_port
('print')
200 comp_self
._add
_input
_port
('insert')
202 with self
.assertRaises(KeyError):
203 comp_self
._input
_ports
['hello']
205 def _user_consume(self
):
208 self
._create
_comp
(MySink
)
210 def test_user_src_output_ports_len(self
):
211 class MyIter(bt2
._UserMessageIterator
):
215 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
216 def __init__(comp_self
, params
):
217 comp_self
._add
_output
_port
('clear')
218 comp_self
._add
_output
_port
('print')
219 comp_self
._add
_output
_port
('insert')
220 self
.assertEqual(len(comp_self
._output
_ports
), 3)
222 self
._create
_comp
(MySource
)
224 def test_user_flt_output_ports_len(self
):
225 class MyIter(bt2
._UserMessageIterator
):
229 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
230 def __init__(comp_self
, params
):
231 comp_self
._add
_output
_port
('clear')
232 comp_self
._add
_output
_port
('print')
233 comp_self
._add
_output
_port
('insert')
234 self
.assertEqual(len(comp_self
._output
_ports
), 3)
236 self
._create
_comp
(MyFilter
)
238 def test_user_flt_input_ports_len(self
):
239 class MyIter(bt2
._UserMessageIterator
):
243 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
244 def __init__(comp_self
, params
):
245 comp_self
._add
_input
_port
('clear')
246 comp_self
._add
_input
_port
('print')
247 comp_self
._add
_input
_port
('insert')
248 self
.assertEqual(len(comp_self
._input
_ports
), 3)
250 self
._create
_comp
(MyFilter
)
252 def test_user_sink_input_ports_len(self
):
253 class MySink(bt2
._UserSinkComponent
):
254 def __init__(comp_self
, params
):
255 comp_self
._add
_input
_port
('clear')
256 comp_self
._add
_input
_port
('print')
257 comp_self
._add
_input
_port
('insert')
258 self
.assertEqual(len(comp_self
._input
_ports
), 3)
260 def _user_consume(self
):
263 self
._create
_comp
(MySink
)
265 def test_user_src_output_ports_iter(self
):
266 class MyIter(bt2
._UserMessageIterator
):
270 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
271 def __init__(comp_self
, params
):
272 port1
= comp_self
._add
_output
_port
('clear')
273 port2
= comp_self
._add
_output
_port
('print')
274 port3
= comp_self
._add
_output
_port
('insert')
277 for port_name
, port
in comp_self
._output
_ports
.items():
278 ports
.append((port_name
, port
))
280 self
.assertEqual(ports
[0][0], 'clear')
281 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
282 self
.assertEqual(ports
[1][0], 'print')
283 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
284 self
.assertEqual(ports
[2][0], 'insert')
285 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
287 self
._create
_comp
(MySource
)
289 def test_user_flt_output_ports_iter(self
):
290 class MyIter(bt2
._UserMessageIterator
):
294 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
295 def __init__(comp_self
, params
):
296 port1
= comp_self
._add
_output
_port
('clear')
297 port2
= comp_self
._add
_output
_port
('print')
298 port3
= comp_self
._add
_output
_port
('insert')
301 for port_name
, port
in comp_self
._output
_ports
.items():
302 ports
.append((port_name
, port
))
304 self
.assertEqual(ports
[0][0], 'clear')
305 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
306 self
.assertEqual(ports
[1][0], 'print')
307 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
308 self
.assertEqual(ports
[2][0], 'insert')
309 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
311 self
._create
_comp
(MyFilter
)
313 def test_user_flt_input_ports_iter(self
):
314 class MyIter(bt2
._UserMessageIterator
):
318 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
319 def __init__(comp_self
, params
):
320 port1
= comp_self
._add
_input
_port
('clear')
321 port2
= comp_self
._add
_input
_port
('print')
322 port3
= comp_self
._add
_input
_port
('insert')
325 for port_name
, port
in comp_self
._input
_ports
.items():
326 ports
.append((port_name
, port
))
328 self
.assertEqual(ports
[0][0], 'clear')
329 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
330 self
.assertEqual(ports
[1][0], 'print')
331 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
332 self
.assertEqual(ports
[2][0], 'insert')
333 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
335 self
._create
_comp
(MyFilter
)
337 def test_user_sink_input_ports_iter(self
):
338 class MySink(bt2
._UserSinkComponent
):
339 def __init__(comp_self
, params
):
340 port1
= comp_self
._add
_input
_port
('clear')
341 port2
= comp_self
._add
_input
_port
('print')
342 port3
= comp_self
._add
_input
_port
('insert')
345 for port_name
, port
in comp_self
._input
_ports
.items():
346 ports
.append((port_name
, port
))
348 self
.assertEqual(ports
[0][0], 'clear')
349 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
350 self
.assertEqual(ports
[1][0], 'print')
351 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
352 self
.assertEqual(ports
[2][0], 'insert')
353 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
355 def _user_consume(self
):
358 self
._create
_comp
(MySink
)
360 def test_gen_src_output_ports_getitem(self
):
361 class MyIter(bt2
._UserMessageIterator
):
369 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
370 def __init__(comp_self
, params
):
371 nonlocal port1
, port2
, port3
372 port1
= comp_self
._add
_output
_port
('clear')
373 port2
= comp_self
._add
_output
_port
('print')
374 port3
= comp_self
._add
_output
_port
('insert')
376 comp
= self
._create
_comp
(MySource
)
377 self
.assertEqual(port3
.addr
, comp
.output_ports
['insert'].addr
)
378 self
.assertEqual(port2
.addr
, comp
.output_ports
['print'].addr
)
379 self
.assertEqual(port1
.addr
, comp
.output_ports
['clear'].addr
)
384 def test_gen_flt_output_ports_getitem(self
):
385 class MyIter(bt2
._UserMessageIterator
):
393 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
394 def __init__(comp_self
, params
):
395 nonlocal port1
, port2
, port3
396 port1
= comp_self
._add
_output
_port
('clear')
397 port2
= comp_self
._add
_output
_port
('print')
398 port3
= comp_self
._add
_output
_port
('insert')
400 comp
= self
._create
_comp
(MyFilter
)
401 self
.assertEqual(port3
.addr
, comp
.output_ports
['insert'].addr
)
402 self
.assertEqual(port2
.addr
, comp
.output_ports
['print'].addr
)
403 self
.assertEqual(port1
.addr
, comp
.output_ports
['clear'].addr
)
408 def test_gen_flt_input_ports_getitem(self
):
409 class MyIter(bt2
._UserMessageIterator
):
417 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
418 def __init__(comp_self
, params
):
419 nonlocal port1
, port2
, port3
420 port1
= comp_self
._add
_input
_port
('clear')
421 port2
= comp_self
._add
_input
_port
('print')
422 port3
= comp_self
._add
_input
_port
('insert')
424 comp
= self
._create
_comp
(MyFilter
)
425 self
.assertEqual(port3
.addr
, comp
.input_ports
['insert'].addr
)
426 self
.assertEqual(port2
.addr
, comp
.input_ports
['print'].addr
)
427 self
.assertEqual(port1
.addr
, comp
.input_ports
['clear'].addr
)
432 def test_gen_sink_input_ports_getitem(self
):
437 class MySink(bt2
._UserSinkComponent
):
438 def __init__(comp_self
, params
):
439 nonlocal port1
, port2
, port3
440 port1
= comp_self
._add
_input
_port
('clear')
441 port2
= comp_self
._add
_input
_port
('print')
442 port3
= comp_self
._add
_input
_port
('insert')
444 def _user_consume(self
):
447 comp
= self
._create
_comp
(MySink
)
448 self
.assertEqual(port3
.addr
, comp
.input_ports
['insert'].addr
)
449 self
.assertEqual(port2
.addr
, comp
.input_ports
['print'].addr
)
450 self
.assertEqual(port1
.addr
, comp
.input_ports
['clear'].addr
)
455 def test_gen_src_output_ports_getitem_invalid_key(self
):
456 class MyIter(bt2
._UserMessageIterator
):
460 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
461 def __init__(comp_self
, params
):
462 comp_self
._add
_output
_port
('clear')
463 comp_self
._add
_output
_port
('print')
464 comp_self
._add
_output
_port
('insert')
466 comp
= self
._create
_comp
(MySource
)
468 with self
.assertRaises(KeyError):
469 comp
.output_ports
['hello']
471 def test_gen_flt_output_ports_getitem_invalid_key(self
):
472 class MyIter(bt2
._UserMessageIterator
):
476 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
477 def __init__(comp_self
, params
):
478 comp_self
._add
_output
_port
('clear')
479 comp_self
._add
_output
_port
('print')
480 comp_self
._add
_output
_port
('insert')
482 comp
= self
._create
_comp
(MyFilter
)
484 with self
.assertRaises(KeyError):
485 comp
.output_ports
['hello']
487 def test_gen_flt_input_ports_getitem_invalid_key(self
):
488 class MyIter(bt2
._UserMessageIterator
):
492 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
493 def __init__(comp_self
, params
):
494 comp_self
._add
_input
_port
('clear')
495 comp_self
._add
_input
_port
('print')
496 comp_self
._add
_input
_port
('insert')
498 comp
= self
._create
_comp
(MyFilter
)
500 with self
.assertRaises(KeyError):
501 comp
.input_ports
['hello']
503 def test_gen_sink_input_ports_getitem_invalid_key(self
):
504 class MySink(bt2
._UserSinkComponent
):
505 def __init__(comp_self
, params
):
506 comp_self
._add
_input
_port
('clear')
507 comp_self
._add
_input
_port
('print')
508 comp_self
._add
_input
_port
('insert')
510 with self
.assertRaises(KeyError):
511 comp_self
._input
_ports
['hello']
513 def _user_consume(self
):
516 comp
= self
._create
_comp
(MySink
)
518 with self
.assertRaises(KeyError):
519 comp
.input_ports
['hello']
521 def test_gen_src_output_ports_len(self
):
522 class MyIter(bt2
._UserMessageIterator
):
526 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
527 def __init__(comp_self
, params
):
528 comp_self
._add
_output
_port
('clear')
529 comp_self
._add
_output
_port
('print')
530 comp_self
._add
_output
_port
('insert')
532 comp
= self
._create
_comp
(MySource
)
533 self
.assertEqual(len(comp
.output_ports
), 3)
535 def test_gen_flt_output_ports_len(self
):
536 class MyIter(bt2
._UserMessageIterator
):
540 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
541 def __init__(comp_self
, params
):
542 comp_self
._add
_output
_port
('clear')
543 comp_self
._add
_output
_port
('print')
544 comp_self
._add
_output
_port
('insert')
546 comp
= self
._create
_comp
(MyFilter
)
547 self
.assertEqual(len(comp
.output_ports
), 3)
549 def test_gen_flt_input_ports_len(self
):
550 class MyIter(bt2
._UserMessageIterator
):
554 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
555 def __init__(comp_self
, params
):
556 comp_self
._add
_input
_port
('clear')
557 comp_self
._add
_input
_port
('print')
558 comp_self
._add
_input
_port
('insert')
560 comp
= self
._create
_comp
(MyFilter
)
561 self
.assertEqual(len(comp
.input_ports
), 3)
563 def test_gen_sink_input_ports_len(self
):
564 class MySink(bt2
._UserSinkComponent
):
565 def __init__(comp_self
, params
):
566 comp_self
._add
_input
_port
('clear')
567 comp_self
._add
_input
_port
('print')
568 comp_self
._add
_input
_port
('insert')
570 def _user_consume(self
):
573 comp
= self
._create
_comp
(MySink
)
574 self
.assertEqual(len(comp
.input_ports
), 3)
576 def test_gen_src_output_ports_iter(self
):
577 class MyIter(bt2
._UserMessageIterator
):
585 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
586 def __init__(comp_self
, params
):
587 nonlocal port1
, port2
, port3
588 port1
= comp_self
._add
_output
_port
('clear')
589 port2
= comp_self
._add
_output
_port
('print')
590 port3
= comp_self
._add
_output
_port
('insert')
592 comp
= self
._create
_comp
(MySource
)
595 for port_name
, port
in comp
.output_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_flt_output_ports_iter(self
):
609 class MyIter(bt2
._UserMessageIterator
):
617 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
618 def __init__(comp_self
, params
):
619 nonlocal port1
, port2
, port3
620 port1
= comp_self
._add
_output
_port
('clear')
621 port2
= comp_self
._add
_output
_port
('print')
622 port3
= comp_self
._add
_output
_port
('insert')
624 comp
= self
._create
_comp
(MyFilter
)
627 for port_name
, port
in comp
.output_ports
.items():
628 ports
.append((port_name
, port
))
630 self
.assertEqual(ports
[0][0], 'clear')
631 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
632 self
.assertEqual(ports
[1][0], 'print')
633 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
634 self
.assertEqual(ports
[2][0], 'insert')
635 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
640 def test_gen_flt_input_ports_iter(self
):
641 class MyIter(bt2
._UserMessageIterator
):
649 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
650 def __init__(comp_self
, params
):
651 nonlocal port1
, port2
, port3
652 port1
= comp_self
._add
_input
_port
('clear')
653 port2
= comp_self
._add
_input
_port
('print')
654 port3
= comp_self
._add
_input
_port
('insert')
656 comp
= self
._create
_comp
(MyFilter
)
659 for port_name
, port
in comp
.input_ports
.items():
660 ports
.append((port_name
, port
))
662 self
.assertEqual(ports
[0][0], 'clear')
663 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
664 self
.assertEqual(ports
[1][0], 'print')
665 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
666 self
.assertEqual(ports
[2][0], 'insert')
667 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
672 def test_gen_sink_input_ports_iter(self
):
677 class MySink(bt2
._UserSinkComponent
):
678 def __init__(comp_self
, params
):
679 nonlocal port1
, port2
, port3
680 port1
= comp_self
._add
_input
_port
('clear')
681 port2
= comp_self
._add
_input
_port
('print')
682 port3
= comp_self
._add
_input
_port
('insert')
684 def _user_consume(self
):
687 comp
= self
._create
_comp
(MySink
)
690 for port_name
, port
in comp
.input_ports
.items():
691 ports
.append((port_name
, port
))
693 self
.assertEqual(ports
[0][0], 'clear')
694 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
695 self
.assertEqual(ports
[1][0], 'print')
696 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
697 self
.assertEqual(ports
[2][0], 'insert')
698 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
704 class MySink(bt2
._UserSinkComponent
):
705 def __init__(comp_self
, params
):
706 comp_self
._add
_input
_port
('clear')
708 def _user_consume(self
):
711 comp
= self
._create
_comp
(MySink
)
712 self
.assertEqual(comp
.input_ports
['clear'].name
, 'clear')
714 def test_connection_none(self
):
715 class MySink(bt2
._UserSinkComponent
):
716 def __init__(comp_self
, params
):
717 comp_self
._add
_input
_port
('clear')
719 def _user_consume(self
):
722 comp
= self
._create
_comp
(MySink
)
723 self
.assertIsNone(comp
.input_ports
['clear'].connection
)
725 def test_is_connected_false(self
):
726 class MySink(bt2
._UserSinkComponent
):
727 def __init__(comp_self
, params
):
728 comp_self
._add
_input
_port
('clear')
730 def _user_consume(self
):
733 comp
= self
._create
_comp
(MySink
)
734 self
.assertFalse(comp
.input_ports
['clear'].is_connected
)
736 def test_self_name(self
):
737 class MySink(bt2
._UserSinkComponent
):
738 def __init__(comp_self
, params
):
739 port
= comp_self
._add
_input
_port
('clear')
740 self
.assertEqual(port
.name
, 'clear')
742 def _user_consume(self
):
745 self
._create
_comp
(MySink
)
747 def test_self_connection_none(self
):
748 class MySink(bt2
._UserSinkComponent
):
749 def __init__(comp_self
, params
):
750 port
= comp_self
._add
_input
_port
('clear')
751 self
.assertIsNone(port
.connection
)
753 def _user_consume(self
):
756 self
._create
_comp
(MySink
)
758 def test_self_is_connected_false(self
):
759 class MySink(bt2
._UserSinkComponent
):
760 def __init__(comp_self
, params
):
761 port
= comp_self
._add
_input
_port
('clear')
762 self
.assertFalse(port
.is_connected
)
764 def _user_consume(self
):
767 self
._create
_comp
(MySink
)
769 def test_source_self_port_user_data(self
):
770 class MyIter(bt2
._UserMessageIterator
):
774 class MySource(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
775 def __init__(comp_self
, params
):
778 p
= comp_self
._add
_output
_port
('port1')
779 user_datas
.append(p
.user_data
)
780 p
= comp_self
._add
_output
_port
('port2', 2)
781 user_datas
.append(p
.user_data
)
785 comp
= self
._create
_comp
(MySource
)
786 self
.assertEqual(user_datas
, [None, 2])
788 def test_filter_self_port_user_data(self
):
789 class MyIter(bt2
._UserMessageIterator
):
793 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
794 def __init__(comp_self
, params
):
797 p
= comp_self
._add
_output
_port
('port1')
798 user_datas
.append(p
.user_data
)
799 p
= comp_self
._add
_output
_port
('port2', 'user data string')
800 user_datas
.append(p
.user_data
)
802 p
= comp_self
._add
_input
_port
('port3')
803 user_datas
.append(p
.user_data
)
804 p
= comp_self
._add
_input
_port
('port4', user_data
={'user data': 'dict'})
805 user_datas
.append(p
.user_data
)
809 comp
= self
._create
_comp
(MyFilter
)
811 user_datas
, [None, 'user data string', None, {'user data': 'dict'}]
814 def test_sink_self_port_user_data(self
):
815 class MyIter(bt2
._UserMessageIterator
):
819 class MySink(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
820 def __init__(comp_self
, params
):
823 p
= comp_self
._add
_input
_port
('port1')
824 user_datas
.append(p
.user_data
)
825 p
= comp_self
._add
_input
_port
('port2', set())
826 user_datas
.append(p
.user_data
)
830 comp
= self
._create
_comp
(MySink
)
831 self
.assertEqual(user_datas
, [None, set()])
This page took 0.047618 seconds and 3 git commands to generate.