lib: make message iterator creation functions return a status
[babeltrace.git] / tests / bindings / python / bt2 / test_message_iterator.py
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
19 import unittest
20 import bt2
21 import sys
22 from utils import TestOutputPortMessageIterator
23
24
25 class UserMessageIteratorTestCase(unittest.TestCase):
26 @staticmethod
27 def _create_graph(src_comp_cls, flt_comp_cls=None):
28 class MySink(bt2._UserSinkComponent):
29 def __init__(self, params, obj):
30 self._add_input_port('in')
31
32 def _user_consume(self):
33 next(self._msg_iter)
34
35 def _user_graph_is_configured(self):
36 self._msg_iter = self._create_input_port_message_iterator(
37 self._input_ports['in']
38 )
39
40 graph = bt2.Graph()
41 src_comp = graph.add_component(src_comp_cls, 'src')
42
43 if flt_comp_cls is not None:
44 flt_comp = graph.add_component(flt_comp_cls, 'flt')
45
46 sink_comp = graph.add_component(MySink, 'sink')
47
48 if flt_comp_cls is not None:
49 assert flt_comp is not None
50 graph.connect_ports(
51 src_comp.output_ports['out'], flt_comp.input_ports['in']
52 )
53 out_port = flt_comp.output_ports['out']
54 else:
55 out_port = src_comp.output_ports['out']
56
57 graph.connect_ports(out_port, sink_comp.input_ports['in'])
58 return graph
59
60 def test_init(self):
61 the_output_port_from_source = None
62 the_output_port_from_iter = None
63
64 class MyIter(bt2._UserMessageIterator):
65 def __init__(self, self_port_output):
66 nonlocal initialized
67 nonlocal the_output_port_from_iter
68 initialized = True
69 the_output_port_from_iter = self_port_output
70
71 class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
72 def __init__(self, params, obj):
73 nonlocal the_output_port_from_source
74 the_output_port_from_source = self._add_output_port('out', 'user data')
75
76 initialized = False
77 graph = self._create_graph(MySource)
78 graph.run()
79 self.assertTrue(initialized)
80 self.assertEqual(
81 the_output_port_from_source.addr, the_output_port_from_iter.addr
82 )
83 self.assertEqual(the_output_port_from_iter.user_data, 'user data')
84
85 def test_create_from_message_iterator(self):
86 class MySourceIter(bt2._UserMessageIterator):
87 def __init__(self, self_port_output):
88 nonlocal src_iter_initialized
89 src_iter_initialized = True
90
91 class MySource(bt2._UserSourceComponent, message_iterator_class=MySourceIter):
92 def __init__(self, params, obj):
93 self._add_output_port('out')
94
95 class MyFilterIter(bt2._UserMessageIterator):
96 def __init__(self, self_port_output):
97 nonlocal flt_iter_initialized
98 flt_iter_initialized = True
99 self._up_iter = self._create_input_port_message_iterator(
100 self._component._input_ports['in']
101 )
102
103 def __next__(self):
104 return next(self._up_iter)
105
106 class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyFilterIter):
107 def __init__(self, params, obj):
108 self._add_input_port('in')
109 self._add_output_port('out')
110
111 src_iter_initialized = False
112 flt_iter_initialized = False
113 graph = self._create_graph(MySource, MyFilter)
114 graph.run()
115 self.assertTrue(src_iter_initialized)
116 self.assertTrue(flt_iter_initialized)
117
118 def test_create_user_error(self):
119 # This tests both error handling by
120 # _UserSinkComponent._create_input_port_message_iterator
121 # and _UserMessageIterator._create_input_port_message_iterator, as they
122 # are both used in the graph.
123 class MySourceIter(bt2._UserMessageIterator):
124 def __init__(self, self_port_output):
125 raise ValueError('Very bad error')
126
127 class MySource(bt2._UserSourceComponent, message_iterator_class=MySourceIter):
128 def __init__(self, params, obj):
129 self._add_output_port('out')
130
131 class MyFilterIter(bt2._UserMessageIterator):
132 def __init__(self, self_port_output):
133 # This is expected to raise because of the error in
134 # MySourceIter.__init__.
135 self._create_input_port_message_iterator(
136 self._component._input_ports['in']
137 )
138
139 class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyFilterIter):
140 def __init__(self, params, obj):
141 self._add_input_port('in')
142 self._add_output_port('out')
143
144 graph = self._create_graph(MySource, MyFilter)
145
146 with self.assertRaises(bt2._Error) as ctx:
147 graph.run()
148
149 exc = ctx.exception
150 cause = exc[0]
151
152 self.assertIsInstance(cause, bt2._MessageIteratorErrorCause)
153 self.assertEqual(cause.component_name, 'src')
154 self.assertEqual(cause.component_output_port_name, 'out')
155 self.assertIn('ValueError: Very bad error', cause.message)
156
157 def test_finalize(self):
158 class MyIter(bt2._UserMessageIterator):
159 def _user_finalize(self):
160 nonlocal finalized
161 finalized = True
162
163 class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
164 def __init__(self, params, obj):
165 self._add_output_port('out')
166
167 finalized = False
168 graph = self._create_graph(MySource)
169 graph.run()
170 del graph
171 self.assertTrue(finalized)
172
173 def test_component(self):
174 class MyIter(bt2._UserMessageIterator):
175 def __init__(self, self_port_output):
176 nonlocal salut
177 salut = self._component._salut
178
179 class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
180 def __init__(self, params, obj):
181 self._add_output_port('out')
182 self._salut = 23
183
184 salut = None
185 graph = self._create_graph(MySource)
186 graph.run()
187 self.assertEqual(salut, 23)
188
189 def test_addr(self):
190 class MyIter(bt2._UserMessageIterator):
191 def __init__(self, self_port_output):
192 nonlocal addr
193 addr = self.addr
194
195 class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
196 def __init__(self, params, obj):
197 self._add_output_port('out')
198
199 addr = None
200 graph = self._create_graph(MySource)
201 graph.run()
202 self.assertIsNotNone(addr)
203 self.assertNotEqual(addr, 0)
204
205 # Test that messages returned by _UserMessageIterator.__next__ remain valid
206 # and can be re-used.
207 def test_reuse_message(self):
208 class MyIter(bt2._UserMessageIterator):
209 def __init__(self, port):
210 tc, sc, ec = port.user_data
211 trace = tc()
212 stream = trace.create_stream(sc)
213 packet = stream.create_packet()
214
215 # This message will be returned twice by __next__.
216 event_message = self._create_event_message(ec, packet)
217
218 self._msgs = [
219 self._create_stream_beginning_message(stream),
220 self._create_packet_beginning_message(packet),
221 event_message,
222 event_message,
223 ]
224
225 def __next__(self):
226 return self._msgs.pop(0)
227
228 class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
229 def __init__(self, params, obj):
230 tc = self._create_trace_class()
231 sc = tc.create_stream_class(supports_packets=True)
232 ec = sc.create_event_class()
233 self._add_output_port('out', (tc, sc, ec))
234
235 graph = bt2.Graph()
236 src = graph.add_component(MySource, 'src')
237 it = TestOutputPortMessageIterator(graph, src.output_ports['out'])
238
239 # Skip beginning messages.
240 msg = next(it)
241 self.assertIsInstance(msg, bt2._StreamBeginningMessage)
242 msg = next(it)
243 self.assertIsInstance(msg, bt2._PacketBeginningMessage)
244
245 msg_ev1 = next(it)
246 msg_ev2 = next(it)
247
248 self.assertIsInstance(msg_ev1, bt2._EventMessage)
249 self.assertIsInstance(msg_ev2, bt2._EventMessage)
250 self.assertEqual(msg_ev1.addr, msg_ev2.addr)
251
252 @staticmethod
253 def _setup_seek_beginning_test(sink_cls):
254 # Use a source, a filter and an output port iterator. This allows us
255 # to test calling `seek_beginning` on both a _OutputPortMessageIterator
256 # and a _UserComponentInputPortMessageIterator, on top of checking that
257 # _UserMessageIterator._seek_beginning is properly called.
258
259 class MySourceIter(bt2._UserMessageIterator):
260 def __init__(self, port):
261 tc, sc, ec = port.user_data
262 trace = tc()
263 stream = trace.create_stream(sc)
264 packet = stream.create_packet()
265
266 self._msgs = [
267 self._create_stream_beginning_message(stream),
268 self._create_packet_beginning_message(packet),
269 self._create_event_message(ec, packet),
270 self._create_event_message(ec, packet),
271 self._create_packet_end_message(packet),
272 self._create_stream_end_message(stream),
273 ]
274 self._at = 0
275
276 def _user_seek_beginning(self):
277 self._at = 0
278
279 def __next__(self):
280 if self._at < len(self._msgs):
281 msg = self._msgs[self._at]
282 self._at += 1
283 return msg
284 else:
285 raise StopIteration
286
287 class MySource(bt2._UserSourceComponent, message_iterator_class=MySourceIter):
288 def __init__(self, params, obj):
289 tc = self._create_trace_class()
290 sc = tc.create_stream_class(supports_packets=True)
291 ec = sc.create_event_class()
292
293 self._add_output_port('out', (tc, sc, ec))
294
295 class MyFilterIter(bt2._UserMessageIterator):
296 def __init__(self, port):
297 input_port = port.user_data
298 self._upstream_iter = self._create_input_port_message_iterator(
299 input_port
300 )
301
302 def __next__(self):
303 return next(self._upstream_iter)
304
305 def _user_seek_beginning(self):
306 self._upstream_iter.seek_beginning()
307
308 @property
309 def _user_can_seek_beginning(self):
310 return self._upstream_iter.can_seek_beginning
311
312 class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyFilterIter):
313 def __init__(self, params, obj):
314 input_port = self._add_input_port('in')
315 self._add_output_port('out', input_port)
316
317 graph = bt2.Graph()
318 src = graph.add_component(MySource, 'src')
319 flt = graph.add_component(MyFilter, 'flt')
320 sink = graph.add_component(sink_cls, 'sink')
321 graph.connect_ports(src.output_ports['out'], flt.input_ports['in'])
322 graph.connect_ports(flt.output_ports['out'], sink.input_ports['in'])
323 return MySourceIter, graph
324
325 def test_can_seek_beginning(self):
326 class MySink(bt2._UserSinkComponent):
327 def __init__(self, params, obj):
328 self._add_input_port('in')
329
330 def _user_graph_is_configured(self):
331 self._msg_iter = self._create_input_port_message_iterator(
332 self._input_ports['in']
333 )
334
335 def _user_consume(self):
336 nonlocal can_seek_beginning
337 can_seek_beginning = self._msg_iter.can_seek_beginning
338
339 MySourceIter, graph = self._setup_seek_beginning_test(MySink)
340
341 def _user_can_seek_beginning(self):
342 nonlocal input_port_iter_can_seek_beginning
343 return input_port_iter_can_seek_beginning
344
345 MySourceIter._user_can_seek_beginning = property(_user_can_seek_beginning)
346
347 input_port_iter_can_seek_beginning = True
348 can_seek_beginning = None
349 graph.run_once()
350 self.assertTrue(can_seek_beginning)
351
352 input_port_iter_can_seek_beginning = False
353 can_seek_beginning = None
354 graph.run_once()
355 self.assertFalse(can_seek_beginning)
356
357 # Once can_seek_beginning returns an error, verify that it raises when
358 # _can_seek_beginning has/returns the wrong type.
359
360 # Remove the _can_seek_beginning method, we now rely on the presence of
361 # a _seek_beginning method to know whether the iterator can seek to
362 # beginning or not.
363 del MySourceIter._user_can_seek_beginning
364 can_seek_beginning = None
365 graph.run_once()
366 self.assertTrue(can_seek_beginning)
367
368 del MySourceIter._user_seek_beginning
369 can_seek_beginning = None
370 graph.run_once()
371 self.assertFalse(can_seek_beginning)
372
373 def test_seek_beginning(self):
374 class MySink(bt2._UserSinkComponent):
375 def __init__(self, params, obj):
376 self._add_input_port('in')
377
378 def _user_graph_is_configured(self):
379 self._msg_iter = self._create_input_port_message_iterator(
380 self._input_ports['in']
381 )
382
383 def _user_consume(self):
384 nonlocal do_seek_beginning
385 nonlocal msg
386
387 if do_seek_beginning:
388 self._msg_iter.seek_beginning()
389 return
390
391 msg = next(self._msg_iter)
392
393 do_seek_beginning = False
394 msg = None
395 MySourceIter, graph = self._setup_seek_beginning_test(MySink)
396 graph.run_once()
397 self.assertIsInstance(msg, bt2._StreamBeginningMessage)
398 graph.run_once()
399 self.assertIsInstance(msg, bt2._PacketBeginningMessage)
400 do_seek_beginning = True
401 graph.run_once()
402 do_seek_beginning = False
403 graph.run_once()
404 self.assertIsInstance(msg, bt2._StreamBeginningMessage)
405
406 def test_seek_beginning_user_error(self):
407 class MySink(bt2._UserSinkComponent):
408 def __init__(self, params, obj):
409 self._add_input_port('in')
410
411 def _user_graph_is_configured(self):
412 self._msg_iter = self._create_input_port_message_iterator(
413 self._input_ports['in']
414 )
415
416 def _user_consume(self):
417 self._msg_iter.seek_beginning()
418
419 MySourceIter, graph = self._setup_seek_beginning_test(MySink)
420
421 def _user_seek_beginning_error(self):
422 raise ValueError('ouch')
423
424 MySourceIter._user_seek_beginning = _user_seek_beginning_error
425
426 with self.assertRaises(bt2._Error):
427 graph.run_once()
428
429 # Try consuming many times from an iterator that always returns TryAgain.
430 # This verifies that we are not missing an incref of Py_None, making the
431 # refcount of Py_None reach 0.
432 def test_try_again_many_times(self):
433 class MyIter(bt2._UserMessageIterator):
434 def __next__(self):
435 raise bt2.TryAgain
436
437 class MySource(bt2._UserSourceComponent, message_iterator_class=MyIter):
438 def __init__(self, params, obj):
439 self._add_output_port('out')
440
441 graph = bt2.Graph()
442 src = graph.add_component(MySource, 'src')
443 it = TestOutputPortMessageIterator(graph, src.output_ports['out'])
444
445 # Three times the initial ref count of `None` iterations should
446 # be enough to catch the bug even if there are small differences
447 # between configurations.
448 none_ref_count = sys.getrefcount(None) * 3
449
450 for i in range(none_ref_count):
451 with self.assertRaises(bt2.TryAgain):
452 next(it)
453
454
455 if __name__ == '__main__':
456 unittest.main()
This page took 0.043241 seconds and 4 git commands to generate.