7 class PortTestCase(unittest
.TestCase
):
9 def _create_comp(comp_cls
, name
=None):
15 return graph
.add_component(comp_cls
, name
)
17 def test_src_add_output_port(self
):
18 class MyIter(bt2
._UserNotificationIterator
):
22 class MySource(bt2
._UserSourceComponent
,
23 notification_iterator_class
=MyIter
):
24 def __init__(comp_self
, params
):
25 port
= comp_self
._add
_output
_port
('out')
26 self
.assertEqual(port
.name
, 'out')
28 comp
= self
._create
_comp
(MySource
)
29 self
.assertEqual(len(comp
.output_ports
), 1)
32 def test_flt_add_output_port(self
):
33 class MyIter(bt2
._UserNotificationIterator
):
37 class MyFilter(bt2
._UserFilterComponent
,
38 notification_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
(MyFilter
)
44 self
.assertEqual(len(comp
.output_ports
), 1)
46 def test_flt_add_input_port(self
):
47 class MyIter(bt2
._UserNotificationIterator
):
51 class MyFilter(bt2
._UserFilterComponent
,
52 notification_iterator_class
=MyIter
):
53 def __init__(comp_self
, params
):
54 port
= comp_self
._add
_input
_port
('in')
55 self
.assertEqual(port
.name
, 'in')
57 comp
= self
._create
_comp
(MyFilter
)
58 self
.assertEqual(len(comp
.input_ports
), 1)
60 def test_sink_add_input_port(self
):
61 class MySink(bt2
._UserSinkComponent
):
62 def __init__(comp_self
, params
):
63 port
= comp_self
._add
_input
_port
('in')
64 self
.assertEqual(port
.name
, 'in')
69 comp
= self
._create
_comp
(MySink
)
70 self
.assertEqual(len(comp
.input_ports
), 1)
72 def test_user_src_output_ports_getitem(self
):
73 class MyIter(bt2
._UserNotificationIterator
):
77 class MySource(bt2
._UserSourceComponent
,
78 notification_iterator_class
=MyIter
):
79 def __init__(comp_self
, params
):
80 port1
= comp_self
._add
_output
_port
('clear')
81 port2
= comp_self
._add
_output
_port
('print')
82 port3
= comp_self
._add
_output
_port
('insert')
83 self
.assertEqual(port3
.addr
, comp_self
._output
_ports
['insert'].addr
)
84 self
.assertEqual(port2
.addr
, comp_self
._output
_ports
['print'].addr
)
85 self
.assertEqual(port1
.addr
, comp_self
._output
_ports
['clear'].addr
)
87 comp
= self
._create
_comp
(MySource
)
89 def test_user_flt_output_ports_getitem(self
):
90 class MyIter(bt2
._UserNotificationIterator
):
94 class MyFilter(bt2
._UserFilterComponent
,
95 notification_iterator_class
=MyIter
):
96 def __init__(comp_self
, params
):
97 port1
= comp_self
._add
_output
_port
('clear')
98 port2
= comp_self
._add
_output
_port
('print')
99 port3
= comp_self
._add
_output
_port
('insert')
100 self
.assertEqual(port3
.addr
, comp_self
._output
_ports
['insert'].addr
)
101 self
.assertEqual(port2
.addr
, comp_self
._output
_ports
['print'].addr
)
102 self
.assertEqual(port1
.addr
, comp_self
._output
_ports
['clear'].addr
)
104 comp
= self
._create
_comp
(MyFilter
)
106 def test_user_flt_input_ports_getitem(self
):
107 class MyIter(bt2
._UserNotificationIterator
):
111 class MyFilter(bt2
._UserFilterComponent
,
112 notification_iterator_class
=MyIter
):
113 def __init__(comp_self
, params
):
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 comp
= self
._create
_comp
(MyFilter
)
123 def test_user_sink_input_ports_getitem(self
):
124 class MySink(bt2
._UserSinkComponent
):
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
)
136 comp
= self
._create
_comp
(MySink
)
138 def test_user_src_output_ports_getitem_invalid_key(self
):
139 class MyIter(bt2
._UserNotificationIterator
):
143 class MySource(bt2
._UserSourceComponent
,
144 notification_iterator_class
=MyIter
):
145 def __init__(comp_self
, params
):
146 comp_self
._add
_output
_port
('clear')
147 comp_self
._add
_output
_port
('print')
148 comp_self
._add
_output
_port
('insert')
150 with self
.assertRaises(KeyError):
151 comp_self
._output
_ports
['hello']
153 comp
= self
._create
_comp
(MySource
)
155 def test_user_flt_output_ports_getitem_invalid_key(self
):
156 class MyIter(bt2
._UserNotificationIterator
):
160 class MyFilter(bt2
._UserFilterComponent
,
161 notification_iterator_class
=MyIter
):
162 def __init__(comp_self
, params
):
163 comp_self
._add
_output
_port
('clear')
164 comp_self
._add
_output
_port
('print')
165 comp_self
._add
_output
_port
('insert')
167 with self
.assertRaises(KeyError):
168 comp_self
._output
_ports
['hello']
170 comp
= self
._create
_comp
(MyFilter
)
172 def test_user_flt_input_ports_getitem_invalid_key(self
):
173 class MyIter(bt2
._UserNotificationIterator
):
177 class MyFilter(bt2
._UserFilterComponent
,
178 notification_iterator_class
=MyIter
):
179 def __init__(comp_self
, params
):
180 comp_self
._add
_input
_port
('clear')
181 comp_self
._add
_input
_port
('print')
182 comp_self
._add
_input
_port
('insert')
184 with self
.assertRaises(KeyError):
185 comp_self
._input
_ports
['hello']
187 comp
= self
._create
_comp
(MyFilter
)
189 def test_user_sink_input_ports_getitem_invalid_key(self
):
190 class MySink(bt2
._UserSinkComponent
):
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']
202 comp
= self
._create
_comp
(MySink
)
204 def test_user_src_output_ports_len(self
):
205 class MyIter(bt2
._UserNotificationIterator
):
209 class MySource(bt2
._UserSourceComponent
,
210 notification_iterator_class
=MyIter
):
211 def __init__(comp_self
, params
):
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 comp
= self
._create
_comp
(MySource
)
219 def test_user_flt_output_ports_len(self
):
220 class MyIter(bt2
._UserNotificationIterator
):
224 class MyFilter(bt2
._UserFilterComponent
,
225 notification_iterator_class
=MyIter
):
226 def __init__(comp_self
, params
):
227 comp_self
._add
_output
_port
('clear')
228 comp_self
._add
_output
_port
('print')
229 comp_self
._add
_output
_port
('insert')
230 self
.assertEqual(len(comp_self
._output
_ports
), 3)
232 comp
= self
._create
_comp
(MyFilter
)
234 def test_user_flt_input_ports_len(self
):
235 class MyIter(bt2
._UserNotificationIterator
):
239 class MyFilter(bt2
._UserFilterComponent
,
240 notification_iterator_class
=MyIter
):
241 def __init__(comp_self
, params
):
242 comp_self
._add
_input
_port
('clear')
243 comp_self
._add
_input
_port
('print')
244 comp_self
._add
_input
_port
('insert')
245 self
.assertEqual(len(comp_self
._input
_ports
), 3)
247 comp
= self
._create
_comp
(MyFilter
)
249 def test_user_sink_input_ports_len(self
):
250 class MySink(bt2
._UserSinkComponent
):
251 def __init__(comp_self
, params
):
252 comp_self
._add
_input
_port
('clear')
253 comp_self
._add
_input
_port
('print')
254 comp_self
._add
_input
_port
('insert')
255 self
.assertEqual(len(comp_self
._input
_ports
), 3)
260 comp
= self
._create
_comp
(MySink
)
262 def test_user_src_output_ports_iter(self
):
263 class MyIter(bt2
._UserNotificationIterator
):
267 class MySource(bt2
._UserSourceComponent
,
268 notification_iterator_class
=MyIter
):
269 def __init__(comp_self
, params
):
270 port1
= comp_self
._add
_output
_port
('clear')
271 port2
= comp_self
._add
_output
_port
('print')
272 port3
= comp_self
._add
_output
_port
('insert')
275 for port_name
, port
in comp_self
._output
_ports
.items():
276 ports
.append((port_name
, port
))
278 self
.assertEqual(ports
[0][0], 'clear')
279 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
280 self
.assertEqual(ports
[1][0], 'print')
281 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
282 self
.assertEqual(ports
[2][0], 'insert')
283 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
285 comp
= self
._create
_comp
(MySource
)
287 def test_user_flt_output_ports_iter(self
):
288 class MyIter(bt2
._UserNotificationIterator
):
292 class MyFilter(bt2
._UserFilterComponent
,
293 notification_iterator_class
=MyIter
):
294 def __init__(comp_self
, params
):
295 port1
= comp_self
._add
_output
_port
('clear')
296 port2
= comp_self
._add
_output
_port
('print')
297 port3
= comp_self
._add
_output
_port
('insert')
300 for port_name
, port
in comp_self
._output
_ports
.items():
301 ports
.append((port_name
, port
))
303 self
.assertEqual(ports
[0][0], 'clear')
304 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
305 self
.assertEqual(ports
[1][0], 'print')
306 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
307 self
.assertEqual(ports
[2][0], 'insert')
308 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
310 comp
= self
._create
_comp
(MyFilter
)
312 def test_user_flt_input_ports_iter(self
):
313 class MyIter(bt2
._UserNotificationIterator
):
317 class MyFilter(bt2
._UserFilterComponent
,
318 notification_iterator_class
=MyIter
):
319 def __init__(comp_self
, params
):
320 port1
= comp_self
._add
_input
_port
('clear')
321 port2
= comp_self
._add
_input
_port
('print')
322 port3
= comp_self
._add
_input
_port
('insert')
325 for port_name
, port
in comp_self
._input
_ports
.items():
326 ports
.append((port_name
, port
))
328 self
.assertEqual(ports
[0][0], 'clear')
329 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
330 self
.assertEqual(ports
[1][0], 'print')
331 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
332 self
.assertEqual(ports
[2][0], 'insert')
333 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
335 comp
= self
._create
_comp
(MyFilter
)
337 def test_user_sink_input_ports_iter(self
):
338 class MySink(bt2
._UserSinkComponent
):
339 def __init__(comp_self
, params
):
340 port1
= comp_self
._add
_input
_port
('clear')
341 port2
= comp_self
._add
_input
_port
('print')
342 port3
= comp_self
._add
_input
_port
('insert')
345 for port_name
, port
in comp_self
._input
_ports
.items():
346 ports
.append((port_name
, port
))
348 self
.assertEqual(ports
[0][0], 'clear')
349 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
350 self
.assertEqual(ports
[1][0], 'print')
351 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
352 self
.assertEqual(ports
[2][0], 'insert')
353 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
358 comp
= self
._create
_comp
(MySink
)
360 def test_gen_src_output_ports_getitem(self
):
361 class MyIter(bt2
._UserNotificationIterator
):
369 class MySource(bt2
._UserSourceComponent
,
370 notification_iterator_class
=MyIter
):
371 def __init__(comp_self
, params
):
372 nonlocal port1
, port2
, port3
373 port1
= comp_self
._add
_output
_port
('clear')
374 port2
= comp_self
._add
_output
_port
('print')
375 port3
= comp_self
._add
_output
_port
('insert')
377 comp
= self
._create
_comp
(MySource
)
378 self
.assertEqual(port3
.addr
, comp
.output_ports
['insert'].addr
)
379 self
.assertEqual(port2
.addr
, comp
.output_ports
['print'].addr
)
380 self
.assertEqual(port1
.addr
, comp
.output_ports
['clear'].addr
)
385 def test_gen_flt_output_ports_getitem(self
):
386 class MyIter(bt2
._UserNotificationIterator
):
394 class MyFilter(bt2
._UserFilterComponent
,
395 notification_iterator_class
=MyIter
):
396 def __init__(comp_self
, params
):
397 nonlocal port1
, port2
, port3
398 port1
= comp_self
._add
_output
_port
('clear')
399 port2
= comp_self
._add
_output
_port
('print')
400 port3
= comp_self
._add
_output
_port
('insert')
402 comp
= self
._create
_comp
(MyFilter
)
403 self
.assertEqual(port3
.addr
, comp
.output_ports
['insert'].addr
)
404 self
.assertEqual(port2
.addr
, comp
.output_ports
['print'].addr
)
405 self
.assertEqual(port1
.addr
, comp
.output_ports
['clear'].addr
)
410 def test_gen_flt_input_ports_getitem(self
):
411 class MyIter(bt2
._UserNotificationIterator
):
419 class MyFilter(bt2
._UserFilterComponent
,
420 notification_iterator_class
=MyIter
):
421 def __init__(comp_self
, params
):
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
, params
):
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')
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
._UserNotificationIterator
):
463 class MySource(bt2
._UserSourceComponent
,
464 notification_iterator_class
=MyIter
):
465 def __init__(comp_self
, params
):
466 comp_self
._add
_output
_port
('clear')
467 comp_self
._add
_output
_port
('print')
468 comp_self
._add
_output
_port
('insert')
470 comp
= self
._create
_comp
(MySource
)
472 with self
.assertRaises(KeyError):
473 comp
.output_ports
['hello']
475 def test_gen_flt_output_ports_getitem_invalid_key(self
):
476 class MyIter(bt2
._UserNotificationIterator
):
480 class MyFilter(bt2
._UserFilterComponent
,
481 notification_iterator_class
=MyIter
):
482 def __init__(comp_self
, params
):
483 comp_self
._add
_output
_port
('clear')
484 comp_self
._add
_output
_port
('print')
485 comp_self
._add
_output
_port
('insert')
487 comp
= self
._create
_comp
(MyFilter
)
489 with self
.assertRaises(KeyError):
490 comp
.output_ports
['hello']
492 def test_gen_flt_input_ports_getitem_invalid_key(self
):
493 class MyIter(bt2
._UserNotificationIterator
):
497 class MyFilter(bt2
._UserFilterComponent
,
498 notification_iterator_class
=MyIter
):
499 def __init__(comp_self
, params
):
500 comp_self
._add
_input
_port
('clear')
501 comp_self
._add
_input
_port
('print')
502 comp_self
._add
_input
_port
('insert')
504 comp
= self
._create
_comp
(MyFilter
)
506 with self
.assertRaises(KeyError):
507 comp
.input_ports
['hello']
509 def test_gen_sink_input_ports_getitem_invalid_key(self
):
510 class MySink(bt2
._UserSinkComponent
):
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 with self
.assertRaises(KeyError):
517 comp_self
._input
_ports
['hello']
522 comp
= self
._create
_comp
(MySink
)
524 with self
.assertRaises(KeyError):
525 comp
.input_ports
['hello']
527 def test_gen_src_output_ports_len(self
):
528 class MyIter(bt2
._UserNotificationIterator
):
532 class MySource(bt2
._UserSourceComponent
,
533 notification_iterator_class
=MyIter
):
534 def __init__(comp_self
, params
):
535 comp_self
._add
_output
_port
('clear')
536 comp_self
._add
_output
_port
('print')
537 comp_self
._add
_output
_port
('insert')
539 comp
= self
._create
_comp
(MySource
)
540 self
.assertEqual(len(comp
.output_ports
), 3)
542 def test_gen_flt_output_ports_len(self
):
543 class MyIter(bt2
._UserNotificationIterator
):
547 class MyFilter(bt2
._UserFilterComponent
,
548 notification_iterator_class
=MyIter
):
549 def __init__(comp_self
, params
):
550 comp_self
._add
_output
_port
('clear')
551 comp_self
._add
_output
_port
('print')
552 comp_self
._add
_output
_port
('insert')
554 comp
= self
._create
_comp
(MyFilter
)
555 self
.assertEqual(len(comp
.output_ports
), 3)
557 def test_gen_flt_input_ports_len(self
):
558 class MyIter(bt2
._UserNotificationIterator
):
562 class MyFilter(bt2
._UserFilterComponent
,
563 notification_iterator_class
=MyIter
):
564 def __init__(comp_self
, params
):
565 comp_self
._add
_input
_port
('clear')
566 comp_self
._add
_input
_port
('print')
567 comp_self
._add
_input
_port
('insert')
569 comp
= self
._create
_comp
(MyFilter
)
570 self
.assertEqual(len(comp
.input_ports
), 3)
572 def test_gen_sink_input_ports_len(self
):
573 class MySink(bt2
._UserSinkComponent
):
574 def __init__(comp_self
, params
):
575 comp_self
._add
_input
_port
('clear')
576 comp_self
._add
_input
_port
('print')
577 comp_self
._add
_input
_port
('insert')
582 comp
= self
._create
_comp
(MySink
)
583 self
.assertEqual(len(comp
.input_ports
), 3)
585 def test_gen_src_output_ports_iter(self
):
586 class MyIter(bt2
._UserNotificationIterator
):
594 class MySource(bt2
._UserSourceComponent
,
595 notification_iterator_class
=MyIter
):
596 def __init__(comp_self
, params
):
597 nonlocal port1
, port2
, port3
598 port1
= comp_self
._add
_output
_port
('clear')
599 port2
= comp_self
._add
_output
_port
('print')
600 port3
= comp_self
._add
_output
_port
('insert')
602 comp
= self
._create
_comp
(MySource
)
605 for port_name
, port
in comp
.output_ports
.items():
606 ports
.append((port_name
, port
))
608 self
.assertEqual(ports
[0][0], 'clear')
609 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
610 self
.assertEqual(ports
[1][0], 'print')
611 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
612 self
.assertEqual(ports
[2][0], 'insert')
613 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
618 def test_gen_flt_output_ports_iter(self
):
619 class MyIter(bt2
._UserNotificationIterator
):
627 class MyFilter(bt2
._UserFilterComponent
,
628 notification_iterator_class
=MyIter
):
629 def __init__(comp_self
, params
):
630 nonlocal port1
, port2
, port3
631 port1
= comp_self
._add
_output
_port
('clear')
632 port2
= comp_self
._add
_output
_port
('print')
633 port3
= comp_self
._add
_output
_port
('insert')
635 comp
= self
._create
_comp
(MyFilter
)
638 for port_name
, port
in comp
.output_ports
.items():
639 ports
.append((port_name
, port
))
641 self
.assertEqual(ports
[0][0], 'clear')
642 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
643 self
.assertEqual(ports
[1][0], 'print')
644 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
645 self
.assertEqual(ports
[2][0], 'insert')
646 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
651 def test_gen_flt_input_ports_iter(self
):
652 class MyIter(bt2
._UserNotificationIterator
):
660 class MyFilter(bt2
._UserFilterComponent
,
661 notification_iterator_class
=MyIter
):
662 def __init__(comp_self
, params
):
663 nonlocal port1
, port2
, port3
664 port1
= comp_self
._add
_input
_port
('clear')
665 port2
= comp_self
._add
_input
_port
('print')
666 port3
= comp_self
._add
_input
_port
('insert')
668 comp
= self
._create
_comp
(MyFilter
)
671 for port_name
, port
in comp
.input_ports
.items():
672 ports
.append((port_name
, port
))
674 self
.assertEqual(ports
[0][0], 'clear')
675 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
676 self
.assertEqual(ports
[1][0], 'print')
677 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
678 self
.assertEqual(ports
[2][0], 'insert')
679 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
684 def test_gen_sink_input_ports_iter(self
):
689 class MySink(bt2
._UserSinkComponent
):
690 def __init__(comp_self
, params
):
691 nonlocal port1
, port2
, port3
692 port1
= comp_self
._add
_input
_port
('clear')
693 port2
= comp_self
._add
_input
_port
('print')
694 port3
= comp_self
._add
_input
_port
('insert')
699 comp
= self
._create
_comp
(MySink
)
702 for port_name
, port
in comp
.input_ports
.items():
703 ports
.append((port_name
, port
))
705 self
.assertEqual(ports
[0][0], 'clear')
706 self
.assertEqual(ports
[0][1].addr
, port1
.addr
)
707 self
.assertEqual(ports
[1][0], 'print')
708 self
.assertEqual(ports
[1][1].addr
, port2
.addr
)
709 self
.assertEqual(ports
[2][0], 'insert')
710 self
.assertEqual(ports
[2][1].addr
, port3
.addr
)
716 class MySink(bt2
._UserSinkComponent
):
717 def __init__(comp_self
, params
):
718 comp_self
._add
_input
_port
('clear')
723 comp
= self
._create
_comp
(MySink
)
724 self
.assertEqual(comp
.input_ports
['clear'].name
, 'clear')
726 def test_component(self
):
727 class MySink(bt2
._UserSinkComponent
):
728 def __init__(comp_self
, params
):
729 comp_self
._add
_input
_port
('clear')
734 comp
= self
._create
_comp
(MySink
)
735 self
.assertEqual(comp
.input_ports
['clear'].component
.addr
, comp
.addr
)
737 def test_connection_none(self
):
738 class MySink(bt2
._UserSinkComponent
):
739 def __init__(comp_self
, params
):
740 comp_self
._add
_input
_port
('clear')
745 comp
= self
._create
_comp
(MySink
)
746 self
.assertIsNone(comp
.input_ports
['clear'].connection
)
748 def test_is_connected_false(self
):
749 class MySink(bt2
._UserSinkComponent
):
750 def __init__(comp_self
, params
):
751 comp_self
._add
_input
_port
('clear')
756 comp
= self
._create
_comp
(MySink
)
757 self
.assertFalse(comp
.input_ports
['clear'].is_connected
)
760 class MySink(bt2
._UserSinkComponent
):
761 def __init__(comp_self
, params
):
762 comp_self
._add
_input
_port
('clear')
767 comp
= self
._create
_comp
(MySink
)
768 self
.assertEqual(comp
.input_ports
['clear'],
769 comp
.input_ports
['clear'])
771 def test_eq_invalid(self
):
772 class MySink(bt2
._UserSinkComponent
):
773 def __init__(comp_self
, params
):
774 comp_self
._add
_input
_port
('clear')
779 comp
= self
._create
_comp
(MySink
)
780 self
.assertNotEqual(comp
.input_ports
['clear'], 23)
782 def test_disconnect_no_connection(self
):
783 class MySink(bt2
._UserSinkComponent
):
784 def __init__(comp_self
, params
):
785 port
= comp_self
._add
_input
_port
('clear')
790 comp
= self
._create
_comp
(MySink
)
791 comp
.input_ports
['clear'].disconnect()
793 def test_priv_name(self
):
794 class MySink(bt2
._UserSinkComponent
):
795 def __init__(comp_self
, params
):
796 port
= comp_self
._add
_input
_port
('clear')
797 self
.assertEqual(port
.name
, 'clear')
802 comp
= self
._create
_comp
(MySink
)
804 def test_priv_component(self
):
805 class MySink(bt2
._UserSinkComponent
):
806 def __init__(comp_self
, params
):
807 port
= comp_self
._add
_input
_port
('clear')
808 self
.assertEqual(port
.component
, comp_self
)
813 comp
= self
._create
_comp
(MySink
)
815 def test_priv_connection_none(self
):
816 class MySink(bt2
._UserSinkComponent
):
817 def __init__(comp_self
, params
):
818 port
= comp_self
._add
_input
_port
('clear')
819 self
.assertIsNone(port
.connection
)
824 comp
= self
._create
_comp
(MySink
)
826 def test_priv_is_connected_false(self
):
827 class MySink(bt2
._UserSinkComponent
):
828 def __init__(comp_self
, params
):
829 port
= comp_self
._add
_input
_port
('clear')
830 self
.assertFalse(port
.is_connected
)
835 comp
= self
._create
_comp
(MySink
)
837 def test_priv_eq(self
):
838 class MySink(bt2
._UserSinkComponent
):
839 def __init__(comp_self
, params
):
840 port
= comp_self
._add
_input
_port
('clear')
841 self
.assertEqual(port
, port
)
846 comp
= self
._create
_comp
(MySink
)
848 def test_priv_eq_invalid(self
):
849 class MySink(bt2
._UserSinkComponent
):
850 def __init__(comp_self
, params
):
851 port
= comp_self
._add
_input
_port
('clear')
852 self
.assertNotEqual(port
, 23)
857 comp
= self
._create
_comp
(MySink
)
859 def test_priv_disconnect_no_connection(self
):
860 class MySink(bt2
._UserSinkComponent
):
861 def __init__(comp_self
, params
):
862 port
= comp_self
._add
_input
_port
('clear')
868 comp
= self
._create
_comp
(MySink
)
870 def test_priv_remove_from_component(self
):
871 class MySink(bt2
._UserSinkComponent
):
872 def __init__(comp_self
, params
):
873 port
= comp_self
._add
_input
_port
('clear')
874 self
.assertEqual(len(comp_self
._input
_ports
), 1)
877 port
.remove_from_component()
880 traceback
.print_exc()
882 self
.assertEqual(len(comp_self
._input
_ports
), 0)
883 self
.assertIsNone(port
.component
)
888 comp
= self
._create
_comp
(MySink
)