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
):
36 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
38 def __init__(comp_self
, config
, params
, obj
):
39 port
= comp_self
._add
_output
_port
('out')
40 self
.assertEqual(port
.name
, 'out')
42 comp
= self
._create
_comp
(MySource
)
43 self
.assertEqual(len(comp
.output_ports
), 1)
44 self
.assertIs(type(comp
.output_ports
['out']), bt2_port
._OutputPortConst
)
46 def test_flt_add_output_port(self
):
48 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
50 def __init__(comp_self
, config
, params
, obj
):
51 port
= comp_self
._add
_output
_port
('out')
52 self
.assertEqual(port
.name
, 'out')
54 comp
= self
._create
_comp
(MyFilter
)
55 self
.assertEqual(len(comp
.output_ports
), 1)
57 def test_flt_add_input_port(self
):
59 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
61 def __init__(comp_self
, config
, params
, obj
):
62 port
= comp_self
._add
_input
_port
('in')
63 self
.assertEqual(port
.name
, 'in')
65 comp
= self
._create
_comp
(MyFilter
)
66 self
.assertEqual(len(comp
.input_ports
), 1)
67 self
.assertIs(type(comp
.input_ports
['in']), bt2_port
._InputPortConst
)
69 def test_sink_add_input_port(self
):
70 class MySink(bt2
._UserSinkComponent
):
71 def __init__(comp_self
, config
, params
, obj
):
72 port
= comp_self
._add
_input
_port
('in')
73 self
.assertEqual(port
.name
, 'in')
75 def _user_consume(self
):
78 comp
= self
._create
_comp
(MySink
)
79 self
.assertEqual(len(comp
.input_ports
), 1)
81 def test_user_src_output_ports_getitem(self
):
83 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
85 def __init__(comp_self
, config
, params
, obj
):
86 port1
= comp_self
._add
_output
_port
('clear')
87 port2
= comp_self
._add
_output
_port
('print')
88 port3
= comp_self
._add
_output
_port
('insert')
89 self
.assertEqual(port3
.addr
, comp_self
._output
_ports
['insert'].addr
)
90 self
.assertEqual(port2
.addr
, comp_self
._output
_ports
['print'].addr
)
91 self
.assertEqual(port1
.addr
, comp_self
._output
_ports
['clear'].addr
)
93 self
._create
_comp
(MySource
)
95 def test_user_flt_output_ports_getitem(self
):
97 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
99 def __init__(comp_self
, config
, params
, obj
):
100 port1
= comp_self
._add
_output
_port
('clear')
101 port2
= comp_self
._add
_output
_port
('print')
102 port3
= comp_self
._add
_output
_port
('insert')
103 self
.assertEqual(port3
.addr
, comp_self
._output
_ports
['insert'].addr
)
104 self
.assertEqual(port2
.addr
, comp_self
._output
_ports
['print'].addr
)
105 self
.assertEqual(port1
.addr
, comp_self
._output
_ports
['clear'].addr
)
107 self
._create
_comp
(MyFilter
)
109 def test_user_flt_input_ports_getitem(self
):
111 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
113 def __init__(comp_self
, config
, params
, obj
):
114 port1
= comp_self
._add
_input
_port
('clear')
115 port2
= comp_self
._add
_input
_port
('print')
116 port3
= comp_self
._add
_input
_port
('insert')
117 self
.assertEqual(port3
.addr
, comp_self
._input
_ports
['insert'].addr
)
118 self
.assertEqual(port2
.addr
, comp_self
._input
_ports
['print'].addr
)
119 self
.assertEqual(port1
.addr
, comp_self
._input
_ports
['clear'].addr
)
121 self
._create
_comp
(MyFilter
)
123 def test_user_sink_input_ports_getitem(self
):
124 class MySink(bt2
._UserSinkComponent
):
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 def _user_consume(self
):
136 self
._create
_comp
(MySink
)
138 def test_user_src_output_ports_getitem_invalid_key(self
):
140 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
142 def __init__(comp_self
, config
, params
, obj
):
143 comp_self
._add
_output
_port
('clear')
144 comp_self
._add
_output
_port
('print')
145 comp_self
._add
_output
_port
('insert')
147 with self
.assertRaises(KeyError):
148 comp_self
._output
_ports
['hello']
150 self
._create
_comp
(MySource
)
152 def test_user_flt_output_ports_getitem_invalid_key(self
):
154 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
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
(MyFilter
)
166 def test_user_flt_input_ports_getitem_invalid_key(self
):
168 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
170 def __init__(comp_self
, config
, params
, obj
):
171 comp_self
._add
_input
_port
('clear')
172 comp_self
._add
_input
_port
('print')
173 comp_self
._add
_input
_port
('insert')
175 with self
.assertRaises(KeyError):
176 comp_self
._input
_ports
['hello']
178 self
._create
_comp
(MyFilter
)
180 def test_user_sink_input_ports_getitem_invalid_key(self
):
181 class MySink(bt2
._UserSinkComponent
):
182 def __init__(comp_self
, config
, params
, obj
):
183 comp_self
._add
_input
_port
('clear')
184 comp_self
._add
_input
_port
('print')
185 comp_self
._add
_input
_port
('insert')
187 with self
.assertRaises(KeyError):
188 comp_self
._input
_ports
['hello']
190 def _user_consume(self
):
193 self
._create
_comp
(MySink
)
195 def test_user_src_output_ports_len(self
):
197 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
199 def __init__(comp_self
, config
, params
, obj
):
200 comp_self
._add
_output
_port
('clear')
201 comp_self
._add
_output
_port
('print')
202 comp_self
._add
_output
_port
('insert')
203 self
.assertEqual(len(comp_self
._output
_ports
), 3)
205 self
._create
_comp
(MySource
)
207 def test_user_flt_output_ports_len(self
):
209 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
211 def __init__(comp_self
, config
, params
, obj
):
212 comp_self
._add
_output
_port
('clear')
213 comp_self
._add
_output
_port
('print')
214 comp_self
._add
_output
_port
('insert')
215 self
.assertEqual(len(comp_self
._output
_ports
), 3)
217 self
._create
_comp
(MyFilter
)
219 def test_user_flt_input_ports_len(self
):
221 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
223 def __init__(comp_self
, config
, params
, obj
):
224 comp_self
._add
_input
_port
('clear')
225 comp_self
._add
_input
_port
('print')
226 comp_self
._add
_input
_port
('insert')
227 self
.assertEqual(len(comp_self
._input
_ports
), 3)
229 self
._create
_comp
(MyFilter
)
231 def test_user_sink_input_ports_len(self
):
232 class MySink(bt2
._UserSinkComponent
):
233 def __init__(comp_self
, config
, params
, obj
):
234 comp_self
._add
_input
_port
('clear')
235 comp_self
._add
_input
_port
('print')
236 comp_self
._add
_input
_port
('insert')
237 self
.assertEqual(len(comp_self
._input
_ports
), 3)
239 def _user_consume(self
):
242 self
._create
_comp
(MySink
)
244 def test_user_src_output_ports_iter(self
):
246 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
248 def __init__(comp_self
, config
, params
, obj
):
249 port1
= comp_self
._add
_output
_port
('clear')
250 port2
= comp_self
._add
_output
_port
('print')
251 port3
= comp_self
._add
_output
_port
('insert')
254 for port_name
, port
in comp_self
._output
_ports
.items():
255 ports
.append((port_name
, port
))
257 self
.assertEqual(ports
[0][0], 'clear')
258 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
259 self
.assertEqual(ports
[1][0], 'print')
260 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
261 self
.assertEqual(ports
[2][0], 'insert')
262 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
264 self
._create
_comp
(MySource
)
266 def test_user_flt_output_ports_iter(self
):
268 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
270 def __init__(comp_self
, config
, params
, obj
):
271 port1
= comp_self
._add
_output
_port
('clear')
272 port2
= comp_self
._add
_output
_port
('print')
273 port3
= comp_self
._add
_output
_port
('insert')
276 for port_name
, port
in comp_self
._output
_ports
.items():
277 ports
.append((port_name
, port
))
279 self
.assertEqual(ports
[0][0], 'clear')
280 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
281 self
.assertEqual(ports
[1][0], 'print')
282 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
283 self
.assertEqual(ports
[2][0], 'insert')
284 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
286 self
._create
_comp
(MyFilter
)
288 def test_user_flt_input_ports_iter(self
):
290 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
292 def __init__(comp_self
, config
, params
, obj
):
293 port1
= comp_self
._add
_input
_port
('clear')
294 port2
= comp_self
._add
_input
_port
('print')
295 port3
= comp_self
._add
_input
_port
('insert')
298 for port_name
, port
in comp_self
._input
_ports
.items():
299 ports
.append((port_name
, port
))
301 self
.assertEqual(ports
[0][0], 'clear')
302 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
303 self
.assertEqual(ports
[1][0], 'print')
304 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
305 self
.assertEqual(ports
[2][0], 'insert')
306 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
308 self
._create
_comp
(MyFilter
)
310 def test_user_sink_input_ports_iter(self
):
311 class MySink(bt2
._UserSinkComponent
):
312 def __init__(comp_self
, config
, params
, obj
):
313 port1
= comp_self
._add
_input
_port
('clear')
314 port2
= comp_self
._add
_input
_port
('print')
315 port3
= comp_self
._add
_input
_port
('insert')
318 for port_name
, port
in comp_self
._input
_ports
.items():
319 ports
.append((port_name
, port
))
321 self
.assertEqual(ports
[0][0], 'clear')
322 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
323 self
.assertEqual(ports
[1][0], 'print')
324 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
325 self
.assertEqual(ports
[2][0], 'insert')
326 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
328 def _user_consume(self
):
331 self
._create
_comp
(MySink
)
333 def test_gen_src_output_ports_getitem(self
):
339 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
341 def __init__(comp_self
, config
, params
, obj
):
342 nonlocal port1
, port2
, port3
343 port1
= comp_self
._add
_output
_port
('clear')
344 port2
= comp_self
._add
_output
_port
('print')
345 port3
= comp_self
._add
_output
_port
('insert')
347 comp
= self
._create
_comp
(MySource
)
348 self
.assertEqual(port3
.addr
, comp
.output_ports
['insert'].addr
)
349 self
.assertEqual(port2
.addr
, comp
.output_ports
['print'].addr
)
350 self
.assertEqual(port1
.addr
, comp
.output_ports
['clear'].addr
)
355 def test_gen_flt_output_ports_getitem(self
):
361 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
363 def __init__(comp_self
, config
, params
, obj
):
364 nonlocal port1
, port2
, port3
365 port1
= comp_self
._add
_output
_port
('clear')
366 port2
= comp_self
._add
_output
_port
('print')
367 port3
= comp_self
._add
_output
_port
('insert')
369 comp
= self
._create
_comp
(MyFilter
)
370 self
.assertEqual(port3
.addr
, comp
.output_ports
['insert'].addr
)
371 self
.assertEqual(port2
.addr
, comp
.output_ports
['print'].addr
)
372 self
.assertEqual(port1
.addr
, comp
.output_ports
['clear'].addr
)
377 def test_gen_flt_input_ports_getitem(self
):
383 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
385 def __init__(comp_self
, config
, params
, obj
):
386 nonlocal port1
, port2
, port3
387 port1
= comp_self
._add
_input
_port
('clear')
388 port2
= comp_self
._add
_input
_port
('print')
389 port3
= comp_self
._add
_input
_port
('insert')
391 comp
= self
._create
_comp
(MyFilter
)
392 self
.assertEqual(port3
.addr
, comp
.input_ports
['insert'].addr
)
393 self
.assertEqual(port2
.addr
, comp
.input_ports
['print'].addr
)
394 self
.assertEqual(port1
.addr
, comp
.input_ports
['clear'].addr
)
399 def test_gen_sink_input_ports_getitem(self
):
404 class MySink(bt2
._UserSinkComponent
):
405 def __init__(comp_self
, config
, params
, obj
):
406 nonlocal port1
, port2
, port3
407 port1
= comp_self
._add
_input
_port
('clear')
408 port2
= comp_self
._add
_input
_port
('print')
409 port3
= comp_self
._add
_input
_port
('insert')
411 def _user_consume(self
):
414 comp
= self
._create
_comp
(MySink
)
415 self
.assertEqual(port3
.addr
, comp
.input_ports
['insert'].addr
)
416 self
.assertEqual(port2
.addr
, comp
.input_ports
['print'].addr
)
417 self
.assertEqual(port1
.addr
, comp
.input_ports
['clear'].addr
)
422 def test_gen_src_output_ports_getitem_invalid_key(self
):
424 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
426 def __init__(comp_self
, config
, params
, obj
):
427 comp_self
._add
_output
_port
('clear')
428 comp_self
._add
_output
_port
('print')
429 comp_self
._add
_output
_port
('insert')
431 comp
= self
._create
_comp
(MySource
)
433 with self
.assertRaises(KeyError):
434 comp
.output_ports
['hello']
436 def test_gen_flt_output_ports_getitem_invalid_key(self
):
438 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
440 def __init__(comp_self
, config
, params
, obj
):
441 comp_self
._add
_output
_port
('clear')
442 comp_self
._add
_output
_port
('print')
443 comp_self
._add
_output
_port
('insert')
445 comp
= self
._create
_comp
(MyFilter
)
447 with self
.assertRaises(KeyError):
448 comp
.output_ports
['hello']
450 def test_gen_flt_input_ports_getitem_invalid_key(self
):
452 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
454 def __init__(comp_self
, config
, params
, obj
):
455 comp_self
._add
_input
_port
('clear')
456 comp_self
._add
_input
_port
('print')
457 comp_self
._add
_input
_port
('insert')
459 comp
= self
._create
_comp
(MyFilter
)
461 with self
.assertRaises(KeyError):
462 comp
.input_ports
['hello']
464 def test_gen_sink_input_ports_getitem_invalid_key(self
):
465 class MySink(bt2
._UserSinkComponent
):
466 def __init__(comp_self
, config
, params
, obj
):
467 comp_self
._add
_input
_port
('clear')
468 comp_self
._add
_input
_port
('print')
469 comp_self
._add
_input
_port
('insert')
471 with self
.assertRaises(KeyError):
472 comp_self
._input
_ports
['hello']
474 def _user_consume(self
):
477 comp
= self
._create
_comp
(MySink
)
479 with self
.assertRaises(KeyError):
480 comp
.input_ports
['hello']
482 def test_gen_src_output_ports_len(self
):
484 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
486 def __init__(comp_self
, config
, params
, obj
):
487 comp_self
._add
_output
_port
('clear')
488 comp_self
._add
_output
_port
('print')
489 comp_self
._add
_output
_port
('insert')
491 comp
= self
._create
_comp
(MySource
)
492 self
.assertEqual(len(comp
.output_ports
), 3)
494 def test_gen_flt_output_ports_len(self
):
496 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
498 def __init__(comp_self
, config
, params
, obj
):
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
(MyFilter
)
504 self
.assertEqual(len(comp
.output_ports
), 3)
506 def test_gen_flt_input_ports_len(self
):
508 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
510 def __init__(comp_self
, config
, params
, obj
):
511 comp_self
._add
_input
_port
('clear')
512 comp_self
._add
_input
_port
('print')
513 comp_self
._add
_input
_port
('insert')
515 comp
= self
._create
_comp
(MyFilter
)
516 self
.assertEqual(len(comp
.input_ports
), 3)
518 def test_gen_sink_input_ports_len(self
):
519 class MySink(bt2
._UserSinkComponent
):
520 def __init__(comp_self
, config
, params
, obj
):
521 comp_self
._add
_input
_port
('clear')
522 comp_self
._add
_input
_port
('print')
523 comp_self
._add
_input
_port
('insert')
525 def _user_consume(self
):
528 comp
= self
._create
_comp
(MySink
)
529 self
.assertEqual(len(comp
.input_ports
), 3)
531 def test_gen_src_output_ports_iter(self
):
537 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
539 def __init__(comp_self
, config
, params
, obj
):
540 nonlocal port1
, port2
, port3
541 port1
= comp_self
._add
_output
_port
('clear')
542 port2
= comp_self
._add
_output
_port
('print')
543 port3
= comp_self
._add
_output
_port
('insert')
545 comp
= self
._create
_comp
(MySource
)
548 for port_name
, port
in comp
.output_ports
.items():
549 ports
.append((port_name
, port
))
551 self
.assertEqual(ports
[0][0], 'clear')
552 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
553 self
.assertEqual(ports
[1][0], 'print')
554 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
555 self
.assertEqual(ports
[2][0], 'insert')
556 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
561 def test_gen_flt_output_ports_iter(self
):
567 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
569 def __init__(comp_self
, config
, params
, obj
):
570 nonlocal port1
, port2
, port3
571 port1
= comp_self
._add
_output
_port
('clear')
572 port2
= comp_self
._add
_output
_port
('print')
573 port3
= comp_self
._add
_output
_port
('insert')
575 comp
= self
._create
_comp
(MyFilter
)
578 for port_name
, port
in comp
.output_ports
.items():
579 ports
.append((port_name
, port
))
581 self
.assertEqual(ports
[0][0], 'clear')
582 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
583 self
.assertEqual(ports
[1][0], 'print')
584 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
585 self
.assertEqual(ports
[2][0], 'insert')
586 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
591 def test_gen_flt_input_ports_iter(self
):
597 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
599 def __init__(comp_self
, config
, params
, obj
):
600 nonlocal port1
, port2
, port3
601 port1
= comp_self
._add
_input
_port
('clear')
602 port2
= comp_self
._add
_input
_port
('print')
603 port3
= comp_self
._add
_input
_port
('insert')
605 comp
= self
._create
_comp
(MyFilter
)
608 for port_name
, port
in comp
.input_ports
.items():
609 ports
.append((port_name
, port
))
611 self
.assertEqual(ports
[0][0], 'clear')
612 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
613 self
.assertEqual(ports
[1][0], 'print')
614 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
615 self
.assertEqual(ports
[2][0], 'insert')
616 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
621 def test_gen_sink_input_ports_iter(self
):
626 class MySink(bt2
._UserSinkComponent
):
627 def __init__(comp_self
, config
, params
, obj
):
628 nonlocal port1
, port2
, port3
629 port1
= comp_self
._add
_input
_port
('clear')
630 port2
= comp_self
._add
_input
_port
('print')
631 port3
= comp_self
._add
_input
_port
('insert')
633 def _user_consume(self
):
636 comp
= self
._create
_comp
(MySink
)
639 for port_name
, port
in comp
.input_ports
.items():
640 ports
.append((port_name
, port
))
642 self
.assertEqual(ports
[0][0], 'clear')
643 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
644 self
.assertEqual(ports
[1][0], 'print')
645 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
646 self
.assertEqual(ports
[2][0], 'insert')
647 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
653 class MySink(bt2
._UserSinkComponent
):
654 def __init__(comp_self
, config
, params
, obj
):
655 comp_self
._add
_input
_port
('clear')
657 def _user_consume(self
):
660 comp
= self
._create
_comp
(MySink
)
661 self
.assertEqual(comp
.input_ports
['clear'].name
, 'clear')
663 def test_connection_none(self
):
664 class MySink(bt2
._UserSinkComponent
):
665 def __init__(comp_self
, config
, params
, obj
):
666 comp_self
._add
_input
_port
('clear')
668 def _user_consume(self
):
671 comp
= self
._create
_comp
(MySink
)
672 self
.assertIsNone(comp
.input_ports
['clear'].connection
)
674 def test_is_connected_false(self
):
675 class MySink(bt2
._UserSinkComponent
):
676 def __init__(comp_self
, config
, params
, obj
):
677 comp_self
._add
_input
_port
('clear')
679 def _user_consume(self
):
682 comp
= self
._create
_comp
(MySink
)
683 self
.assertFalse(comp
.input_ports
['clear'].is_connected
)
685 def test_self_name(self
):
686 class MySink(bt2
._UserSinkComponent
):
687 def __init__(comp_self
, config
, params
, obj
):
688 port
= comp_self
._add
_input
_port
('clear')
689 self
.assertEqual(port
.name
, 'clear')
691 def _user_consume(self
):
694 self
._create
_comp
(MySink
)
696 def test_self_connection_none(self
):
697 class MySink(bt2
._UserSinkComponent
):
698 def __init__(comp_self
, config
, params
, obj
):
699 port
= comp_self
._add
_input
_port
('clear')
700 self
.assertIsNone(port
.connection
)
702 def _user_consume(self
):
705 self
._create
_comp
(MySink
)
707 def test_self_is_connected_false(self
):
708 class MySink(bt2
._UserSinkComponent
):
709 def __init__(comp_self
, config
, params
, obj
):
710 port
= comp_self
._add
_input
_port
('clear')
711 self
.assertFalse(port
.is_connected
)
713 def _user_consume(self
):
716 self
._create
_comp
(MySink
)
718 def test_source_self_port_user_data(self
):
721 nonlocal objects_deleted
725 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
727 def __init__(comp_self
, config
, params
, obj
):
730 p
= comp_self
._add
_output
_port
('port1')
731 user_datas
.append(p
.user_data
)
732 p
= comp_self
._add
_output
_port
('port2', 2)
733 user_datas
.append(p
.user_data
)
734 p
= comp_self
._add
_output
_port
('port3', MyUserData())
735 user_datas
.append(p
.user_data
)
740 comp
= self
._create
_comp
(MySource
)
741 self
.assertEqual(len(user_datas
), 3)
742 self
.assertIs(user_datas
[0], None)
743 self
.assertEqual(user_datas
[1], 2)
744 self
.assertIs(type(user_datas
[2]), MyUserData
)
746 # Verify that the user data gets freed.
747 self
.assertEqual(objects_deleted
, 0)
750 self
.assertEqual(objects_deleted
, 1)
752 def test_filter_self_port_user_data(self
):
755 nonlocal objects_deleted
759 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
761 def __init__(comp_self
, config
, params
, obj
):
764 p
= comp_self
._add
_output
_port
('port1')
765 user_datas
.append(p
.user_data
)
766 p
= comp_self
._add
_output
_port
('port2', 'user data string')
767 user_datas
.append(p
.user_data
)
768 p
= comp_self
._add
_output
_port
('port3', MyUserData())
769 user_datas
.append(p
.user_data
)
771 p
= comp_self
._add
_input
_port
('port4')
772 user_datas
.append(p
.user_data
)
773 p
= comp_self
._add
_input
_port
('port5', user_data
={'user data': 'dict'})
774 user_datas
.append(p
.user_data
)
775 p
= comp_self
._add
_input
_port
('port6', MyUserData())
776 user_datas
.append(p
.user_data
)
781 comp
= self
._create
_comp
(MyFilter
)
782 self
.assertEqual(len(user_datas
), 6)
783 self
.assertIs(user_datas
[0], None)
784 self
.assertEqual(user_datas
[1], 'user data string')
785 self
.assertIs(type(user_datas
[2]), MyUserData
)
786 self
.assertIs(user_datas
[3], None)
787 self
.assertEqual(user_datas
[4], {'user data': 'dict'})
788 self
.assertIs(type(user_datas
[5]), MyUserData
)
790 # Verify that the user data gets freed.
791 self
.assertEqual(objects_deleted
, 0)
794 self
.assertEqual(objects_deleted
, 2)
796 def test_sink_self_port_user_data(self
):
799 nonlocal objects_deleted
802 class MySink(bt2
._UserSinkComponent
):
803 def __init__(comp_self
, config
, params
, obj
):
806 p
= comp_self
._add
_input
_port
('port1')
807 user_datas
.append(p
.user_data
)
808 p
= comp_self
._add
_input
_port
('port2', MyUserData())
809 user_datas
.append(p
.user_data
)
811 def _user_consume(self
):
817 comp
= self
._create
_comp
(MySink
)
818 self
.assertEqual(len(user_datas
), 2)
819 self
.assertIs(user_datas
[0], None)
820 self
.assertIs(type(user_datas
[1]), MyUserData
)
822 # Verify that the user data gets freed.
823 self
.assertEqual(objects_deleted
, 0)
826 self
.assertEqual(objects_deleted
, 1)
829 if __name__
== '__main__':
This page took 0.079227 seconds and 4 git commands to generate.