Fix: log.h: missing defines of `_ERRNO()` macros to `_UNUSED()`
[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 from bt2 import port as bt2_port
22
23
24 class PortTestCase(unittest.TestCase):
25 @staticmethod
26 def _create_comp(comp_cls, name=None):
27 graph = bt2.Graph()
28
29 if name is None:
30 name = 'comp'
31
32 return graph.add_component(comp_cls, name)
33
34 def test_src_add_output_port(self):
35 class MyIter(bt2._UserMessageIterator):
36 def __next__(self):
37 raise bt2.Stop
38
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')
43
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)
47
48 def test_flt_add_output_port(self):
49 class MyIter(bt2._UserMessageIterator):
50 def __next__(self):
51 raise bt2.Stop
52
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')
57
58 comp = self._create_comp(MyFilter)
59 self.assertEqual(len(comp.output_ports), 1)
60
61 def test_flt_add_input_port(self):
62 class MyIter(bt2._UserMessageIterator):
63 def __next__(self):
64 raise bt2.Stop
65
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')
70
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)
74
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')
80
81 def _user_consume(self):
82 pass
83
84 comp = self._create_comp(MySink)
85 self.assertEqual(len(comp.input_ports), 1)
86
87 def test_user_src_output_ports_getitem(self):
88 class MyIter(bt2._UserMessageIterator):
89 def __next__(self):
90 raise bt2.Stop
91
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)
100
101 self._create_comp(MySource)
102
103 def test_user_flt_output_ports_getitem(self):
104 class MyIter(bt2._UserMessageIterator):
105 def __next__(self):
106 raise bt2.Stop
107
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)
116
117 self._create_comp(MyFilter)
118
119 def test_user_flt_input_ports_getitem(self):
120 class MyIter(bt2._UserMessageIterator):
121 def __next__(self):
122 raise bt2.Stop
123
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)
132
133 self._create_comp(MyFilter)
134
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)
144
145 def _user_consume(self):
146 pass
147
148 self._create_comp(MySink)
149
150 def test_user_src_output_ports_getitem_invalid_key(self):
151 class MyIter(bt2._UserMessageIterator):
152 def __next__(self):
153 raise bt2.Stop
154
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')
160
161 with self.assertRaises(KeyError):
162 comp_self._output_ports['hello']
163
164 self._create_comp(MySource)
165
166 def test_user_flt_output_ports_getitem_invalid_key(self):
167 class MyIter(bt2._UserMessageIterator):
168 def __next__(self):
169 raise bt2.Stop
170
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')
176
177 with self.assertRaises(KeyError):
178 comp_self._output_ports['hello']
179
180 self._create_comp(MyFilter)
181
182 def test_user_flt_input_ports_getitem_invalid_key(self):
183 class MyIter(bt2._UserMessageIterator):
184 def __next__(self):
185 raise bt2.Stop
186
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')
192
193 with self.assertRaises(KeyError):
194 comp_self._input_ports['hello']
195
196 self._create_comp(MyFilter)
197
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')
204
205 with self.assertRaises(KeyError):
206 comp_self._input_ports['hello']
207
208 def _user_consume(self):
209 pass
210
211 self._create_comp(MySink)
212
213 def test_user_src_output_ports_len(self):
214 class MyIter(bt2._UserMessageIterator):
215 def __next__(self):
216 raise bt2.Stop
217
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)
224
225 self._create_comp(MySource)
226
227 def test_user_flt_output_ports_len(self):
228 class MyIter(bt2._UserMessageIterator):
229 def __next__(self):
230 raise bt2.Stop
231
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)
238
239 self._create_comp(MyFilter)
240
241 def test_user_flt_input_ports_len(self):
242 class MyIter(bt2._UserMessageIterator):
243 def __next__(self):
244 raise bt2.Stop
245
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)
252
253 self._create_comp(MyFilter)
254
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)
262
263 def _user_consume(self):
264 pass
265
266 self._create_comp(MySink)
267
268 def test_user_src_output_ports_iter(self):
269 class MyIter(bt2._UserMessageIterator):
270 def __next__(self):
271 raise bt2.Stop
272
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')
278 ports = []
279
280 for port_name, port in comp_self._output_ports.items():
281 ports.append((port_name, port))
282
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)
289
290 self._create_comp(MySource)
291
292 def test_user_flt_output_ports_iter(self):
293 class MyIter(bt2._UserMessageIterator):
294 def __next__(self):
295 raise bt2.Stop
296
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')
302 ports = []
303
304 for port_name, port in comp_self._output_ports.items():
305 ports.append((port_name, port))
306
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)
313
314 self._create_comp(MyFilter)
315
316 def test_user_flt_input_ports_iter(self):
317 class MyIter(bt2._UserMessageIterator):
318 def __next__(self):
319 raise bt2.Stop
320
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')
326 ports = []
327
328 for port_name, port in comp_self._input_ports.items():
329 ports.append((port_name, port))
330
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)
337
338 self._create_comp(MyFilter)
339
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')
346 ports = []
347
348 for port_name, port in comp_self._input_ports.items():
349 ports.append((port_name, port))
350
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)
357
358 def _user_consume(self):
359 pass
360
361 self._create_comp(MySink)
362
363 def test_gen_src_output_ports_getitem(self):
364 class MyIter(bt2._UserMessageIterator):
365 def __next__(self):
366 raise bt2.Stop
367
368 port1 = None
369 port2 = None
370 port3 = None
371
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')
378
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)
383 del port1
384 del port2
385 del port3
386
387 def test_gen_flt_output_ports_getitem(self):
388 class MyIter(bt2._UserMessageIterator):
389 def __next__(self):
390 raise bt2.Stop
391
392 port1 = None
393 port2 = None
394 port3 = None
395
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')
402
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)
407 del port1
408 del port2
409 del port3
410
411 def test_gen_flt_input_ports_getitem(self):
412 class MyIter(bt2._UserMessageIterator):
413 def __next__(self):
414 raise bt2.Stop
415
416 port1 = None
417 port2 = None
418 port3 = None
419
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')
426
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)
431 del port1
432 del port2
433 del port3
434
435 def test_gen_sink_input_ports_getitem(self):
436 port1 = None
437 port2 = None
438 port3 = None
439
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')
446
447 def _user_consume(self):
448 pass
449
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)
454 del port1
455 del port2
456 del port3
457
458 def test_gen_src_output_ports_getitem_invalid_key(self):
459 class MyIter(bt2._UserMessageIterator):
460 def __next__(self):
461 raise bt2.Stop
462
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')
468
469 comp = self._create_comp(MySource)
470
471 with self.assertRaises(KeyError):
472 comp.output_ports['hello']
473
474 def test_gen_flt_output_ports_getitem_invalid_key(self):
475 class MyIter(bt2._UserMessageIterator):
476 def __next__(self):
477 raise bt2.Stop
478
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')
484
485 comp = self._create_comp(MyFilter)
486
487 with self.assertRaises(KeyError):
488 comp.output_ports['hello']
489
490 def test_gen_flt_input_ports_getitem_invalid_key(self):
491 class MyIter(bt2._UserMessageIterator):
492 def __next__(self):
493 raise bt2.Stop
494
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')
500
501 comp = self._create_comp(MyFilter)
502
503 with self.assertRaises(KeyError):
504 comp.input_ports['hello']
505
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')
512
513 with self.assertRaises(KeyError):
514 comp_self._input_ports['hello']
515
516 def _user_consume(self):
517 pass
518
519 comp = self._create_comp(MySink)
520
521 with self.assertRaises(KeyError):
522 comp.input_ports['hello']
523
524 def test_gen_src_output_ports_len(self):
525 class MyIter(bt2._UserMessageIterator):
526 def __next__(self):
527 raise bt2.Stop
528
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')
534
535 comp = self._create_comp(MySource)
536 self.assertEqual(len(comp.output_ports), 3)
537
538 def test_gen_flt_output_ports_len(self):
539 class MyIter(bt2._UserMessageIterator):
540 def __next__(self):
541 raise bt2.Stop
542
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')
548
549 comp = self._create_comp(MyFilter)
550 self.assertEqual(len(comp.output_ports), 3)
551
552 def test_gen_flt_input_ports_len(self):
553 class MyIter(bt2._UserMessageIterator):
554 def __next__(self):
555 raise bt2.Stop
556
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')
562
563 comp = self._create_comp(MyFilter)
564 self.assertEqual(len(comp.input_ports), 3)
565
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')
572
573 def _user_consume(self):
574 pass
575
576 comp = self._create_comp(MySink)
577 self.assertEqual(len(comp.input_ports), 3)
578
579 def test_gen_src_output_ports_iter(self):
580 class MyIter(bt2._UserMessageIterator):
581 def __next__(self):
582 raise bt2.Stop
583
584 port1 = None
585 port2 = None
586 port3 = None
587
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')
594
595 comp = self._create_comp(MySource)
596 ports = []
597
598 for port_name, port in comp.output_ports.items():
599 ports.append((port_name, port))
600
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)
607 del port1
608 del port2
609 del port3
610
611 def test_gen_flt_output_ports_iter(self):
612 class MyIter(bt2._UserMessageIterator):
613 def __next__(self):
614 raise bt2.Stop
615
616 port1 = None
617 port2 = None
618 port3 = None
619
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')
626
627 comp = self._create_comp(MyFilter)
628 ports = []
629
630 for port_name, port in comp.output_ports.items():
631 ports.append((port_name, port))
632
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)
639 del port1
640 del port2
641 del port3
642
643 def test_gen_flt_input_ports_iter(self):
644 class MyIter(bt2._UserMessageIterator):
645 def __next__(self):
646 raise bt2.Stop
647
648 port1 = None
649 port2 = None
650 port3 = None
651
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')
658
659 comp = self._create_comp(MyFilter)
660 ports = []
661
662 for port_name, port in comp.input_ports.items():
663 ports.append((port_name, port))
664
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)
671 del port1
672 del port2
673 del port3
674
675 def test_gen_sink_input_ports_iter(self):
676 port1 = None
677 port2 = None
678 port3 = None
679
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')
686
687 def _user_consume(self):
688 pass
689
690 comp = self._create_comp(MySink)
691 ports = []
692
693 for port_name, port in comp.input_ports.items():
694 ports.append((port_name, port))
695
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)
702 del port1
703 del port2
704 del port3
705
706 def test_name(self):
707 class MySink(bt2._UserSinkComponent):
708 def __init__(comp_self, config, params, obj):
709 comp_self._add_input_port('clear')
710
711 def _user_consume(self):
712 pass
713
714 comp = self._create_comp(MySink)
715 self.assertEqual(comp.input_ports['clear'].name, 'clear')
716
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')
721
722 def _user_consume(self):
723 pass
724
725 comp = self._create_comp(MySink)
726 self.assertIsNone(comp.input_ports['clear'].connection)
727
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')
732
733 def _user_consume(self):
734 pass
735
736 comp = self._create_comp(MySink)
737 self.assertFalse(comp.input_ports['clear'].is_connected)
738
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')
744
745 def _user_consume(self):
746 pass
747
748 self._create_comp(MySink)
749
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)
755
756 def _user_consume(self):
757 pass
758
759 self._create_comp(MySink)
760
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)
766
767 def _user_consume(self):
768 pass
769
770 self._create_comp(MySink)
771
772 def test_source_self_port_user_data(self):
773 class MyIter(bt2._UserMessageIterator):
774 def __next__(self):
775 raise bt2.Stop
776
777 class MySource(bt2._UserFilterComponent, message_iterator_class=MyIter):
778 def __init__(comp_self, config, params, obj):
779 nonlocal user_datas
780
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)
785
786 user_datas = []
787
788 self._create_comp(MySource)
789 self.assertEqual(user_datas, [None, 2])
790
791 def test_filter_self_port_user_data(self):
792 class MyIter(bt2._UserMessageIterator):
793 def __next__(self):
794 raise bt2.Stop
795
796 class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyIter):
797 def __init__(comp_self, config, params, obj):
798 nonlocal user_datas
799
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)
804
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)
809
810 user_datas = []
811
812 self._create_comp(MyFilter)
813 self.assertEqual(
814 user_datas, [None, 'user data string', None, {'user data': 'dict'}]
815 )
816
817 def test_sink_self_port_user_data(self):
818 class MyIter(bt2._UserMessageIterator):
819 def __next__(self):
820 raise bt2.Stop
821
822 class MySink(bt2._UserFilterComponent, message_iterator_class=MyIter):
823 def __init__(comp_self, config, params, obj):
824 nonlocal user_datas
825
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)
830
831 user_datas = []
832
833 self._create_comp(MySink)
834 self.assertEqual(user_datas, [None, set()])
835
836
837 if __name__ == '__main__':
838 unittest.main()
This page took 0.047551 seconds and 4 git commands to generate.