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