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')
81 def _graph_is_configured(self
):
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
, message_iterator_class
=MyIter
):
93 def __init__(comp_self
, params
):
94 port1
= comp_self
._add
_output
_port
('clear')
95 port2
= comp_self
._add
_output
_port
('print')
96 port3
= comp_self
._add
_output
_port
('insert')
97 self
.assertEqual(port3
.addr
, comp_self
._output
_ports
['insert'].addr
)
98 self
.assertEqual(port2
.addr
, comp_self
._output
_ports
['print'].addr
)
99 self
.assertEqual(port1
.addr
, comp_self
._output
_ports
['clear'].addr
)
101 self
._create
_comp
(MySource
)
103 def test_user_flt_output_ports_getitem(self
):
104 class MyIter(bt2
._UserMessageIterator
):
108 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
109 def __init__(comp_self
, params
):
110 port1
= comp_self
._add
_output
_port
('clear')
111 port2
= comp_self
._add
_output
_port
('print')
112 port3
= comp_self
._add
_output
_port
('insert')
113 self
.assertEqual(port3
.addr
, comp_self
._output
_ports
['insert'].addr
)
114 self
.assertEqual(port2
.addr
, comp_self
._output
_ports
['print'].addr
)
115 self
.assertEqual(port1
.addr
, comp_self
._output
_ports
['clear'].addr
)
117 self
._create
_comp
(MyFilter
)
119 def test_user_flt_input_ports_getitem(self
):
120 class MyIter(bt2
._UserMessageIterator
):
124 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
125 def __init__(comp_self
, params
):
126 port1
= comp_self
._add
_input
_port
('clear')
127 port2
= comp_self
._add
_input
_port
('print')
128 port3
= comp_self
._add
_input
_port
('insert')
129 self
.assertEqual(port3
.addr
, comp_self
._input
_ports
['insert'].addr
)
130 self
.assertEqual(port2
.addr
, comp_self
._input
_ports
['print'].addr
)
131 self
.assertEqual(port1
.addr
, comp_self
._input
_ports
['clear'].addr
)
133 self
._create
_comp
(MyFilter
)
135 def test_user_sink_input_ports_getitem(self
):
136 class MySink(bt2
._UserSinkComponent
):
137 def __init__(comp_self
, params
):
138 port1
= comp_self
._add
_input
_port
('clear')
139 port2
= comp_self
._add
_input
_port
('print')
140 port3
= comp_self
._add
_input
_port
('insert')
141 self
.assertEqual(port3
.addr
, comp_self
._input
_ports
['insert'].addr
)
142 self
.assertEqual(port2
.addr
, comp_self
._input
_ports
['print'].addr
)
143 self
.assertEqual(port1
.addr
, comp_self
._input
_ports
['clear'].addr
)
148 def _graph_is_configured(self
):
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
, message_iterator_class
=MyIter
):
159 def __init__(comp_self
, params
):
160 comp_self
._add
_output
_port
('clear')
161 comp_self
._add
_output
_port
('print')
162 comp_self
._add
_output
_port
('insert')
164 with self
.assertRaises(KeyError):
165 comp_self
._output
_ports
['hello']
167 self
._create
_comp
(MySource
)
169 def test_user_flt_output_ports_getitem_invalid_key(self
):
170 class MyIter(bt2
._UserMessageIterator
):
174 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
175 def __init__(comp_self
, params
):
176 comp_self
._add
_output
_port
('clear')
177 comp_self
._add
_output
_port
('print')
178 comp_self
._add
_output
_port
('insert')
180 with self
.assertRaises(KeyError):
181 comp_self
._output
_ports
['hello']
183 self
._create
_comp
(MyFilter
)
185 def test_user_flt_input_ports_getitem_invalid_key(self
):
186 class MyIter(bt2
._UserMessageIterator
):
190 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
191 def __init__(comp_self
, params
):
192 comp_self
._add
_input
_port
('clear')
193 comp_self
._add
_input
_port
('print')
194 comp_self
._add
_input
_port
('insert')
196 with self
.assertRaises(KeyError):
197 comp_self
._input
_ports
['hello']
199 self
._create
_comp
(MyFilter
)
201 def test_user_sink_input_ports_getitem_invalid_key(self
):
202 class MySink(bt2
._UserSinkComponent
):
203 def __init__(comp_self
, params
):
204 comp_self
._add
_input
_port
('clear')
205 comp_self
._add
_input
_port
('print')
206 comp_self
._add
_input
_port
('insert')
208 with self
.assertRaises(KeyError):
209 comp_self
._input
_ports
['hello']
214 def _graph_is_configured(self
):
217 self
._create
_comp
(MySink
)
219 def test_user_src_output_ports_len(self
):
220 class MyIter(bt2
._UserMessageIterator
):
224 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
225 def __init__(comp_self
, params
):
226 comp_self
._add
_output
_port
('clear')
227 comp_self
._add
_output
_port
('print')
228 comp_self
._add
_output
_port
('insert')
229 self
.assertEqual(len(comp_self
._output
_ports
), 3)
231 self
._create
_comp
(MySource
)
233 def test_user_flt_output_ports_len(self
):
234 class MyIter(bt2
._UserMessageIterator
):
238 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
239 def __init__(comp_self
, params
):
240 comp_self
._add
_output
_port
('clear')
241 comp_self
._add
_output
_port
('print')
242 comp_self
._add
_output
_port
('insert')
243 self
.assertEqual(len(comp_self
._output
_ports
), 3)
245 self
._create
_comp
(MyFilter
)
247 def test_user_flt_input_ports_len(self
):
248 class MyIter(bt2
._UserMessageIterator
):
252 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
253 def __init__(comp_self
, params
):
254 comp_self
._add
_input
_port
('clear')
255 comp_self
._add
_input
_port
('print')
256 comp_self
._add
_input
_port
('insert')
257 self
.assertEqual(len(comp_self
._input
_ports
), 3)
259 self
._create
_comp
(MyFilter
)
261 def test_user_sink_input_ports_len(self
):
262 class MySink(bt2
._UserSinkComponent
):
263 def __init__(comp_self
, params
):
264 comp_self
._add
_input
_port
('clear')
265 comp_self
._add
_input
_port
('print')
266 comp_self
._add
_input
_port
('insert')
267 self
.assertEqual(len(comp_self
._input
_ports
), 3)
272 def _graph_is_configured(self
):
275 self
._create
_comp
(MySink
)
277 def test_user_src_output_ports_iter(self
):
278 class MyIter(bt2
._UserMessageIterator
):
282 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
283 def __init__(comp_self
, params
):
284 port1
= comp_self
._add
_output
_port
('clear')
285 port2
= comp_self
._add
_output
_port
('print')
286 port3
= comp_self
._add
_output
_port
('insert')
289 for port_name
, port
in comp_self
._output
_ports
.items():
290 ports
.append((port_name
, port
))
292 self
.assertEqual(ports
[0][0], 'clear')
293 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
294 self
.assertEqual(ports
[1][0], 'print')
295 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
296 self
.assertEqual(ports
[2][0], 'insert')
297 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
299 self
._create
_comp
(MySource
)
301 def test_user_flt_output_ports_iter(self
):
302 class MyIter(bt2
._UserMessageIterator
):
306 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
307 def __init__(comp_self
, params
):
308 port1
= comp_self
._add
_output
_port
('clear')
309 port2
= comp_self
._add
_output
_port
('print')
310 port3
= comp_self
._add
_output
_port
('insert')
313 for port_name
, port
in comp_self
._output
_ports
.items():
314 ports
.append((port_name
, port
))
316 self
.assertEqual(ports
[0][0], 'clear')
317 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
318 self
.assertEqual(ports
[1][0], 'print')
319 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
320 self
.assertEqual(ports
[2][0], 'insert')
321 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
323 self
._create
_comp
(MyFilter
)
325 def test_user_flt_input_ports_iter(self
):
326 class MyIter(bt2
._UserMessageIterator
):
330 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
331 def __init__(comp_self
, params
):
332 port1
= comp_self
._add
_input
_port
('clear')
333 port2
= comp_self
._add
_input
_port
('print')
334 port3
= comp_self
._add
_input
_port
('insert')
337 for port_name
, port
in comp_self
._input
_ports
.items():
338 ports
.append((port_name
, port
))
340 self
.assertEqual(ports
[0][0], 'clear')
341 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
342 self
.assertEqual(ports
[1][0], 'print')
343 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
344 self
.assertEqual(ports
[2][0], 'insert')
345 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
347 self
._create
_comp
(MyFilter
)
349 def test_user_sink_input_ports_iter(self
):
350 class MySink(bt2
._UserSinkComponent
):
351 def __init__(comp_self
, params
):
352 port1
= comp_self
._add
_input
_port
('clear')
353 port2
= comp_self
._add
_input
_port
('print')
354 port3
= comp_self
._add
_input
_port
('insert')
357 for port_name
, port
in comp_self
._input
_ports
.items():
358 ports
.append((port_name
, port
))
360 self
.assertEqual(ports
[0][0], 'clear')
361 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
362 self
.assertEqual(ports
[1][0], 'print')
363 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
364 self
.assertEqual(ports
[2][0], 'insert')
365 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
370 def _graph_is_configured(self
):
373 self
._create
_comp
(MySink
)
375 def test_gen_src_output_ports_getitem(self
):
376 class MyIter(bt2
._UserMessageIterator
):
384 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
385 def __init__(comp_self
, params
):
386 nonlocal port1
, port2
, port3
387 port1
= comp_self
._add
_output
_port
('clear')
388 port2
= comp_self
._add
_output
_port
('print')
389 port3
= comp_self
._add
_output
_port
('insert')
391 comp
= self
._create
_comp
(MySource
)
392 self
.assertEqual(port3
.addr
, comp
.output_ports
['insert'].addr
)
393 self
.assertEqual(port2
.addr
, comp
.output_ports
['print'].addr
)
394 self
.assertEqual(port1
.addr
, comp
.output_ports
['clear'].addr
)
399 def test_gen_flt_output_ports_getitem(self
):
400 class MyIter(bt2
._UserMessageIterator
):
408 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
409 def __init__(comp_self
, params
):
410 nonlocal port1
, port2
, port3
411 port1
= comp_self
._add
_output
_port
('clear')
412 port2
= comp_self
._add
_output
_port
('print')
413 port3
= comp_self
._add
_output
_port
('insert')
415 comp
= self
._create
_comp
(MyFilter
)
416 self
.assertEqual(port3
.addr
, comp
.output_ports
['insert'].addr
)
417 self
.assertEqual(port2
.addr
, comp
.output_ports
['print'].addr
)
418 self
.assertEqual(port1
.addr
, comp
.output_ports
['clear'].addr
)
423 def test_gen_flt_input_ports_getitem(self
):
424 class MyIter(bt2
._UserMessageIterator
):
432 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
433 def __init__(comp_self
, params
):
434 nonlocal port1
, port2
, port3
435 port1
= comp_self
._add
_input
_port
('clear')
436 port2
= comp_self
._add
_input
_port
('print')
437 port3
= comp_self
._add
_input
_port
('insert')
439 comp
= self
._create
_comp
(MyFilter
)
440 self
.assertEqual(port3
.addr
, comp
.input_ports
['insert'].addr
)
441 self
.assertEqual(port2
.addr
, comp
.input_ports
['print'].addr
)
442 self
.assertEqual(port1
.addr
, comp
.input_ports
['clear'].addr
)
447 def test_gen_sink_input_ports_getitem(self
):
452 class MySink(bt2
._UserSinkComponent
):
453 def __init__(comp_self
, params
):
454 nonlocal port1
, port2
, port3
455 port1
= comp_self
._add
_input
_port
('clear')
456 port2
= comp_self
._add
_input
_port
('print')
457 port3
= comp_self
._add
_input
_port
('insert')
462 def _graph_is_configured(self
):
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
, message_iterator_class
=MyIter
):
479 def __init__(comp_self
, params
):
480 comp_self
._add
_output
_port
('clear')
481 comp_self
._add
_output
_port
('print')
482 comp_self
._add
_output
_port
('insert')
484 comp
= self
._create
_comp
(MySource
)
486 with self
.assertRaises(KeyError):
487 comp
.output_ports
['hello']
489 def test_gen_flt_output_ports_getitem_invalid_key(self
):
490 class MyIter(bt2
._UserMessageIterator
):
494 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
495 def __init__(comp_self
, params
):
496 comp_self
._add
_output
_port
('clear')
497 comp_self
._add
_output
_port
('print')
498 comp_self
._add
_output
_port
('insert')
500 comp
= self
._create
_comp
(MyFilter
)
502 with self
.assertRaises(KeyError):
503 comp
.output_ports
['hello']
505 def test_gen_flt_input_ports_getitem_invalid_key(self
):
506 class MyIter(bt2
._UserMessageIterator
):
510 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
511 def __init__(comp_self
, params
):
512 comp_self
._add
_input
_port
('clear')
513 comp_self
._add
_input
_port
('print')
514 comp_self
._add
_input
_port
('insert')
516 comp
= self
._create
_comp
(MyFilter
)
518 with self
.assertRaises(KeyError):
519 comp
.input_ports
['hello']
521 def test_gen_sink_input_ports_getitem_invalid_key(self
):
522 class MySink(bt2
._UserSinkComponent
):
523 def __init__(comp_self
, params
):
524 comp_self
._add
_input
_port
('clear')
525 comp_self
._add
_input
_port
('print')
526 comp_self
._add
_input
_port
('insert')
528 with self
.assertRaises(KeyError):
529 comp_self
._input
_ports
['hello']
534 def _graph_is_configured(self
):
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
, message_iterator_class
=MyIter
):
548 def __init__(comp_self
, params
):
549 comp_self
._add
_output
_port
('clear')
550 comp_self
._add
_output
_port
('print')
551 comp_self
._add
_output
_port
('insert')
553 comp
= self
._create
_comp
(MySource
)
554 self
.assertEqual(len(comp
.output_ports
), 3)
556 def test_gen_flt_output_ports_len(self
):
557 class MyIter(bt2
._UserMessageIterator
):
561 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
562 def __init__(comp_self
, params
):
563 comp_self
._add
_output
_port
('clear')
564 comp_self
._add
_output
_port
('print')
565 comp_self
._add
_output
_port
('insert')
567 comp
= self
._create
_comp
(MyFilter
)
568 self
.assertEqual(len(comp
.output_ports
), 3)
570 def test_gen_flt_input_ports_len(self
):
571 class MyIter(bt2
._UserMessageIterator
):
575 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
576 def __init__(comp_self
, params
):
577 comp_self
._add
_input
_port
('clear')
578 comp_self
._add
_input
_port
('print')
579 comp_self
._add
_input
_port
('insert')
581 comp
= self
._create
_comp
(MyFilter
)
582 self
.assertEqual(len(comp
.input_ports
), 3)
584 def test_gen_sink_input_ports_len(self
):
585 class MySink(bt2
._UserSinkComponent
):
586 def __init__(comp_self
, params
):
587 comp_self
._add
_input
_port
('clear')
588 comp_self
._add
_input
_port
('print')
589 comp_self
._add
_input
_port
('insert')
594 def _graph_is_configured(self
):
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
, message_iterator_class
=MyIter
):
610 def __init__(comp_self
, params
):
611 nonlocal port1
, port2
, port3
612 port1
= comp_self
._add
_output
_port
('clear')
613 port2
= comp_self
._add
_output
_port
('print')
614 port3
= comp_self
._add
_output
_port
('insert')
616 comp
= self
._create
_comp
(MySource
)
619 for port_name
, port
in comp
.output_ports
.items():
620 ports
.append((port_name
, port
))
622 self
.assertEqual(ports
[0][0], 'clear')
623 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
624 self
.assertEqual(ports
[1][0], 'print')
625 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
626 self
.assertEqual(ports
[2][0], 'insert')
627 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
632 def test_gen_flt_output_ports_iter(self
):
633 class MyIter(bt2
._UserMessageIterator
):
641 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
642 def __init__(comp_self
, params
):
643 nonlocal port1
, port2
, port3
644 port1
= comp_self
._add
_output
_port
('clear')
645 port2
= comp_self
._add
_output
_port
('print')
646 port3
= comp_self
._add
_output
_port
('insert')
648 comp
= self
._create
_comp
(MyFilter
)
651 for port_name
, port
in comp
.output_ports
.items():
652 ports
.append((port_name
, port
))
654 self
.assertEqual(ports
[0][0], 'clear')
655 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
656 self
.assertEqual(ports
[1][0], 'print')
657 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
658 self
.assertEqual(ports
[2][0], 'insert')
659 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
664 def test_gen_flt_input_ports_iter(self
):
665 class MyIter(bt2
._UserMessageIterator
):
673 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
674 def __init__(comp_self
, params
):
675 nonlocal port1
, port2
, port3
676 port1
= comp_self
._add
_input
_port
('clear')
677 port2
= comp_self
._add
_input
_port
('print')
678 port3
= comp_self
._add
_input
_port
('insert')
680 comp
= self
._create
_comp
(MyFilter
)
683 for port_name
, port
in comp
.input_ports
.items():
684 ports
.append((port_name
, port
))
686 self
.assertEqual(ports
[0][0], 'clear')
687 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
688 self
.assertEqual(ports
[1][0], 'print')
689 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
690 self
.assertEqual(ports
[2][0], 'insert')
691 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
696 def test_gen_sink_input_ports_iter(self
):
701 class MySink(bt2
._UserSinkComponent
):
702 def __init__(comp_self
, params
):
703 nonlocal port1
, port2
, port3
704 port1
= comp_self
._add
_input
_port
('clear')
705 port2
= comp_self
._add
_input
_port
('print')
706 port3
= comp_self
._add
_input
_port
('insert')
711 def _graph_is_configured(self
):
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 def _graph_is_configured(self
):
741 comp
= self
._create
_comp
(MySink
)
742 self
.assertEqual(comp
.input_ports
['clear'].name
, 'clear')
744 def test_connection_none(self
):
745 class MySink(bt2
._UserSinkComponent
):
746 def __init__(comp_self
, params
):
747 comp_self
._add
_input
_port
('clear')
752 def _graph_is_configured(self
):
755 comp
= self
._create
_comp
(MySink
)
756 self
.assertIsNone(comp
.input_ports
['clear'].connection
)
758 def test_is_connected_false(self
):
759 class MySink(bt2
._UserSinkComponent
):
760 def __init__(comp_self
, params
):
761 comp_self
._add
_input
_port
('clear')
766 def _graph_is_configured(self
):
769 comp
= self
._create
_comp
(MySink
)
770 self
.assertFalse(comp
.input_ports
['clear'].is_connected
)
772 def test_self_name(self
):
773 class MySink(bt2
._UserSinkComponent
):
774 def __init__(comp_self
, params
):
775 port
= comp_self
._add
_input
_port
('clear')
776 self
.assertEqual(port
.name
, 'clear')
781 def _graph_is_configured(self
):
784 self
._create
_comp
(MySink
)
786 def test_self_connection_none(self
):
787 class MySink(bt2
._UserSinkComponent
):
788 def __init__(comp_self
, params
):
789 port
= comp_self
._add
_input
_port
('clear')
790 self
.assertIsNone(port
.connection
)
795 def _graph_is_configured(self
):
798 self
._create
_comp
(MySink
)
800 def test_self_is_connected_false(self
):
801 class MySink(bt2
._UserSinkComponent
):
802 def __init__(comp_self
, params
):
803 port
= comp_self
._add
_input
_port
('clear')
804 self
.assertFalse(port
.is_connected
)
809 def _graph_is_configured(self
):
812 self
._create
_comp
(MySink
)
814 def test_source_self_port_user_data(self
):
815 class MyIter(bt2
._UserMessageIterator
):
819 class MySource(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
820 def __init__(comp_self
, params
):
823 p
= comp_self
._add
_output
_port
('port1')
824 user_datas
.append(p
.user_data
)
825 p
= comp_self
._add
_output
_port
('port2', 2)
826 user_datas
.append(p
.user_data
)
830 comp
= self
._create
_comp
(MySource
)
831 self
.assertEqual(user_datas
, [None, 2])
833 def test_filter_self_port_user_data(self
):
834 class MyIter(bt2
._UserMessageIterator
):
838 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
839 def __init__(comp_self
, params
):
842 p
= comp_self
._add
_output
_port
('port1')
843 user_datas
.append(p
.user_data
)
844 p
= comp_self
._add
_output
_port
('port2', 'user data string')
845 user_datas
.append(p
.user_data
)
847 p
= comp_self
._add
_input
_port
('port3')
848 user_datas
.append(p
.user_data
)
849 p
= comp_self
._add
_input
_port
('port4', user_data
={'user data': 'dict'})
850 user_datas
.append(p
.user_data
)
854 comp
= self
._create
_comp
(MyFilter
)
856 user_datas
, [None, 'user data string', None, {'user data': 'dict'}]
859 def test_sink_self_port_user_data(self
):
860 class MyIter(bt2
._UserMessageIterator
):
864 class MySink(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
865 def __init__(comp_self
, params
):
868 p
= comp_self
._add
_input
_port
('port1')
869 user_datas
.append(p
.user_data
)
870 p
= comp_self
._add
_input
_port
('port2', set())
871 user_datas
.append(p
.user_data
)
875 comp
= self
._create
_comp
(MySink
)
876 self
.assertEqual(user_datas
, [None, set()])
This page took 0.05169 seconds and 4 git commands to generate.