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