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