build: try calling python-config with --embed
[babeltrace.git] / tests / bindings / python / bt2 / test_port.py
CommitLineData
d2d857a8
MJ
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
811644b8 19import unittest
811644b8 20import bt2
5813b3a3 21from bt2 import port as bt2_port
811644b8
PP
22
23
24class 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):
5c61fb9d
SM
35 class MySource(
36 bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
37 ):
59225a3e 38 def __init__(comp_self, config, params, obj):
811644b8
PP
39 port = comp_self._add_output_port('out')
40 self.assertEqual(port.name, 'out')
41
42 comp = self._create_comp(MySource)
43 self.assertEqual(len(comp.output_ports), 1)
5813b3a3 44 self.assertIs(type(comp.output_ports['out']), bt2_port._OutputPortConst)
811644b8 45
811644b8 46 def test_flt_add_output_port(self):
5c61fb9d
SM
47 class MyFilter(
48 bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
49 ):
59225a3e 50 def __init__(comp_self, config, params, obj):
811644b8
PP
51 port = comp_self._add_output_port('out')
52 self.assertEqual(port.name, 'out')
53
54 comp = self._create_comp(MyFilter)
55 self.assertEqual(len(comp.output_ports), 1)
56
57 def test_flt_add_input_port(self):
5c61fb9d
SM
58 class MyFilter(
59 bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
60 ):
59225a3e 61 def __init__(comp_self, config, params, obj):
811644b8
PP
62 port = comp_self._add_input_port('in')
63 self.assertEqual(port.name, 'in')
64
65 comp = self._create_comp(MyFilter)
66 self.assertEqual(len(comp.input_ports), 1)
5813b3a3 67 self.assertIs(type(comp.input_ports['in']), bt2_port._InputPortConst)
811644b8
PP
68
69 def test_sink_add_input_port(self):
70 class MySink(bt2._UserSinkComponent):
59225a3e 71 def __init__(comp_self, config, params, obj):
811644b8
PP
72 port = comp_self._add_input_port('in')
73 self.assertEqual(port.name, 'in')
74
6a91742b 75 def _user_consume(self):
a01b452b
SM
76 pass
77
811644b8
PP
78 comp = self._create_comp(MySink)
79 self.assertEqual(len(comp.input_ports), 1)
80
81 def test_user_src_output_ports_getitem(self):
5c61fb9d
SM
82 class MySource(
83 bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
84 ):
59225a3e 85 def __init__(comp_self, config, params, obj):
811644b8
PP
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)
92
894a8df5 93 self._create_comp(MySource)
811644b8
PP
94
95 def test_user_flt_output_ports_getitem(self):
5c61fb9d
SM
96 class MyFilter(
97 bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
98 ):
59225a3e 99 def __init__(comp_self, config, params, obj):
811644b8
PP
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)
106
894a8df5 107 self._create_comp(MyFilter)
811644b8
PP
108
109 def test_user_flt_input_ports_getitem(self):
5c61fb9d
SM
110 class MyFilter(
111 bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
112 ):
59225a3e 113 def __init__(comp_self, config, params, obj):
811644b8
PP
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)
120
894a8df5 121 self._create_comp(MyFilter)
811644b8
PP
122
123 def test_user_sink_input_ports_getitem(self):
124 class MySink(bt2._UserSinkComponent):
59225a3e 125 def __init__(comp_self, config, params, obj):
811644b8
PP
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
6a91742b 133 def _user_consume(self):
a01b452b
SM
134 pass
135
894a8df5 136 self._create_comp(MySink)
811644b8
PP
137
138 def test_user_src_output_ports_getitem_invalid_key(self):
5c61fb9d
SM
139 class MySource(
140 bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
141 ):
59225a3e 142 def __init__(comp_self, config, params, obj):
811644b8
PP
143 comp_self._add_output_port('clear')
144 comp_self._add_output_port('print')
145 comp_self._add_output_port('insert')
146
147 with self.assertRaises(KeyError):
148 comp_self._output_ports['hello']
149
894a8df5 150 self._create_comp(MySource)
811644b8
PP
151
152 def test_user_flt_output_ports_getitem_invalid_key(self):
5c61fb9d
SM
153 class MyFilter(
154 bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
155 ):
59225a3e 156 def __init__(comp_self, config, params, obj):
811644b8
PP
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
894a8df5 164 self._create_comp(MyFilter)
811644b8
PP
165
166 def test_user_flt_input_ports_getitem_invalid_key(self):
5c61fb9d
SM
167 class MyFilter(
168 bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
169 ):
59225a3e 170 def __init__(comp_self, config, params, obj):
811644b8
PP
171 comp_self._add_input_port('clear')
172 comp_self._add_input_port('print')
173 comp_self._add_input_port('insert')
174
175 with self.assertRaises(KeyError):
176 comp_self._input_ports['hello']
177
894a8df5 178 self._create_comp(MyFilter)
811644b8
PP
179
180 def test_user_sink_input_ports_getitem_invalid_key(self):
181 class MySink(bt2._UserSinkComponent):
59225a3e 182 def __init__(comp_self, config, params, obj):
811644b8
PP
183 comp_self._add_input_port('clear')
184 comp_self._add_input_port('print')
185 comp_self._add_input_port('insert')
186
187 with self.assertRaises(KeyError):
188 comp_self._input_ports['hello']
189
6a91742b 190 def _user_consume(self):
a01b452b
SM
191 pass
192
894a8df5 193 self._create_comp(MySink)
811644b8
PP
194
195 def test_user_src_output_ports_len(self):
5c61fb9d
SM
196 class MySource(
197 bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
198 ):
59225a3e 199 def __init__(comp_self, config, params, obj):
811644b8
PP
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)
204
894a8df5 205 self._create_comp(MySource)
811644b8
PP
206
207 def test_user_flt_output_ports_len(self):
5c61fb9d
SM
208 class MyFilter(
209 bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
210 ):
59225a3e 211 def __init__(comp_self, config, params, obj):
811644b8
PP
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)
216
894a8df5 217 self._create_comp(MyFilter)
811644b8
PP
218
219 def test_user_flt_input_ports_len(self):
5c61fb9d
SM
220 class MyFilter(
221 bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
222 ):
59225a3e 223 def __init__(comp_self, config, params, obj):
811644b8
PP
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)
228
894a8df5 229 self._create_comp(MyFilter)
811644b8
PP
230
231 def test_user_sink_input_ports_len(self):
232 class MySink(bt2._UserSinkComponent):
59225a3e 233 def __init__(comp_self, config, params, obj):
811644b8
PP
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)
238
6a91742b 239 def _user_consume(self):
a01b452b
SM
240 pass
241
894a8df5 242 self._create_comp(MySink)
811644b8
PP
243
244 def test_user_src_output_ports_iter(self):
5c61fb9d
SM
245 class MySource(
246 bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
247 ):
59225a3e 248 def __init__(comp_self, config, params, obj):
811644b8
PP
249 port1 = comp_self._add_output_port('clear')
250 port2 = comp_self._add_output_port('print')
251 port3 = comp_self._add_output_port('insert')
252 ports = []
253
254 for port_name, port in comp_self._output_ports.items():
255 ports.append((port_name, port))
256
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)
263
894a8df5 264 self._create_comp(MySource)
811644b8
PP
265
266 def test_user_flt_output_ports_iter(self):
5c61fb9d
SM
267 class MyFilter(
268 bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
269 ):
59225a3e 270 def __init__(comp_self, config, params, obj):
811644b8
PP
271 port1 = comp_self._add_output_port('clear')
272 port2 = comp_self._add_output_port('print')
273 port3 = comp_self._add_output_port('insert')
274 ports = []
275
276 for port_name, port in comp_self._output_ports.items():
277 ports.append((port_name, port))
278
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)
285
894a8df5 286 self._create_comp(MyFilter)
811644b8
PP
287
288 def test_user_flt_input_ports_iter(self):
5c61fb9d
SM
289 class MyFilter(
290 bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
291 ):
59225a3e 292 def __init__(comp_self, config, params, obj):
811644b8
PP
293 port1 = comp_self._add_input_port('clear')
294 port2 = comp_self._add_input_port('print')
295 port3 = comp_self._add_input_port('insert')
296 ports = []
297
298 for port_name, port in comp_self._input_ports.items():
299 ports.append((port_name, port))
300
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)
307
894a8df5 308 self._create_comp(MyFilter)
811644b8
PP
309
310 def test_user_sink_input_ports_iter(self):
311 class MySink(bt2._UserSinkComponent):
59225a3e 312 def __init__(comp_self, config, params, obj):
811644b8
PP
313 port1 = comp_self._add_input_port('clear')
314 port2 = comp_self._add_input_port('print')
315 port3 = comp_self._add_input_port('insert')
316 ports = []
317
318 for port_name, port in comp_self._input_ports.items():
319 ports.append((port_name, port))
320
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)
327
6a91742b 328 def _user_consume(self):
a01b452b
SM
329 pass
330
894a8df5 331 self._create_comp(MySink)
811644b8
PP
332
333 def test_gen_src_output_ports_getitem(self):
811644b8
PP
334 port1 = None
335 port2 = None
336 port3 = None
337
5c61fb9d
SM
338 class MySource(
339 bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
340 ):
59225a3e 341 def __init__(comp_self, config, params, obj):
811644b8
PP
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')
346
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)
351 del port1
352 del port2
353 del port3
354
355 def test_gen_flt_output_ports_getitem(self):
811644b8
PP
356 port1 = None
357 port2 = None
358 port3 = None
359
5c61fb9d
SM
360 class MyFilter(
361 bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
362 ):
59225a3e 363 def __init__(comp_self, config, params, obj):
811644b8
PP
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')
368
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)
373 del port1
374 del port2
375 del port3
376
377 def test_gen_flt_input_ports_getitem(self):
811644b8
PP
378 port1 = None
379 port2 = None
380 port3 = None
381
5c61fb9d
SM
382 class MyFilter(
383 bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
384 ):
59225a3e 385 def __init__(comp_self, config, params, obj):
811644b8
PP
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')
390
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)
395 del port1
396 del port2
397 del port3
398
399 def test_gen_sink_input_ports_getitem(self):
400 port1 = None
401 port2 = None
402 port3 = None
403
404 class MySink(bt2._UserSinkComponent):
59225a3e 405 def __init__(comp_self, config, params, obj):
811644b8
PP
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')
410
6a91742b 411 def _user_consume(self):
a01b452b
SM
412 pass
413
811644b8
PP
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)
418 del port1
419 del port2
420 del port3
421
422 def test_gen_src_output_ports_getitem_invalid_key(self):
5c61fb9d
SM
423 class MySource(
424 bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
425 ):
59225a3e 426 def __init__(comp_self, config, params, obj):
811644b8
PP
427 comp_self._add_output_port('clear')
428 comp_self._add_output_port('print')
429 comp_self._add_output_port('insert')
430
431 comp = self._create_comp(MySource)
432
433 with self.assertRaises(KeyError):
434 comp.output_ports['hello']
435
436 def test_gen_flt_output_ports_getitem_invalid_key(self):
5c61fb9d
SM
437 class MyFilter(
438 bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
439 ):
59225a3e 440 def __init__(comp_self, config, params, obj):
811644b8
PP
441 comp_self._add_output_port('clear')
442 comp_self._add_output_port('print')
443 comp_self._add_output_port('insert')
444
445 comp = self._create_comp(MyFilter)
446
447 with self.assertRaises(KeyError):
448 comp.output_ports['hello']
449
450 def test_gen_flt_input_ports_getitem_invalid_key(self):
5c61fb9d
SM
451 class MyFilter(
452 bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
453 ):
59225a3e 454 def __init__(comp_self, config, params, obj):
811644b8
PP
455 comp_self._add_input_port('clear')
456 comp_self._add_input_port('print')
457 comp_self._add_input_port('insert')
458
459 comp = self._create_comp(MyFilter)
460
461 with self.assertRaises(KeyError):
462 comp.input_ports['hello']
463
464 def test_gen_sink_input_ports_getitem_invalid_key(self):
465 class MySink(bt2._UserSinkComponent):
59225a3e 466 def __init__(comp_self, config, params, obj):
811644b8
PP
467 comp_self._add_input_port('clear')
468 comp_self._add_input_port('print')
469 comp_self._add_input_port('insert')
470
471 with self.assertRaises(KeyError):
472 comp_self._input_ports['hello']
473
6a91742b 474 def _user_consume(self):
a01b452b
SM
475 pass
476
811644b8
PP
477 comp = self._create_comp(MySink)
478
479 with self.assertRaises(KeyError):
480 comp.input_ports['hello']
481
482 def test_gen_src_output_ports_len(self):
5c61fb9d
SM
483 class MySource(
484 bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
485 ):
59225a3e 486 def __init__(comp_self, config, params, obj):
811644b8
PP
487 comp_self._add_output_port('clear')
488 comp_self._add_output_port('print')
489 comp_self._add_output_port('insert')
490
491 comp = self._create_comp(MySource)
492 self.assertEqual(len(comp.output_ports), 3)
493
494 def test_gen_flt_output_ports_len(self):
5c61fb9d
SM
495 class MyFilter(
496 bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
497 ):
59225a3e 498 def __init__(comp_self, config, params, obj):
811644b8
PP
499 comp_self._add_output_port('clear')
500 comp_self._add_output_port('print')
501 comp_self._add_output_port('insert')
502
503 comp = self._create_comp(MyFilter)
504 self.assertEqual(len(comp.output_ports), 3)
505
506 def test_gen_flt_input_ports_len(self):
5c61fb9d
SM
507 class MyFilter(
508 bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
509 ):
59225a3e 510 def __init__(comp_self, config, params, obj):
811644b8
PP
511 comp_self._add_input_port('clear')
512 comp_self._add_input_port('print')
513 comp_self._add_input_port('insert')
514
515 comp = self._create_comp(MyFilter)
516 self.assertEqual(len(comp.input_ports), 3)
517
518 def test_gen_sink_input_ports_len(self):
519 class MySink(bt2._UserSinkComponent):
59225a3e 520 def __init__(comp_self, config, params, obj):
811644b8
PP
521 comp_self._add_input_port('clear')
522 comp_self._add_input_port('print')
523 comp_self._add_input_port('insert')
524
6a91742b 525 def _user_consume(self):
a01b452b
SM
526 pass
527
811644b8
PP
528 comp = self._create_comp(MySink)
529 self.assertEqual(len(comp.input_ports), 3)
530
531 def test_gen_src_output_ports_iter(self):
811644b8
PP
532 port1 = None
533 port2 = None
534 port3 = None
535
5c61fb9d
SM
536 class MySource(
537 bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
538 ):
59225a3e 539 def __init__(comp_self, config, params, obj):
811644b8
PP
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')
544
545 comp = self._create_comp(MySource)
546 ports = []
547
548 for port_name, port in comp.output_ports.items():
549 ports.append((port_name, port))
550
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)
557 del port1
558 del port2
559 del port3
560
561 def test_gen_flt_output_ports_iter(self):
811644b8
PP
562 port1 = None
563 port2 = None
564 port3 = None
565
5c61fb9d
SM
566 class MyFilter(
567 bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
568 ):
59225a3e 569 def __init__(comp_self, config, params, obj):
811644b8
PP
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')
574
575 comp = self._create_comp(MyFilter)
576 ports = []
577
578 for port_name, port in comp.output_ports.items():
579 ports.append((port_name, port))
580
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)
587 del port1
588 del port2
589 del port3
590
591 def test_gen_flt_input_ports_iter(self):
811644b8
PP
592 port1 = None
593 port2 = None
594 port3 = None
595
5c61fb9d
SM
596 class MyFilter(
597 bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
598 ):
59225a3e 599 def __init__(comp_self, config, params, obj):
811644b8
PP
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')
604
605 comp = self._create_comp(MyFilter)
606 ports = []
607
608 for port_name, port in comp.input_ports.items():
609 ports.append((port_name, port))
610
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)
617 del port1
618 del port2
619 del port3
620
621 def test_gen_sink_input_ports_iter(self):
622 port1 = None
623 port2 = None
624 port3 = None
625
626 class MySink(bt2._UserSinkComponent):
59225a3e 627 def __init__(comp_self, config, params, obj):
811644b8
PP
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')
632
6a91742b 633 def _user_consume(self):
a01b452b
SM
634 pass
635
811644b8
PP
636 comp = self._create_comp(MySink)
637 ports = []
638
639 for port_name, port in comp.input_ports.items():
640 ports.append((port_name, port))
641
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)
648 del port1
649 del port2
650 del port3
651
652 def test_name(self):
653 class MySink(bt2._UserSinkComponent):
59225a3e 654 def __init__(comp_self, config, params, obj):
811644b8
PP
655 comp_self._add_input_port('clear')
656
6a91742b 657 def _user_consume(self):
a01b452b
SM
658 pass
659
811644b8
PP
660 comp = self._create_comp(MySink)
661 self.assertEqual(comp.input_ports['clear'].name, 'clear')
662
811644b8
PP
663 def test_connection_none(self):
664 class MySink(bt2._UserSinkComponent):
59225a3e 665 def __init__(comp_self, config, params, obj):
811644b8
PP
666 comp_self._add_input_port('clear')
667
6a91742b 668 def _user_consume(self):
a01b452b
SM
669 pass
670
811644b8
PP
671 comp = self._create_comp(MySink)
672 self.assertIsNone(comp.input_ports['clear'].connection)
673
674 def test_is_connected_false(self):
675 class MySink(bt2._UserSinkComponent):
59225a3e 676 def __init__(comp_self, config, params, obj):
811644b8
PP
677 comp_self._add_input_port('clear')
678
6a91742b 679 def _user_consume(self):
a01b452b
SM
680 pass
681
811644b8
PP
682 comp = self._create_comp(MySink)
683 self.assertFalse(comp.input_ports['clear'].is_connected)
684
894a8df5 685 def test_self_name(self):
811644b8 686 class MySink(bt2._UserSinkComponent):
59225a3e 687 def __init__(comp_self, config, params, obj):
811644b8
PP
688 port = comp_self._add_input_port('clear')
689 self.assertEqual(port.name, 'clear')
690
6a91742b 691 def _user_consume(self):
a01b452b
SM
692 pass
693
894a8df5 694 self._create_comp(MySink)
811644b8 695
894a8df5 696 def test_self_connection_none(self):
811644b8 697 class MySink(bt2._UserSinkComponent):
59225a3e 698 def __init__(comp_self, config, params, obj):
811644b8
PP
699 port = comp_self._add_input_port('clear')
700 self.assertIsNone(port.connection)
701
6a91742b 702 def _user_consume(self):
a01b452b
SM
703 pass
704
894a8df5 705 self._create_comp(MySink)
811644b8 706
894a8df5 707 def test_self_is_connected_false(self):
811644b8 708 class MySink(bt2._UserSinkComponent):
59225a3e 709 def __init__(comp_self, config, params, obj):
811644b8
PP
710 port = comp_self._add_input_port('clear')
711 self.assertFalse(port.is_connected)
712
6a91742b 713 def _user_consume(self):
a01b452b
SM
714 pass
715
894a8df5 716 self._create_comp(MySink)
2e00bc76
SM
717
718 def test_source_self_port_user_data(self):
5c61fb9d
SM
719 class MySource(
720 bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
721 ):
59225a3e 722 def __init__(comp_self, config, params, obj):
2e00bc76
SM
723 nonlocal user_datas
724
725 p = comp_self._add_output_port('port1')
726 user_datas.append(p.user_data)
727 p = comp_self._add_output_port('port2', 2)
728 user_datas.append(p.user_data)
729
730 user_datas = []
731
082db648 732 self._create_comp(MySource)
2e00bc76
SM
733 self.assertEqual(user_datas, [None, 2])
734
735 def test_filter_self_port_user_data(self):
5c61fb9d
SM
736 class MyFilter(
737 bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
738 ):
59225a3e 739 def __init__(comp_self, config, params, obj):
2e00bc76
SM
740 nonlocal user_datas
741
742 p = comp_self._add_output_port('port1')
743 user_datas.append(p.user_data)
744 p = comp_self._add_output_port('port2', 'user data string')
745 user_datas.append(p.user_data)
746
747 p = comp_self._add_input_port('port3')
748 user_datas.append(p.user_data)
749 p = comp_self._add_input_port('port4', user_data={'user data': 'dict'})
750 user_datas.append(p.user_data)
751
752 user_datas = []
753
082db648 754 self._create_comp(MyFilter)
cfbd7cf3
FD
755 self.assertEqual(
756 user_datas, [None, 'user data string', None, {'user data': 'dict'}]
757 )
2e00bc76
SM
758
759 def test_sink_self_port_user_data(self):
1606ab4c 760 class MySink(bt2._UserSinkComponent):
59225a3e 761 def __init__(comp_self, config, params, obj):
2e00bc76
SM
762 nonlocal user_datas
763
764 p = comp_self._add_input_port('port1')
765 user_datas.append(p.user_data)
766 p = comp_self._add_input_port('port2', set())
767 user_datas.append(p.user_data)
768
1606ab4c
SM
769 def _user_consume(self):
770 pass
771
2e00bc76
SM
772 user_datas = []
773
082db648 774 self._create_comp(MySink)
2e00bc76 775 self.assertEqual(user_datas, [None, set()])
d14ddbba
SM
776
777
778if __name__ == '__main__':
779 unittest.main()
This page took 0.08402 seconds and 4 git commands to generate.