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.
21 from bt2
import port
as bt2_port
24 class PortTestCase(unittest
.TestCase
):
26 def _create_comp(comp_cls
, name
=None):
32 return graph
.add_component(comp_cls
, name
)
34 def test_src_add_output_port(self
):
35 class MyIter(bt2
._UserMessageIterator
):
39 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
40 def __init__(comp_self
, config
, params
, obj
):
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)
46 self
.assertIs(type(comp
.output_ports
['out']), bt2_port
._OutputPortConst
)
48 def test_flt_add_output_port(self
):
49 class MyIter(bt2
._UserMessageIterator
):
53 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
54 def __init__(comp_self
, config
, params
, obj
):
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
, message_iterator_class
=MyIter
):
67 def __init__(comp_self
, config
, params
, obj
):
68 port
= comp_self
._add
_input
_port
('in')
69 self
.assertEqual(port
.name
, 'in')
71 comp
= self
._create
_comp
(MyFilter
)
72 self
.assertEqual(len(comp
.input_ports
), 1)
73 self
.assertIs(type(comp
.input_ports
['in']), bt2_port
._InputPortConst
)
75 def test_sink_add_input_port(self
):
76 class MySink(bt2
._UserSinkComponent
):
77 def __init__(comp_self
, config
, params
, obj
):
78 port
= comp_self
._add
_input
_port
('in')
79 self
.assertEqual(port
.name
, 'in')
81 def _user_consume(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
, config
, params
, obj
):
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
, config
, params
, obj
):
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
, config
, params
, obj
):
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
, config
, params
, obj
):
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
)
145 def _user_consume(self
):
148 self
._create
_comp
(MySink
)
150 def test_user_src_output_ports_getitem_invalid_key(self
):
151 class MyIter(bt2
._UserMessageIterator
):
155 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
156 def __init__(comp_self
, config
, params
, obj
):
157 comp_self
._add
_output
_port
('clear')
158 comp_self
._add
_output
_port
('print')
159 comp_self
._add
_output
_port
('insert')
161 with self
.assertRaises(KeyError):
162 comp_self
._output
_ports
['hello']
164 self
._create
_comp
(MySource
)
166 def test_user_flt_output_ports_getitem_invalid_key(self
):
167 class MyIter(bt2
._UserMessageIterator
):
171 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
172 def __init__(comp_self
, config
, params
, obj
):
173 comp_self
._add
_output
_port
('clear')
174 comp_self
._add
_output
_port
('print')
175 comp_self
._add
_output
_port
('insert')
177 with self
.assertRaises(KeyError):
178 comp_self
._output
_ports
['hello']
180 self
._create
_comp
(MyFilter
)
182 def test_user_flt_input_ports_getitem_invalid_key(self
):
183 class MyIter(bt2
._UserMessageIterator
):
187 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
188 def __init__(comp_self
, config
, params
, obj
):
189 comp_self
._add
_input
_port
('clear')
190 comp_self
._add
_input
_port
('print')
191 comp_self
._add
_input
_port
('insert')
193 with self
.assertRaises(KeyError):
194 comp_self
._input
_ports
['hello']
196 self
._create
_comp
(MyFilter
)
198 def test_user_sink_input_ports_getitem_invalid_key(self
):
199 class MySink(bt2
._UserSinkComponent
):
200 def __init__(comp_self
, config
, params
, obj
):
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 def _user_consume(self
):
211 self
._create
_comp
(MySink
)
213 def test_user_src_output_ports_len(self
):
214 class MyIter(bt2
._UserMessageIterator
):
218 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
219 def __init__(comp_self
, config
, params
, obj
):
220 comp_self
._add
_output
_port
('clear')
221 comp_self
._add
_output
_port
('print')
222 comp_self
._add
_output
_port
('insert')
223 self
.assertEqual(len(comp_self
._output
_ports
), 3)
225 self
._create
_comp
(MySource
)
227 def test_user_flt_output_ports_len(self
):
228 class MyIter(bt2
._UserMessageIterator
):
232 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
233 def __init__(comp_self
, config
, params
, obj
):
234 comp_self
._add
_output
_port
('clear')
235 comp_self
._add
_output
_port
('print')
236 comp_self
._add
_output
_port
('insert')
237 self
.assertEqual(len(comp_self
._output
_ports
), 3)
239 self
._create
_comp
(MyFilter
)
241 def test_user_flt_input_ports_len(self
):
242 class MyIter(bt2
._UserMessageIterator
):
246 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
247 def __init__(comp_self
, config
, params
, obj
):
248 comp_self
._add
_input
_port
('clear')
249 comp_self
._add
_input
_port
('print')
250 comp_self
._add
_input
_port
('insert')
251 self
.assertEqual(len(comp_self
._input
_ports
), 3)
253 self
._create
_comp
(MyFilter
)
255 def test_user_sink_input_ports_len(self
):
256 class MySink(bt2
._UserSinkComponent
):
257 def __init__(comp_self
, config
, params
, obj
):
258 comp_self
._add
_input
_port
('clear')
259 comp_self
._add
_input
_port
('print')
260 comp_self
._add
_input
_port
('insert')
261 self
.assertEqual(len(comp_self
._input
_ports
), 3)
263 def _user_consume(self
):
266 self
._create
_comp
(MySink
)
268 def test_user_src_output_ports_iter(self
):
269 class MyIter(bt2
._UserMessageIterator
):
273 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
274 def __init__(comp_self
, config
, params
, obj
):
275 port1
= comp_self
._add
_output
_port
('clear')
276 port2
= comp_self
._add
_output
_port
('print')
277 port3
= comp_self
._add
_output
_port
('insert')
280 for port_name
, port
in comp_self
._output
_ports
.items():
281 ports
.append((port_name
, port
))
283 self
.assertEqual(ports
[0][0], 'clear')
284 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
285 self
.assertEqual(ports
[1][0], 'print')
286 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
287 self
.assertEqual(ports
[2][0], 'insert')
288 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
290 self
._create
_comp
(MySource
)
292 def test_user_flt_output_ports_iter(self
):
293 class MyIter(bt2
._UserMessageIterator
):
297 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
298 def __init__(comp_self
, config
, params
, obj
):
299 port1
= comp_self
._add
_output
_port
('clear')
300 port2
= comp_self
._add
_output
_port
('print')
301 port3
= comp_self
._add
_output
_port
('insert')
304 for port_name
, port
in comp_self
._output
_ports
.items():
305 ports
.append((port_name
, port
))
307 self
.assertEqual(ports
[0][0], 'clear')
308 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
309 self
.assertEqual(ports
[1][0], 'print')
310 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
311 self
.assertEqual(ports
[2][0], 'insert')
312 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
314 self
._create
_comp
(MyFilter
)
316 def test_user_flt_input_ports_iter(self
):
317 class MyIter(bt2
._UserMessageIterator
):
321 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
322 def __init__(comp_self
, config
, params
, obj
):
323 port1
= comp_self
._add
_input
_port
('clear')
324 port2
= comp_self
._add
_input
_port
('print')
325 port3
= comp_self
._add
_input
_port
('insert')
328 for port_name
, port
in comp_self
._input
_ports
.items():
329 ports
.append((port_name
, port
))
331 self
.assertEqual(ports
[0][0], 'clear')
332 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
333 self
.assertEqual(ports
[1][0], 'print')
334 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
335 self
.assertEqual(ports
[2][0], 'insert')
336 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
338 self
._create
_comp
(MyFilter
)
340 def test_user_sink_input_ports_iter(self
):
341 class MySink(bt2
._UserSinkComponent
):
342 def __init__(comp_self
, config
, params
, obj
):
343 port1
= comp_self
._add
_input
_port
('clear')
344 port2
= comp_self
._add
_input
_port
('print')
345 port3
= comp_self
._add
_input
_port
('insert')
348 for port_name
, port
in comp_self
._input
_ports
.items():
349 ports
.append((port_name
, port
))
351 self
.assertEqual(ports
[0][0], 'clear')
352 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
353 self
.assertEqual(ports
[1][0], 'print')
354 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
355 self
.assertEqual(ports
[2][0], 'insert')
356 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
358 def _user_consume(self
):
361 self
._create
_comp
(MySink
)
363 def test_gen_src_output_ports_getitem(self
):
364 class MyIter(bt2
._UserMessageIterator
):
372 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
373 def __init__(comp_self
, config
, params
, obj
):
374 nonlocal port1
, port2
, port3
375 port1
= comp_self
._add
_output
_port
('clear')
376 port2
= comp_self
._add
_output
_port
('print')
377 port3
= comp_self
._add
_output
_port
('insert')
379 comp
= self
._create
_comp
(MySource
)
380 self
.assertEqual(port3
.addr
, comp
.output_ports
['insert'].addr
)
381 self
.assertEqual(port2
.addr
, comp
.output_ports
['print'].addr
)
382 self
.assertEqual(port1
.addr
, comp
.output_ports
['clear'].addr
)
387 def test_gen_flt_output_ports_getitem(self
):
388 class MyIter(bt2
._UserMessageIterator
):
396 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
397 def __init__(comp_self
, config
, params
, obj
):
398 nonlocal port1
, port2
, port3
399 port1
= comp_self
._add
_output
_port
('clear')
400 port2
= comp_self
._add
_output
_port
('print')
401 port3
= comp_self
._add
_output
_port
('insert')
403 comp
= self
._create
_comp
(MyFilter
)
404 self
.assertEqual(port3
.addr
, comp
.output_ports
['insert'].addr
)
405 self
.assertEqual(port2
.addr
, comp
.output_ports
['print'].addr
)
406 self
.assertEqual(port1
.addr
, comp
.output_ports
['clear'].addr
)
411 def test_gen_flt_input_ports_getitem(self
):
412 class MyIter(bt2
._UserMessageIterator
):
420 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
421 def __init__(comp_self
, config
, params
, obj
):
422 nonlocal port1
, port2
, port3
423 port1
= comp_self
._add
_input
_port
('clear')
424 port2
= comp_self
._add
_input
_port
('print')
425 port3
= comp_self
._add
_input
_port
('insert')
427 comp
= self
._create
_comp
(MyFilter
)
428 self
.assertEqual(port3
.addr
, comp
.input_ports
['insert'].addr
)
429 self
.assertEqual(port2
.addr
, comp
.input_ports
['print'].addr
)
430 self
.assertEqual(port1
.addr
, comp
.input_ports
['clear'].addr
)
435 def test_gen_sink_input_ports_getitem(self
):
440 class MySink(bt2
._UserSinkComponent
):
441 def __init__(comp_self
, config
, params
, obj
):
442 nonlocal port1
, port2
, port3
443 port1
= comp_self
._add
_input
_port
('clear')
444 port2
= comp_self
._add
_input
_port
('print')
445 port3
= comp_self
._add
_input
_port
('insert')
447 def _user_consume(self
):
450 comp
= self
._create
_comp
(MySink
)
451 self
.assertEqual(port3
.addr
, comp
.input_ports
['insert'].addr
)
452 self
.assertEqual(port2
.addr
, comp
.input_ports
['print'].addr
)
453 self
.assertEqual(port1
.addr
, comp
.input_ports
['clear'].addr
)
458 def test_gen_src_output_ports_getitem_invalid_key(self
):
459 class MyIter(bt2
._UserMessageIterator
):
463 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
464 def __init__(comp_self
, config
, params
, obj
):
465 comp_self
._add
_output
_port
('clear')
466 comp_self
._add
_output
_port
('print')
467 comp_self
._add
_output
_port
('insert')
469 comp
= self
._create
_comp
(MySource
)
471 with self
.assertRaises(KeyError):
472 comp
.output_ports
['hello']
474 def test_gen_flt_output_ports_getitem_invalid_key(self
):
475 class MyIter(bt2
._UserMessageIterator
):
479 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
480 def __init__(comp_self
, config
, params
, obj
):
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
(MyFilter
)
487 with self
.assertRaises(KeyError):
488 comp
.output_ports
['hello']
490 def test_gen_flt_input_ports_getitem_invalid_key(self
):
491 class MyIter(bt2
._UserMessageIterator
):
495 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
496 def __init__(comp_self
, config
, params
, obj
):
497 comp_self
._add
_input
_port
('clear')
498 comp_self
._add
_input
_port
('print')
499 comp_self
._add
_input
_port
('insert')
501 comp
= self
._create
_comp
(MyFilter
)
503 with self
.assertRaises(KeyError):
504 comp
.input_ports
['hello']
506 def test_gen_sink_input_ports_getitem_invalid_key(self
):
507 class MySink(bt2
._UserSinkComponent
):
508 def __init__(comp_self
, config
, params
, obj
):
509 comp_self
._add
_input
_port
('clear')
510 comp_self
._add
_input
_port
('print')
511 comp_self
._add
_input
_port
('insert')
513 with self
.assertRaises(KeyError):
514 comp_self
._input
_ports
['hello']
516 def _user_consume(self
):
519 comp
= self
._create
_comp
(MySink
)
521 with self
.assertRaises(KeyError):
522 comp
.input_ports
['hello']
524 def test_gen_src_output_ports_len(self
):
525 class MyIter(bt2
._UserMessageIterator
):
529 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
530 def __init__(comp_self
, config
, params
, obj
):
531 comp_self
._add
_output
_port
('clear')
532 comp_self
._add
_output
_port
('print')
533 comp_self
._add
_output
_port
('insert')
535 comp
= self
._create
_comp
(MySource
)
536 self
.assertEqual(len(comp
.output_ports
), 3)
538 def test_gen_flt_output_ports_len(self
):
539 class MyIter(bt2
._UserMessageIterator
):
543 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
544 def __init__(comp_self
, config
, params
, obj
):
545 comp_self
._add
_output
_port
('clear')
546 comp_self
._add
_output
_port
('print')
547 comp_self
._add
_output
_port
('insert')
549 comp
= self
._create
_comp
(MyFilter
)
550 self
.assertEqual(len(comp
.output_ports
), 3)
552 def test_gen_flt_input_ports_len(self
):
553 class MyIter(bt2
._UserMessageIterator
):
557 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
558 def __init__(comp_self
, config
, params
, obj
):
559 comp_self
._add
_input
_port
('clear')
560 comp_self
._add
_input
_port
('print')
561 comp_self
._add
_input
_port
('insert')
563 comp
= self
._create
_comp
(MyFilter
)
564 self
.assertEqual(len(comp
.input_ports
), 3)
566 def test_gen_sink_input_ports_len(self
):
567 class MySink(bt2
._UserSinkComponent
):
568 def __init__(comp_self
, config
, params
, obj
):
569 comp_self
._add
_input
_port
('clear')
570 comp_self
._add
_input
_port
('print')
571 comp_self
._add
_input
_port
('insert')
573 def _user_consume(self
):
576 comp
= self
._create
_comp
(MySink
)
577 self
.assertEqual(len(comp
.input_ports
), 3)
579 def test_gen_src_output_ports_iter(self
):
580 class MyIter(bt2
._UserMessageIterator
):
588 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
589 def __init__(comp_self
, config
, params
, obj
):
590 nonlocal port1
, port2
, port3
591 port1
= comp_self
._add
_output
_port
('clear')
592 port2
= comp_self
._add
_output
_port
('print')
593 port3
= comp_self
._add
_output
_port
('insert')
595 comp
= self
._create
_comp
(MySource
)
598 for port_name
, port
in comp
.output_ports
.items():
599 ports
.append((port_name
, port
))
601 self
.assertEqual(ports
[0][0], 'clear')
602 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
603 self
.assertEqual(ports
[1][0], 'print')
604 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
605 self
.assertEqual(ports
[2][0], 'insert')
606 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
611 def test_gen_flt_output_ports_iter(self
):
612 class MyIter(bt2
._UserMessageIterator
):
620 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
621 def __init__(comp_self
, config
, params
, obj
):
622 nonlocal port1
, port2
, port3
623 port1
= comp_self
._add
_output
_port
('clear')
624 port2
= comp_self
._add
_output
_port
('print')
625 port3
= comp_self
._add
_output
_port
('insert')
627 comp
= self
._create
_comp
(MyFilter
)
630 for port_name
, port
in comp
.output_ports
.items():
631 ports
.append((port_name
, port
))
633 self
.assertEqual(ports
[0][0], 'clear')
634 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
635 self
.assertEqual(ports
[1][0], 'print')
636 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
637 self
.assertEqual(ports
[2][0], 'insert')
638 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
643 def test_gen_flt_input_ports_iter(self
):
644 class MyIter(bt2
._UserMessageIterator
):
652 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
653 def __init__(comp_self
, config
, params
, obj
):
654 nonlocal port1
, port2
, port3
655 port1
= comp_self
._add
_input
_port
('clear')
656 port2
= comp_self
._add
_input
_port
('print')
657 port3
= comp_self
._add
_input
_port
('insert')
659 comp
= self
._create
_comp
(MyFilter
)
662 for port_name
, port
in comp
.input_ports
.items():
663 ports
.append((port_name
, port
))
665 self
.assertEqual(ports
[0][0], 'clear')
666 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
667 self
.assertEqual(ports
[1][0], 'print')
668 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
669 self
.assertEqual(ports
[2][0], 'insert')
670 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
675 def test_gen_sink_input_ports_iter(self
):
680 class MySink(bt2
._UserSinkComponent
):
681 def __init__(comp_self
, config
, params
, obj
):
682 nonlocal port1
, port2
, port3
683 port1
= comp_self
._add
_input
_port
('clear')
684 port2
= comp_self
._add
_input
_port
('print')
685 port3
= comp_self
._add
_input
_port
('insert')
687 def _user_consume(self
):
690 comp
= self
._create
_comp
(MySink
)
693 for port_name
, port
in comp
.input_ports
.items():
694 ports
.append((port_name
, port
))
696 self
.assertEqual(ports
[0][0], 'clear')
697 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
698 self
.assertEqual(ports
[1][0], 'print')
699 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
700 self
.assertEqual(ports
[2][0], 'insert')
701 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
707 class MySink(bt2
._UserSinkComponent
):
708 def __init__(comp_self
, config
, params
, obj
):
709 comp_self
._add
_input
_port
('clear')
711 def _user_consume(self
):
714 comp
= self
._create
_comp
(MySink
)
715 self
.assertEqual(comp
.input_ports
['clear'].name
, 'clear')
717 def test_connection_none(self
):
718 class MySink(bt2
._UserSinkComponent
):
719 def __init__(comp_self
, config
, params
, obj
):
720 comp_self
._add
_input
_port
('clear')
722 def _user_consume(self
):
725 comp
= self
._create
_comp
(MySink
)
726 self
.assertIsNone(comp
.input_ports
['clear'].connection
)
728 def test_is_connected_false(self
):
729 class MySink(bt2
._UserSinkComponent
):
730 def __init__(comp_self
, config
, params
, obj
):
731 comp_self
._add
_input
_port
('clear')
733 def _user_consume(self
):
736 comp
= self
._create
_comp
(MySink
)
737 self
.assertFalse(comp
.input_ports
['clear'].is_connected
)
739 def test_self_name(self
):
740 class MySink(bt2
._UserSinkComponent
):
741 def __init__(comp_self
, config
, params
, obj
):
742 port
= comp_self
._add
_input
_port
('clear')
743 self
.assertEqual(port
.name
, 'clear')
745 def _user_consume(self
):
748 self
._create
_comp
(MySink
)
750 def test_self_connection_none(self
):
751 class MySink(bt2
._UserSinkComponent
):
752 def __init__(comp_self
, config
, params
, obj
):
753 port
= comp_self
._add
_input
_port
('clear')
754 self
.assertIsNone(port
.connection
)
756 def _user_consume(self
):
759 self
._create
_comp
(MySink
)
761 def test_self_is_connected_false(self
):
762 class MySink(bt2
._UserSinkComponent
):
763 def __init__(comp_self
, config
, params
, obj
):
764 port
= comp_self
._add
_input
_port
('clear')
765 self
.assertFalse(port
.is_connected
)
767 def _user_consume(self
):
770 self
._create
_comp
(MySink
)
772 def test_source_self_port_user_data(self
):
773 class MyIter(bt2
._UserMessageIterator
):
777 class MySource(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
778 def __init__(comp_self
, config
, params
, obj
):
781 p
= comp_self
._add
_output
_port
('port1')
782 user_datas
.append(p
.user_data
)
783 p
= comp_self
._add
_output
_port
('port2', 2)
784 user_datas
.append(p
.user_data
)
788 self
._create
_comp
(MySource
)
789 self
.assertEqual(user_datas
, [None, 2])
791 def test_filter_self_port_user_data(self
):
792 class MyIter(bt2
._UserMessageIterator
):
796 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
797 def __init__(comp_self
, config
, params
, obj
):
800 p
= comp_self
._add
_output
_port
('port1')
801 user_datas
.append(p
.user_data
)
802 p
= comp_self
._add
_output
_port
('port2', 'user data string')
803 user_datas
.append(p
.user_data
)
805 p
= comp_self
._add
_input
_port
('port3')
806 user_datas
.append(p
.user_data
)
807 p
= comp_self
._add
_input
_port
('port4', user_data
={'user data': 'dict'})
808 user_datas
.append(p
.user_data
)
812 self
._create
_comp
(MyFilter
)
814 user_datas
, [None, 'user data string', None, {'user data': 'dict'}]
817 def test_sink_self_port_user_data(self
):
818 class MyIter(bt2
._UserMessageIterator
):
822 class MySink(bt2
._UserFilterComponent
, message_iterator_class
=MyIter
):
823 def __init__(comp_self
, config
, params
, obj
):
826 p
= comp_self
._add
_input
_port
('port1')
827 user_datas
.append(p
.user_data
)
828 p
= comp_self
._add
_input
_port
('port2', set())
829 user_datas
.append(p
.user_data
)
833 self
._create
_comp
(MySink
)
834 self
.assertEqual(user_datas
, [None, set()])
837 if __name__
== '__main__':
This page took 0.047551 seconds and 4 git commands to generate.