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 |
811644b8 | 7 | import bt2 |
5813b3a3 | 8 | from bt2 import port as bt2_port |
811644b8 PP |
9 | |
10 | ||
11 | class 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 | 901 | if __name__ == "__main__": |
d14ddbba | 902 | unittest.main() |