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