8be3f38cb2868ce3d691568e2a8e008797bec5c7
[babeltrace.git] / tests / bindings / python / bt2 / test_port.py
1 #
2 # Copyright (C) 2019 EfficiOS Inc.
3 #
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
7 # of the License.
8 #
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.
13 #
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.
17 #
18
19 import unittest
20 import bt2
21
22
23 class PortTestCase(unittest.TestCase):
24 @staticmethod
25 def _create_comp(comp_cls, name=None):
26 graph = bt2.Graph()
27
28 if name is None:
29 name = 'comp'
30
31 return graph.add_component(comp_cls, name)
32
33 def test_src_add_output_port(self):
34 class MyIter(bt2._UserMessageIterator):
35 def __next__(self):
36 raise bt2.Stop
37
38 class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
39 def __init__(comp_self, params):
40 port = comp_self._add_output_port('out')
41 self.assertEqual(port.name, 'out')
42
43 comp = self._create_comp(MySource)
44 self.assertEqual(len(comp.output_ports), 1)
45
46 def test_flt_add_output_port(self):
47 class MyIter(bt2._UserMessageIterator):
48 def __next__(self):
49 raise bt2.Stop
50
51 class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
52 def __init__(comp_self, params):
53 port = comp_self._add_output_port('out')
54 self.assertEqual(port.name, 'out')
55
56 comp = self._create_comp(MyFilter)
57 self.assertEqual(len(comp.output_ports), 1)
58
59 def test_flt_add_input_port(self):
60 class MyIter(bt2._UserMessageIterator):
61 def __next__(self):
62 raise bt2.Stop
63
64 class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
65 def __init__(comp_self, params):
66 port = comp_self._add_input_port('in')
67 self.assertEqual(port.name, 'in')
68
69 comp = self._create_comp(MyFilter)
70 self.assertEqual(len(comp.input_ports), 1)
71
72 def test_sink_add_input_port(self):
73 class MySink(bt2._UserSinkComponent):
74 def __init__(comp_self, params):
75 port = comp_self._add_input_port('in')
76 self.assertEqual(port.name, 'in')
77
78 def _user_consume(self):
79 pass
80
81 comp = self._create_comp(MySink)
82 self.assertEqual(len(comp.input_ports), 1)
83
84 def test_user_src_output_ports_getitem(self):
85 class MyIter(bt2._UserMessageIterator):
86 def __next__(self):
87 raise bt2.Stop
88
89 class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
90 def __init__(comp_self, params):
91 port1 = comp_self._add_output_port('clear')
92 port2 = comp_self._add_output_port('print')
93 port3 = comp_self._add_output_port('insert')
94 self.assertEqual(port3.addr, comp_self._output_ports['insert'].addr)
95 self.assertEqual(port2.addr, comp_self._output_ports['print'].addr)
96 self.assertEqual(port1.addr, comp_self._output_ports['clear'].addr)
97
98 self._create_comp(MySource)
99
100 def test_user_flt_output_ports_getitem(self):
101 class MyIter(bt2._UserMessageIterator):
102 def __next__(self):
103 raise bt2.Stop
104
105 class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
106 def __init__(comp_self, params):
107 port1 = comp_self._add_output_port('clear')
108 port2 = comp_self._add_output_port('print')
109 port3 = comp_self._add_output_port('insert')
110 self.assertEqual(port3.addr, comp_self._output_ports['insert'].addr)
111 self.assertEqual(port2.addr, comp_self._output_ports['print'].addr)
112 self.assertEqual(port1.addr, comp_self._output_ports['clear'].addr)
113
114 self._create_comp(MyFilter)
115
116 def test_user_flt_input_ports_getitem(self):
117 class MyIter(bt2._UserMessageIterator):
118 def __next__(self):
119 raise bt2.Stop
120
121 class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
122 def __init__(comp_self, params):
123 port1 = comp_self._add_input_port('clear')
124 port2 = comp_self._add_input_port('print')
125 port3 = comp_self._add_input_port('insert')
126 self.assertEqual(port3.addr, comp_self._input_ports['insert'].addr)
127 self.assertEqual(port2.addr, comp_self._input_ports['print'].addr)
128 self.assertEqual(port1.addr, comp_self._input_ports['clear'].addr)
129
130 self._create_comp(MyFilter)
131
132 def test_user_sink_input_ports_getitem(self):
133 class MySink(bt2._UserSinkComponent):
134 def __init__(comp_self, params):
135 port1 = comp_self._add_input_port('clear')
136 port2 = comp_self._add_input_port('print')
137 port3 = comp_self._add_input_port('insert')
138 self.assertEqual(port3.addr, comp_self._input_ports['insert'].addr)
139 self.assertEqual(port2.addr, comp_self._input_ports['print'].addr)
140 self.assertEqual(port1.addr, comp_self._input_ports['clear'].addr)
141
142 def _user_consume(self):
143 pass
144
145 self._create_comp(MySink)
146
147 def test_user_src_output_ports_getitem_invalid_key(self):
148 class MyIter(bt2._UserMessageIterator):
149 def __next__(self):
150 raise bt2.Stop
151
152 class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
153 def __init__(comp_self, params):
154 comp_self._add_output_port('clear')
155 comp_self._add_output_port('print')
156 comp_self._add_output_port('insert')
157
158 with self.assertRaises(KeyError):
159 comp_self._output_ports['hello']
160
161 self._create_comp(MySource)
162
163 def test_user_flt_output_ports_getitem_invalid_key(self):
164 class MyIter(bt2._UserMessageIterator):
165 def __next__(self):
166 raise bt2.Stop
167
168 class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
169 def __init__(comp_self, params):
170 comp_self._add_output_port('clear')
171 comp_self._add_output_port('print')
172 comp_self._add_output_port('insert')
173
174 with self.assertRaises(KeyError):
175 comp_self._output_ports['hello']
176
177 self._create_comp(MyFilter)
178
179 def test_user_flt_input_ports_getitem_invalid_key(self):
180 class MyIter(bt2._UserMessageIterator):
181 def __next__(self):
182 raise bt2.Stop
183
184 class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
185 def __init__(comp_self, params):
186 comp_self._add_input_port('clear')
187 comp_self._add_input_port('print')
188 comp_self._add_input_port('insert')
189
190 with self.assertRaises(KeyError):
191 comp_self._input_ports['hello']
192
193 self._create_comp(MyFilter)
194
195 def test_user_sink_input_ports_getitem_invalid_key(self):
196 class MySink(bt2._UserSinkComponent):
197 def __init__(comp_self, params):
198 comp_self._add_input_port('clear')
199 comp_self._add_input_port('print')
200 comp_self._add_input_port('insert')
201
202 with self.assertRaises(KeyError):
203 comp_self._input_ports['hello']
204
205 def _user_consume(self):
206 pass
207
208 self._create_comp(MySink)
209
210 def test_user_src_output_ports_len(self):
211 class MyIter(bt2._UserMessageIterator):
212 def __next__(self):
213 raise bt2.Stop
214
215 class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
216 def __init__(comp_self, params):
217 comp_self._add_output_port('clear')
218 comp_self._add_output_port('print')
219 comp_self._add_output_port('insert')
220 self.assertEqual(len(comp_self._output_ports), 3)
221
222 self._create_comp(MySource)
223
224 def test_user_flt_output_ports_len(self):
225 class MyIter(bt2._UserMessageIterator):
226 def __next__(self):
227 raise bt2.Stop
228
229 class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
230 def __init__(comp_self, params):
231 comp_self._add_output_port('clear')
232 comp_self._add_output_port('print')
233 comp_self._add_output_port('insert')
234 self.assertEqual(len(comp_self._output_ports), 3)
235
236 self._create_comp(MyFilter)
237
238 def test_user_flt_input_ports_len(self):
239 class MyIter(bt2._UserMessageIterator):
240 def __next__(self):
241 raise bt2.Stop
242
243 class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
244 def __init__(comp_self, params):
245 comp_self._add_input_port('clear')
246 comp_self._add_input_port('print')
247 comp_self._add_input_port('insert')
248 self.assertEqual(len(comp_self._input_ports), 3)
249
250 self._create_comp(MyFilter)
251
252 def test_user_sink_input_ports_len(self):
253 class MySink(bt2._UserSinkComponent):
254 def __init__(comp_self, params):
255 comp_self._add_input_port('clear')
256 comp_self._add_input_port('print')
257 comp_self._add_input_port('insert')
258 self.assertEqual(len(comp_self._input_ports), 3)
259
260 def _user_consume(self):
261 pass
262
263 self._create_comp(MySink)
264
265 def test_user_src_output_ports_iter(self):
266 class MyIter(bt2._UserMessageIterator):
267 def __next__(self):
268 raise bt2.Stop
269
270 class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
271 def __init__(comp_self, params):
272 port1 = comp_self._add_output_port('clear')
273 port2 = comp_self._add_output_port('print')
274 port3 = comp_self._add_output_port('insert')
275 ports = []
276
277 for port_name, port in comp_self._output_ports.items():
278 ports.append((port_name, port))
279
280 self.assertEqual(ports[0][0], 'clear')
281 self.assertEqual(ports[0][1].addr, port1.addr)
282 self.assertEqual(ports[1][0], 'print')
283 self.assertEqual(ports[1][1].addr, port2.addr)
284 self.assertEqual(ports[2][0], 'insert')
285 self.assertEqual(ports[2][1].addr, port3.addr)
286
287 self._create_comp(MySource)
288
289 def test_user_flt_output_ports_iter(self):
290 class MyIter(bt2._UserMessageIterator):
291 def __next__(self):
292 raise bt2.Stop
293
294 class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
295 def __init__(comp_self, params):
296 port1 = comp_self._add_output_port('clear')
297 port2 = comp_self._add_output_port('print')
298 port3 = comp_self._add_output_port('insert')
299 ports = []
300
301 for port_name, port in comp_self._output_ports.items():
302 ports.append((port_name, port))
303
304 self.assertEqual(ports[0][0], 'clear')
305 self.assertEqual(ports[0][1].addr, port1.addr)
306 self.assertEqual(ports[1][0], 'print')
307 self.assertEqual(ports[1][1].addr, port2.addr)
308 self.assertEqual(ports[2][0], 'insert')
309 self.assertEqual(ports[2][1].addr, port3.addr)
310
311 self._create_comp(MyFilter)
312
313 def test_user_flt_input_ports_iter(self):
314 class MyIter(bt2._UserMessageIterator):
315 def __next__(self):
316 raise bt2.Stop
317
318 class MyFilter(bt2._UserFilterComponent, message_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')
323 ports = []
324
325 for port_name, port in comp_self._input_ports.items():
326 ports.append((port_name, port))
327
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)
334
335 self._create_comp(MyFilter)
336
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')
343 ports = []
344
345 for port_name, port in comp_self._input_ports.items():
346 ports.append((port_name, port))
347
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)
354
355 def _user_consume(self):
356 pass
357
358 self._create_comp(MySink)
359
360 def test_gen_src_output_ports_getitem(self):
361 class MyIter(bt2._UserMessageIterator):
362 def __next__(self):
363 raise bt2.Stop
364
365 port1 = None
366 port2 = None
367 port3 = None
368
369 class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
370 def __init__(comp_self, params):
371 nonlocal port1, port2, port3
372 port1 = comp_self._add_output_port('clear')
373 port2 = comp_self._add_output_port('print')
374 port3 = comp_self._add_output_port('insert')
375
376 comp = self._create_comp(MySource)
377 self.assertEqual(port3.addr, comp.output_ports['insert'].addr)
378 self.assertEqual(port2.addr, comp.output_ports['print'].addr)
379 self.assertEqual(port1.addr, comp.output_ports['clear'].addr)
380 del port1
381 del port2
382 del port3
383
384 def test_gen_flt_output_ports_getitem(self):
385 class MyIter(bt2._UserMessageIterator):
386 def __next__(self):
387 raise bt2.Stop
388
389 port1 = None
390 port2 = None
391 port3 = None
392
393 class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
394 def __init__(comp_self, params):
395 nonlocal port1, port2, port3
396 port1 = comp_self._add_output_port('clear')
397 port2 = comp_self._add_output_port('print')
398 port3 = comp_self._add_output_port('insert')
399
400 comp = self._create_comp(MyFilter)
401 self.assertEqual(port3.addr, comp.output_ports['insert'].addr)
402 self.assertEqual(port2.addr, comp.output_ports['print'].addr)
403 self.assertEqual(port1.addr, comp.output_ports['clear'].addr)
404 del port1
405 del port2
406 del port3
407
408 def test_gen_flt_input_ports_getitem(self):
409 class MyIter(bt2._UserMessageIterator):
410 def __next__(self):
411 raise bt2.Stop
412
413 port1 = None
414 port2 = None
415 port3 = None
416
417 class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
418 def __init__(comp_self, params):
419 nonlocal port1, port2, port3
420 port1 = comp_self._add_input_port('clear')
421 port2 = comp_self._add_input_port('print')
422 port3 = comp_self._add_input_port('insert')
423
424 comp = self._create_comp(MyFilter)
425 self.assertEqual(port3.addr, comp.input_ports['insert'].addr)
426 self.assertEqual(port2.addr, comp.input_ports['print'].addr)
427 self.assertEqual(port1.addr, comp.input_ports['clear'].addr)
428 del port1
429 del port2
430 del port3
431
432 def test_gen_sink_input_ports_getitem(self):
433 port1 = None
434 port2 = None
435 port3 = None
436
437 class MySink(bt2._UserSinkComponent):
438 def __init__(comp_self, params):
439 nonlocal port1, port2, port3
440 port1 = comp_self._add_input_port('clear')
441 port2 = comp_self._add_input_port('print')
442 port3 = comp_self._add_input_port('insert')
443
444 def _user_consume(self):
445 pass
446
447 comp = self._create_comp(MySink)
448 self.assertEqual(port3.addr, comp.input_ports['insert'].addr)
449 self.assertEqual(port2.addr, comp.input_ports['print'].addr)
450 self.assertEqual(port1.addr, comp.input_ports['clear'].addr)
451 del port1
452 del port2
453 del port3
454
455 def test_gen_src_output_ports_getitem_invalid_key(self):
456 class MyIter(bt2._UserMessageIterator):
457 def __next__(self):
458 raise bt2.Stop
459
460 class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
461 def __init__(comp_self, params):
462 comp_self._add_output_port('clear')
463 comp_self._add_output_port('print')
464 comp_self._add_output_port('insert')
465
466 comp = self._create_comp(MySource)
467
468 with self.assertRaises(KeyError):
469 comp.output_ports['hello']
470
471 def test_gen_flt_output_ports_getitem_invalid_key(self):
472 class MyIter(bt2._UserMessageIterator):
473 def __next__(self):
474 raise bt2.Stop
475
476 class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
477 def __init__(comp_self, params):
478 comp_self._add_output_port('clear')
479 comp_self._add_output_port('print')
480 comp_self._add_output_port('insert')
481
482 comp = self._create_comp(MyFilter)
483
484 with self.assertRaises(KeyError):
485 comp.output_ports['hello']
486
487 def test_gen_flt_input_ports_getitem_invalid_key(self):
488 class MyIter(bt2._UserMessageIterator):
489 def __next__(self):
490 raise bt2.Stop
491
492 class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
493 def __init__(comp_self, params):
494 comp_self._add_input_port('clear')
495 comp_self._add_input_port('print')
496 comp_self._add_input_port('insert')
497
498 comp = self._create_comp(MyFilter)
499
500 with self.assertRaises(KeyError):
501 comp.input_ports['hello']
502
503 def test_gen_sink_input_ports_getitem_invalid_key(self):
504 class MySink(bt2._UserSinkComponent):
505 def __init__(comp_self, params):
506 comp_self._add_input_port('clear')
507 comp_self._add_input_port('print')
508 comp_self._add_input_port('insert')
509
510 with self.assertRaises(KeyError):
511 comp_self._input_ports['hello']
512
513 def _user_consume(self):
514 pass
515
516 comp = self._create_comp(MySink)
517
518 with self.assertRaises(KeyError):
519 comp.input_ports['hello']
520
521 def test_gen_src_output_ports_len(self):
522 class MyIter(bt2._UserMessageIterator):
523 def __next__(self):
524 raise bt2.Stop
525
526 class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
527 def __init__(comp_self, params):
528 comp_self._add_output_port('clear')
529 comp_self._add_output_port('print')
530 comp_self._add_output_port('insert')
531
532 comp = self._create_comp(MySource)
533 self.assertEqual(len(comp.output_ports), 3)
534
535 def test_gen_flt_output_ports_len(self):
536 class MyIter(bt2._UserMessageIterator):
537 def __next__(self):
538 raise bt2.Stop
539
540 class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
541 def __init__(comp_self, params):
542 comp_self._add_output_port('clear')
543 comp_self._add_output_port('print')
544 comp_self._add_output_port('insert')
545
546 comp = self._create_comp(MyFilter)
547 self.assertEqual(len(comp.output_ports), 3)
548
549 def test_gen_flt_input_ports_len(self):
550 class MyIter(bt2._UserMessageIterator):
551 def __next__(self):
552 raise bt2.Stop
553
554 class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
555 def __init__(comp_self, params):
556 comp_self._add_input_port('clear')
557 comp_self._add_input_port('print')
558 comp_self._add_input_port('insert')
559
560 comp = self._create_comp(MyFilter)
561 self.assertEqual(len(comp.input_ports), 3)
562
563 def test_gen_sink_input_ports_len(self):
564 class MySink(bt2._UserSinkComponent):
565 def __init__(comp_self, params):
566 comp_self._add_input_port('clear')
567 comp_self._add_input_port('print')
568 comp_self._add_input_port('insert')
569
570 def _user_consume(self):
571 pass
572
573 comp = self._create_comp(MySink)
574 self.assertEqual(len(comp.input_ports), 3)
575
576 def test_gen_src_output_ports_iter(self):
577 class MyIter(bt2._UserMessageIterator):
578 def __next__(self):
579 raise bt2.Stop
580
581 port1 = None
582 port2 = None
583 port3 = None
584
585 class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
586 def __init__(comp_self, params):
587 nonlocal port1, port2, port3
588 port1 = comp_self._add_output_port('clear')
589 port2 = comp_self._add_output_port('print')
590 port3 = comp_self._add_output_port('insert')
591
592 comp = self._create_comp(MySource)
593 ports = []
594
595 for port_name, port in comp.output_ports.items():
596 ports.append((port_name, port))
597
598 self.assertEqual(ports[0][0], 'clear')
599 self.assertEqual(ports[0][1].addr, port1.addr)
600 self.assertEqual(ports[1][0], 'print')
601 self.assertEqual(ports[1][1].addr, port2.addr)
602 self.assertEqual(ports[2][0], 'insert')
603 self.assertEqual(ports[2][1].addr, port3.addr)
604 del port1
605 del port2
606 del port3
607
608 def test_gen_flt_output_ports_iter(self):
609 class MyIter(bt2._UserMessageIterator):
610 def __next__(self):
611 raise bt2.Stop
612
613 port1 = None
614 port2 = None
615 port3 = None
616
617 class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
618 def __init__(comp_self, params):
619 nonlocal port1, port2, port3
620 port1 = comp_self._add_output_port('clear')
621 port2 = comp_self._add_output_port('print')
622 port3 = comp_self._add_output_port('insert')
623
624 comp = self._create_comp(MyFilter)
625 ports = []
626
627 for port_name, port in comp.output_ports.items():
628 ports.append((port_name, port))
629
630 self.assertEqual(ports[0][0], 'clear')
631 self.assertEqual(ports[0][1].addr, port1.addr)
632 self.assertEqual(ports[1][0], 'print')
633 self.assertEqual(ports[1][1].addr, port2.addr)
634 self.assertEqual(ports[2][0], 'insert')
635 self.assertEqual(ports[2][1].addr, port3.addr)
636 del port1
637 del port2
638 del port3
639
640 def test_gen_flt_input_ports_iter(self):
641 class MyIter(bt2._UserMessageIterator):
642 def __next__(self):
643 raise bt2.Stop
644
645 port1 = None
646 port2 = None
647 port3 = None
648
649 class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
650 def __init__(comp_self, params):
651 nonlocal port1, port2, port3
652 port1 = comp_self._add_input_port('clear')
653 port2 = comp_self._add_input_port('print')
654 port3 = comp_self._add_input_port('insert')
655
656 comp = self._create_comp(MyFilter)
657 ports = []
658
659 for port_name, port in comp.input_ports.items():
660 ports.append((port_name, port))
661
662 self.assertEqual(ports[0][0], 'clear')
663 self.assertEqual(ports[0][1].addr, port1.addr)
664 self.assertEqual(ports[1][0], 'print')
665 self.assertEqual(ports[1][1].addr, port2.addr)
666 self.assertEqual(ports[2][0], 'insert')
667 self.assertEqual(ports[2][1].addr, port3.addr)
668 del port1
669 del port2
670 del port3
671
672 def test_gen_sink_input_ports_iter(self):
673 port1 = None
674 port2 = None
675 port3 = None
676
677 class MySink(bt2._UserSinkComponent):
678 def __init__(comp_self, params):
679 nonlocal port1, port2, port3
680 port1 = comp_self._add_input_port('clear')
681 port2 = comp_self._add_input_port('print')
682 port3 = comp_self._add_input_port('insert')
683
684 def _user_consume(self):
685 pass
686
687 comp = self._create_comp(MySink)
688 ports = []
689
690 for port_name, port in comp.input_ports.items():
691 ports.append((port_name, port))
692
693 self.assertEqual(ports[0][0], 'clear')
694 self.assertEqual(ports[0][1].addr, port1.addr)
695 self.assertEqual(ports[1][0], 'print')
696 self.assertEqual(ports[1][1].addr, port2.addr)
697 self.assertEqual(ports[2][0], 'insert')
698 self.assertEqual(ports[2][1].addr, port3.addr)
699 del port1
700 del port2
701 del port3
702
703 def test_name(self):
704 class MySink(bt2._UserSinkComponent):
705 def __init__(comp_self, params):
706 comp_self._add_input_port('clear')
707
708 def _user_consume(self):
709 pass
710
711 comp = self._create_comp(MySink)
712 self.assertEqual(comp.input_ports['clear'].name, 'clear')
713
714 def test_connection_none(self):
715 class MySink(bt2._UserSinkComponent):
716 def __init__(comp_self, params):
717 comp_self._add_input_port('clear')
718
719 def _user_consume(self):
720 pass
721
722 comp = self._create_comp(MySink)
723 self.assertIsNone(comp.input_ports['clear'].connection)
724
725 def test_is_connected_false(self):
726 class MySink(bt2._UserSinkComponent):
727 def __init__(comp_self, params):
728 comp_self._add_input_port('clear')
729
730 def _user_consume(self):
731 pass
732
733 comp = self._create_comp(MySink)
734 self.assertFalse(comp.input_ports['clear'].is_connected)
735
736 def test_self_name(self):
737 class MySink(bt2._UserSinkComponent):
738 def __init__(comp_self, params):
739 port = comp_self._add_input_port('clear')
740 self.assertEqual(port.name, 'clear')
741
742 def _user_consume(self):
743 pass
744
745 self._create_comp(MySink)
746
747 def test_self_connection_none(self):
748 class MySink(bt2._UserSinkComponent):
749 def __init__(comp_self, params):
750 port = comp_self._add_input_port('clear')
751 self.assertIsNone(port.connection)
752
753 def _user_consume(self):
754 pass
755
756 self._create_comp(MySink)
757
758 def test_self_is_connected_false(self):
759 class MySink(bt2._UserSinkComponent):
760 def __init__(comp_self, params):
761 port = comp_self._add_input_port('clear')
762 self.assertFalse(port.is_connected)
763
764 def _user_consume(self):
765 pass
766
767 self._create_comp(MySink)
768
769 def test_source_self_port_user_data(self):
770 class MyIter(bt2._UserMessageIterator):
771 def __next__(self):
772 raise bt2.Stop
773
774 class MySource(bt2._UserFilterComponent, message_iterator_class=MyIter):
775 def __init__(comp_self, params):
776 nonlocal user_datas
777
778 p = comp_self._add_output_port('port1')
779 user_datas.append(p.user_data)
780 p = comp_self._add_output_port('port2', 2)
781 user_datas.append(p.user_data)
782
783 user_datas = []
784
785 comp = self._create_comp(MySource)
786 self.assertEqual(user_datas, [None, 2])
787
788 def test_filter_self_port_user_data(self):
789 class MyIter(bt2._UserMessageIterator):
790 def __next__(self):
791 raise bt2.Stop
792
793 class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
794 def __init__(comp_self, params):
795 nonlocal user_datas
796
797 p = comp_self._add_output_port('port1')
798 user_datas.append(p.user_data)
799 p = comp_self._add_output_port('port2', 'user data string')
800 user_datas.append(p.user_data)
801
802 p = comp_self._add_input_port('port3')
803 user_datas.append(p.user_data)
804 p = comp_self._add_input_port('port4', user_data={'user data': 'dict'})
805 user_datas.append(p.user_data)
806
807 user_datas = []
808
809 comp = self._create_comp(MyFilter)
810 self.assertEqual(
811 user_datas, [None, 'user data string', None, {'user data': 'dict'}]
812 )
813
814 def test_sink_self_port_user_data(self):
815 class MyIter(bt2._UserMessageIterator):
816 def __next__(self):
817 raise bt2.Stop
818
819 class MySink(bt2._UserFilterComponent, message_iterator_class=MyIter):
820 def __init__(comp_self, params):
821 nonlocal user_datas
822
823 p = comp_self._add_input_port('port1')
824 user_datas.append(p.user_data)
825 p = comp_self._add_input_port('port2', set())
826 user_datas.append(p.user_data)
827
828 user_datas = []
829
830 comp = self._create_comp(MySink)
831 self.assertEqual(user_datas, [None, set()])
This page took 0.047618 seconds and 3 git commands to generate.