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