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 def _graph_is_configured(self
):
87 comp
= self
._create
_comp
(MySink
)
88 self
.assertEqual(len(comp
.input_ports
), 1)
90 def test_user_src_output_ports_getitem(self
):
91 class MyIter(bt2
._UserMessageIterator
):
95 class MySource(bt2
._UserSourceComponent
,
96 message_iterator_class
=MyIter
):
97 def __init__(comp_self
, params
):
98 port1
= comp_self
._add
_output
_port
('clear')
99 port2
= comp_self
._add
_output
_port
('print')
100 port3
= comp_self
._add
_output
_port
('insert')
101 self
.assertEqual(port3
.addr
, comp_self
._output
_ports
['insert'].addr
)
102 self
.assertEqual(port2
.addr
, comp_self
._output
_ports
['print'].addr
)
103 self
.assertEqual(port1
.addr
, comp_self
._output
_ports
['clear'].addr
)
105 self
._create
_comp
(MySource
)
107 def test_user_flt_output_ports_getitem(self
):
108 class MyIter(bt2
._UserMessageIterator
):
112 class MyFilter(bt2
._UserFilterComponent
,
113 message_iterator_class
=MyIter
):
114 def __init__(comp_self
, params
):
115 port1
= comp_self
._add
_output
_port
('clear')
116 port2
= comp_self
._add
_output
_port
('print')
117 port3
= comp_self
._add
_output
_port
('insert')
118 self
.assertEqual(port3
.addr
, comp_self
._output
_ports
['insert'].addr
)
119 self
.assertEqual(port2
.addr
, comp_self
._output
_ports
['print'].addr
)
120 self
.assertEqual(port1
.addr
, comp_self
._output
_ports
['clear'].addr
)
122 self
._create
_comp
(MyFilter
)
124 def test_user_flt_input_ports_getitem(self
):
125 class MyIter(bt2
._UserMessageIterator
):
129 class MyFilter(bt2
._UserFilterComponent
,
130 message_iterator_class
=MyIter
):
131 def __init__(comp_self
, params
):
132 port1
= comp_self
._add
_input
_port
('clear')
133 port2
= comp_self
._add
_input
_port
('print')
134 port3
= comp_self
._add
_input
_port
('insert')
135 self
.assertEqual(port3
.addr
, comp_self
._input
_ports
['insert'].addr
)
136 self
.assertEqual(port2
.addr
, comp_self
._input
_ports
['print'].addr
)
137 self
.assertEqual(port1
.addr
, comp_self
._input
_ports
['clear'].addr
)
139 self
._create
_comp
(MyFilter
)
141 def test_user_sink_input_ports_getitem(self
):
142 class MySink(bt2
._UserSinkComponent
):
143 def __init__(comp_self
, params
):
144 port1
= comp_self
._add
_input
_port
('clear')
145 port2
= comp_self
._add
_input
_port
('print')
146 port3
= comp_self
._add
_input
_port
('insert')
147 self
.assertEqual(port3
.addr
, comp_self
._input
_ports
['insert'].addr
)
148 self
.assertEqual(port2
.addr
, comp_self
._input
_ports
['print'].addr
)
149 self
.assertEqual(port1
.addr
, comp_self
._input
_ports
['clear'].addr
)
154 def _graph_is_configured(self
):
157 self
._create
_comp
(MySink
)
159 def test_user_src_output_ports_getitem_invalid_key(self
):
160 class MyIter(bt2
._UserMessageIterator
):
164 class MySource(bt2
._UserSourceComponent
,
165 message_iterator_class
=MyIter
):
166 def __init__(comp_self
, params
):
167 comp_self
._add
_output
_port
('clear')
168 comp_self
._add
_output
_port
('print')
169 comp_self
._add
_output
_port
('insert')
171 with self
.assertRaises(KeyError):
172 comp_self
._output
_ports
['hello']
174 self
._create
_comp
(MySource
)
176 def test_user_flt_output_ports_getitem_invalid_key(self
):
177 class MyIter(bt2
._UserMessageIterator
):
181 class MyFilter(bt2
._UserFilterComponent
,
182 message_iterator_class
=MyIter
):
183 def __init__(comp_self
, params
):
184 comp_self
._add
_output
_port
('clear')
185 comp_self
._add
_output
_port
('print')
186 comp_self
._add
_output
_port
('insert')
188 with self
.assertRaises(KeyError):
189 comp_self
._output
_ports
['hello']
191 self
._create
_comp
(MyFilter
)
193 def test_user_flt_input_ports_getitem_invalid_key(self
):
194 class MyIter(bt2
._UserMessageIterator
):
198 class MyFilter(bt2
._UserFilterComponent
,
199 message_iterator_class
=MyIter
):
200 def __init__(comp_self
, params
):
201 comp_self
._add
_input
_port
('clear')
202 comp_self
._add
_input
_port
('print')
203 comp_self
._add
_input
_port
('insert')
205 with self
.assertRaises(KeyError):
206 comp_self
._input
_ports
['hello']
208 self
._create
_comp
(MyFilter
)
210 def test_user_sink_input_ports_getitem_invalid_key(self
):
211 class MySink(bt2
._UserSinkComponent
):
212 def __init__(comp_self
, params
):
213 comp_self
._add
_input
_port
('clear')
214 comp_self
._add
_input
_port
('print')
215 comp_self
._add
_input
_port
('insert')
217 with self
.assertRaises(KeyError):
218 comp_self
._input
_ports
['hello']
223 def _graph_is_configured(self
):
226 self
._create
_comp
(MySink
)
228 def test_user_src_output_ports_len(self
):
229 class MyIter(bt2
._UserMessageIterator
):
233 class MySource(bt2
._UserSourceComponent
,
234 message_iterator_class
=MyIter
):
235 def __init__(comp_self
, params
):
236 comp_self
._add
_output
_port
('clear')
237 comp_self
._add
_output
_port
('print')
238 comp_self
._add
_output
_port
('insert')
239 self
.assertEqual(len(comp_self
._output
_ports
), 3)
241 self
._create
_comp
(MySource
)
243 def test_user_flt_output_ports_len(self
):
244 class MyIter(bt2
._UserMessageIterator
):
248 class MyFilter(bt2
._UserFilterComponent
,
249 message_iterator_class
=MyIter
):
250 def __init__(comp_self
, params
):
251 comp_self
._add
_output
_port
('clear')
252 comp_self
._add
_output
_port
('print')
253 comp_self
._add
_output
_port
('insert')
254 self
.assertEqual(len(comp_self
._output
_ports
), 3)
256 self
._create
_comp
(MyFilter
)
258 def test_user_flt_input_ports_len(self
):
259 class MyIter(bt2
._UserMessageIterator
):
263 class MyFilter(bt2
._UserFilterComponent
,
264 message_iterator_class
=MyIter
):
265 def __init__(comp_self
, params
):
266 comp_self
._add
_input
_port
('clear')
267 comp_self
._add
_input
_port
('print')
268 comp_self
._add
_input
_port
('insert')
269 self
.assertEqual(len(comp_self
._input
_ports
), 3)
271 self
._create
_comp
(MyFilter
)
273 def test_user_sink_input_ports_len(self
):
274 class MySink(bt2
._UserSinkComponent
):
275 def __init__(comp_self
, params
):
276 comp_self
._add
_input
_port
('clear')
277 comp_self
._add
_input
_port
('print')
278 comp_self
._add
_input
_port
('insert')
279 self
.assertEqual(len(comp_self
._input
_ports
), 3)
284 def _graph_is_configured(self
):
287 self
._create
_comp
(MySink
)
289 def test_user_src_output_ports_iter(self
):
290 class MyIter(bt2
._UserMessageIterator
):
294 class MySource(bt2
._UserSourceComponent
,
295 message_iterator_class
=MyIter
):
296 def __init__(comp_self
, params
):
297 port1
= comp_self
._add
_output
_port
('clear')
298 port2
= comp_self
._add
_output
_port
('print')
299 port3
= comp_self
._add
_output
_port
('insert')
302 for port_name
, port
in comp_self
._output
_ports
.items():
303 ports
.append((port_name
, port
))
305 self
.assertEqual(ports
[0][0], 'clear')
306 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
307 self
.assertEqual(ports
[1][0], 'print')
308 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
309 self
.assertEqual(ports
[2][0], 'insert')
310 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
312 self
._create
_comp
(MySource
)
314 def test_user_flt_output_ports_iter(self
):
315 class MyIter(bt2
._UserMessageIterator
):
319 class MyFilter(bt2
._UserFilterComponent
,
320 message_iterator_class
=MyIter
):
321 def __init__(comp_self
, params
):
322 port1
= comp_self
._add
_output
_port
('clear')
323 port2
= comp_self
._add
_output
_port
('print')
324 port3
= comp_self
._add
_output
_port
('insert')
327 for port_name
, port
in comp_self
._output
_ports
.items():
328 ports
.append((port_name
, port
))
330 self
.assertEqual(ports
[0][0], 'clear')
331 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
332 self
.assertEqual(ports
[1][0], 'print')
333 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
334 self
.assertEqual(ports
[2][0], 'insert')
335 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
337 self
._create
_comp
(MyFilter
)
339 def test_user_flt_input_ports_iter(self
):
340 class MyIter(bt2
._UserMessageIterator
):
344 class MyFilter(bt2
._UserFilterComponent
,
345 message_iterator_class
=MyIter
):
346 def __init__(comp_self
, params
):
347 port1
= comp_self
._add
_input
_port
('clear')
348 port2
= comp_self
._add
_input
_port
('print')
349 port3
= comp_self
._add
_input
_port
('insert')
352 for port_name
, port
in comp_self
._input
_ports
.items():
353 ports
.append((port_name
, port
))
355 self
.assertEqual(ports
[0][0], 'clear')
356 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
357 self
.assertEqual(ports
[1][0], 'print')
358 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
359 self
.assertEqual(ports
[2][0], 'insert')
360 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
362 self
._create
_comp
(MyFilter
)
364 def test_user_sink_input_ports_iter(self
):
365 class MySink(bt2
._UserSinkComponent
):
366 def __init__(comp_self
, params
):
367 port1
= comp_self
._add
_input
_port
('clear')
368 port2
= comp_self
._add
_input
_port
('print')
369 port3
= comp_self
._add
_input
_port
('insert')
372 for port_name
, port
in comp_self
._input
_ports
.items():
373 ports
.append((port_name
, port
))
375 self
.assertEqual(ports
[0][0], 'clear')
376 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
377 self
.assertEqual(ports
[1][0], 'print')
378 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
379 self
.assertEqual(ports
[2][0], 'insert')
380 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
385 def _graph_is_configured(self
):
388 self
._create
_comp
(MySink
)
390 def test_gen_src_output_ports_getitem(self
):
391 class MyIter(bt2
._UserMessageIterator
):
399 class MySource(bt2
._UserSourceComponent
,
400 message_iterator_class
=MyIter
):
401 def __init__(comp_self
, params
):
402 nonlocal port1
, port2
, port3
403 port1
= comp_self
._add
_output
_port
('clear')
404 port2
= comp_self
._add
_output
_port
('print')
405 port3
= comp_self
._add
_output
_port
('insert')
407 comp
= self
._create
_comp
(MySource
)
408 self
.assertEqual(port3
.addr
, comp
.output_ports
['insert'].addr
)
409 self
.assertEqual(port2
.addr
, comp
.output_ports
['print'].addr
)
410 self
.assertEqual(port1
.addr
, comp
.output_ports
['clear'].addr
)
415 def test_gen_flt_output_ports_getitem(self
):
416 class MyIter(bt2
._UserMessageIterator
):
424 class MyFilter(bt2
._UserFilterComponent
,
425 message_iterator_class
=MyIter
):
426 def __init__(comp_self
, params
):
427 nonlocal port1
, port2
, port3
428 port1
= comp_self
._add
_output
_port
('clear')
429 port2
= comp_self
._add
_output
_port
('print')
430 port3
= comp_self
._add
_output
_port
('insert')
432 comp
= self
._create
_comp
(MyFilter
)
433 self
.assertEqual(port3
.addr
, comp
.output_ports
['insert'].addr
)
434 self
.assertEqual(port2
.addr
, comp
.output_ports
['print'].addr
)
435 self
.assertEqual(port1
.addr
, comp
.output_ports
['clear'].addr
)
440 def test_gen_flt_input_ports_getitem(self
):
441 class MyIter(bt2
._UserMessageIterator
):
449 class MyFilter(bt2
._UserFilterComponent
,
450 message_iterator_class
=MyIter
):
451 def __init__(comp_self
, params
):
452 nonlocal port1
, port2
, port3
453 port1
= comp_self
._add
_input
_port
('clear')
454 port2
= comp_self
._add
_input
_port
('print')
455 port3
= comp_self
._add
_input
_port
('insert')
457 comp
= self
._create
_comp
(MyFilter
)
458 self
.assertEqual(port3
.addr
, comp
.input_ports
['insert'].addr
)
459 self
.assertEqual(port2
.addr
, comp
.input_ports
['print'].addr
)
460 self
.assertEqual(port1
.addr
, comp
.input_ports
['clear'].addr
)
465 def test_gen_sink_input_ports_getitem(self
):
470 class MySink(bt2
._UserSinkComponent
):
471 def __init__(comp_self
, params
):
472 nonlocal port1
, port2
, port3
473 port1
= comp_self
._add
_input
_port
('clear')
474 port2
= comp_self
._add
_input
_port
('print')
475 port3
= comp_self
._add
_input
_port
('insert')
480 def _graph_is_configured(self
):
483 comp
= self
._create
_comp
(MySink
)
484 self
.assertEqual(port3
.addr
, comp
.input_ports
['insert'].addr
)
485 self
.assertEqual(port2
.addr
, comp
.input_ports
['print'].addr
)
486 self
.assertEqual(port1
.addr
, comp
.input_ports
['clear'].addr
)
491 def test_gen_src_output_ports_getitem_invalid_key(self
):
492 class MyIter(bt2
._UserMessageIterator
):
496 class MySource(bt2
._UserSourceComponent
,
497 message_iterator_class
=MyIter
):
498 def __init__(comp_self
, params
):
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
):
509 class MyIter(bt2
._UserMessageIterator
):
513 class MyFilter(bt2
._UserFilterComponent
,
514 message_iterator_class
=MyIter
):
515 def __init__(comp_self
, params
):
516 comp_self
._add
_output
_port
('clear')
517 comp_self
._add
_output
_port
('print')
518 comp_self
._add
_output
_port
('insert')
520 comp
= self
._create
_comp
(MyFilter
)
522 with self
.assertRaises(KeyError):
523 comp
.output_ports
['hello']
525 def test_gen_flt_input_ports_getitem_invalid_key(self
):
526 class MyIter(bt2
._UserMessageIterator
):
530 class MyFilter(bt2
._UserFilterComponent
,
531 message_iterator_class
=MyIter
):
532 def __init__(comp_self
, params
):
533 comp_self
._add
_input
_port
('clear')
534 comp_self
._add
_input
_port
('print')
535 comp_self
._add
_input
_port
('insert')
537 comp
= self
._create
_comp
(MyFilter
)
539 with self
.assertRaises(KeyError):
540 comp
.input_ports
['hello']
542 def test_gen_sink_input_ports_getitem_invalid_key(self
):
543 class MySink(bt2
._UserSinkComponent
):
544 def __init__(comp_self
, params
):
545 comp_self
._add
_input
_port
('clear')
546 comp_self
._add
_input
_port
('print')
547 comp_self
._add
_input
_port
('insert')
549 with self
.assertRaises(KeyError):
550 comp_self
._input
_ports
['hello']
555 def _graph_is_configured(self
):
558 comp
= self
._create
_comp
(MySink
)
560 with self
.assertRaises(KeyError):
561 comp
.input_ports
['hello']
563 def test_gen_src_output_ports_len(self
):
564 class MyIter(bt2
._UserMessageIterator
):
568 class MySource(bt2
._UserSourceComponent
,
569 message_iterator_class
=MyIter
):
570 def __init__(comp_self
, params
):
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
(MySource
)
576 self
.assertEqual(len(comp
.output_ports
), 3)
578 def test_gen_flt_output_ports_len(self
):
579 class MyIter(bt2
._UserMessageIterator
):
583 class MyFilter(bt2
._UserFilterComponent
,
584 message_iterator_class
=MyIter
):
585 def __init__(comp_self
, params
):
586 comp_self
._add
_output
_port
('clear')
587 comp_self
._add
_output
_port
('print')
588 comp_self
._add
_output
_port
('insert')
590 comp
= self
._create
_comp
(MyFilter
)
591 self
.assertEqual(len(comp
.output_ports
), 3)
593 def test_gen_flt_input_ports_len(self
):
594 class MyIter(bt2
._UserMessageIterator
):
598 class MyFilter(bt2
._UserFilterComponent
,
599 message_iterator_class
=MyIter
):
600 def __init__(comp_self
, params
):
601 comp_self
._add
_input
_port
('clear')
602 comp_self
._add
_input
_port
('print')
603 comp_self
._add
_input
_port
('insert')
605 comp
= self
._create
_comp
(MyFilter
)
606 self
.assertEqual(len(comp
.input_ports
), 3)
608 def test_gen_sink_input_ports_len(self
):
609 class MySink(bt2
._UserSinkComponent
):
610 def __init__(comp_self
, params
):
611 comp_self
._add
_input
_port
('clear')
612 comp_self
._add
_input
_port
('print')
613 comp_self
._add
_input
_port
('insert')
618 def _graph_is_configured(self
):
621 comp
= self
._create
_comp
(MySink
)
622 self
.assertEqual(len(comp
.input_ports
), 3)
624 def test_gen_src_output_ports_iter(self
):
625 class MyIter(bt2
._UserMessageIterator
):
633 class MySource(bt2
._UserSourceComponent
,
634 message_iterator_class
=MyIter
):
635 def __init__(comp_self
, params
):
636 nonlocal port1
, port2
, port3
637 port1
= comp_self
._add
_output
_port
('clear')
638 port2
= comp_self
._add
_output
_port
('print')
639 port3
= comp_self
._add
_output
_port
('insert')
641 comp
= self
._create
_comp
(MySource
)
644 for port_name
, port
in comp
.output_ports
.items():
645 ports
.append((port_name
, port
))
647 self
.assertEqual(ports
[0][0], 'clear')
648 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
649 self
.assertEqual(ports
[1][0], 'print')
650 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
651 self
.assertEqual(ports
[2][0], 'insert')
652 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
657 def test_gen_flt_output_ports_iter(self
):
658 class MyIter(bt2
._UserMessageIterator
):
666 class MyFilter(bt2
._UserFilterComponent
,
667 message_iterator_class
=MyIter
):
668 def __init__(comp_self
, params
):
669 nonlocal port1
, port2
, port3
670 port1
= comp_self
._add
_output
_port
('clear')
671 port2
= comp_self
._add
_output
_port
('print')
672 port3
= comp_self
._add
_output
_port
('insert')
674 comp
= self
._create
_comp
(MyFilter
)
677 for port_name
, port
in comp
.output_ports
.items():
678 ports
.append((port_name
, port
))
680 self
.assertEqual(ports
[0][0], 'clear')
681 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
682 self
.assertEqual(ports
[1][0], 'print')
683 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
684 self
.assertEqual(ports
[2][0], 'insert')
685 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
690 def test_gen_flt_input_ports_iter(self
):
691 class MyIter(bt2
._UserMessageIterator
):
699 class MyFilter(bt2
._UserFilterComponent
,
700 message_iterator_class
=MyIter
):
701 def __init__(comp_self
, params
):
702 nonlocal port1
, port2
, port3
703 port1
= comp_self
._add
_input
_port
('clear')
704 port2
= comp_self
._add
_input
_port
('print')
705 port3
= comp_self
._add
_input
_port
('insert')
707 comp
= self
._create
_comp
(MyFilter
)
710 for port_name
, port
in comp
.input_ports
.items():
711 ports
.append((port_name
, port
))
713 self
.assertEqual(ports
[0][0], 'clear')
714 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
715 self
.assertEqual(ports
[1][0], 'print')
716 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
717 self
.assertEqual(ports
[2][0], 'insert')
718 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
723 def test_gen_sink_input_ports_iter(self
):
728 class MySink(bt2
._UserSinkComponent
):
729 def __init__(comp_self
, params
):
730 nonlocal port1
, port2
, port3
731 port1
= comp_self
._add
_input
_port
('clear')
732 port2
= comp_self
._add
_input
_port
('print')
733 port3
= comp_self
._add
_input
_port
('insert')
738 def _graph_is_configured(self
):
741 comp
= self
._create
_comp
(MySink
)
744 for port_name
, port
in comp
.input_ports
.items():
745 ports
.append((port_name
, port
))
747 self
.assertEqual(ports
[0][0], 'clear')
748 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
749 self
.assertEqual(ports
[1][0], 'print')
750 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
751 self
.assertEqual(ports
[2][0], 'insert')
752 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
758 class MySink(bt2
._UserSinkComponent
):
759 def __init__(comp_self
, params
):
760 comp_self
._add
_input
_port
('clear')
765 def _graph_is_configured(self
):
768 comp
= self
._create
_comp
(MySink
)
769 self
.assertEqual(comp
.input_ports
['clear'].name
, 'clear')
771 def test_connection_none(self
):
772 class MySink(bt2
._UserSinkComponent
):
773 def __init__(comp_self
, params
):
774 comp_self
._add
_input
_port
('clear')
779 def _graph_is_configured(self
):
782 comp
= self
._create
_comp
(MySink
)
783 self
.assertIsNone(comp
.input_ports
['clear'].connection
)
785 def test_is_connected_false(self
):
786 class MySink(bt2
._UserSinkComponent
):
787 def __init__(comp_self
, params
):
788 comp_self
._add
_input
_port
('clear')
793 def _graph_is_configured(self
):
796 comp
= self
._create
_comp
(MySink
)
797 self
.assertFalse(comp
.input_ports
['clear'].is_connected
)
799 def test_self_name(self
):
800 class MySink(bt2
._UserSinkComponent
):
801 def __init__(comp_self
, params
):
802 port
= comp_self
._add
_input
_port
('clear')
803 self
.assertEqual(port
.name
, 'clear')
808 def _graph_is_configured(self
):
811 self
._create
_comp
(MySink
)
813 def test_self_connection_none(self
):
814 class MySink(bt2
._UserSinkComponent
):
815 def __init__(comp_self
, params
):
816 port
= comp_self
._add
_input
_port
('clear')
817 self
.assertIsNone(port
.connection
)
822 def _graph_is_configured(self
):
825 self
._create
_comp
(MySink
)
827 def test_self_is_connected_false(self
):
828 class MySink(bt2
._UserSinkComponent
):
829 def __init__(comp_self
, params
):
830 port
= comp_self
._add
_input
_port
('clear')
831 self
.assertFalse(port
.is_connected
)
836 def _graph_is_configured(self
):
839 self
._create
_comp
(MySink
)
841 def test_source_self_port_user_data(self
):
842 class MyIter(bt2
._UserMessageIterator
):
846 class MySource(bt2
._UserFilterComponent
,
847 message_iterator_class
=MyIter
):
848 def __init__(comp_self
, params
):
851 p
= comp_self
._add
_output
_port
('port1')
852 user_datas
.append(p
.user_data
)
853 p
= comp_self
._add
_output
_port
('port2', 2)
854 user_datas
.append(p
.user_data
)
858 comp
= self
._create
_comp
(MySource
)
859 self
.assertEqual(user_datas
, [None, 2])
861 def test_filter_self_port_user_data(self
):
862 class MyIter(bt2
._UserMessageIterator
):
866 class MyFilter(bt2
._UserFilterComponent
,
867 message_iterator_class
=MyIter
):
868 def __init__(comp_self
, params
):
871 p
= comp_self
._add
_output
_port
('port1')
872 user_datas
.append(p
.user_data
)
873 p
= comp_self
._add
_output
_port
('port2', 'user data string')
874 user_datas
.append(p
.user_data
)
876 p
= comp_self
._add
_input
_port
('port3')
877 user_datas
.append(p
.user_data
)
878 p
= comp_self
._add
_input
_port
('port4', user_data
={'user data': 'dict'})
879 user_datas
.append(p
.user_data
)
883 comp
= self
._create
_comp
(MyFilter
)
884 self
.assertEqual(user_datas
,
885 [None, 'user data string', None, {'user data': 'dict'}])
887 def test_sink_self_port_user_data(self
):
888 class MyIter(bt2
._UserMessageIterator
):
892 class MySink(bt2
._UserFilterComponent
,
893 message_iterator_class
=MyIter
):
894 def __init__(comp_self
, params
):
897 p
= comp_self
._add
_input
_port
('port1')
898 user_datas
.append(p
.user_data
)
899 p
= comp_self
._add
_input
_port
('port2', set())
900 user_datas
.append(p
.user_data
)
904 comp
= self
._create
_comp
(MySink
)
905 self
.assertEqual(user_datas
, [None, set()])
This page took 0.048305 seconds and 5 git commands to generate.