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