7c26a935536e5e9b2fe3a9bec9c6fa922328c66d
[babeltrace.git] / tests / bindings / python / bt2 / test_message.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 utils
22 from utils import TestOutputPortMessageIterator
23 from bt2 import clock_snapshot as bt2_clock_snapshot
24 from bt2 import event as bt2_event
25 from bt2 import event_class as bt2_event_class
26 from bt2 import field as bt2_field
27 from bt2 import packet as bt2_packet
28 from bt2 import stream as bt2_stream
29 from bt2 import stream_class as bt2_stream_class
30 from bt2 import trace as bt2_trace
31 from bt2 import trace_class as bt2_trace_class
32
33
34 class AllMessagesTestCase(unittest.TestCase):
35 def setUp(self):
36 class MyIter(bt2._UserMessageIterator):
37 def __init__(self, config, self_port_output):
38 self._at = 0
39 self._with_stream_msgs_clock_snapshots = self_port_output.user_data.get(
40 'with_stream_msgs_clock_snapshots', False
41 )
42
43 def __next__(self):
44 if test_obj._clock_class:
45 if self._at == 0:
46 if self._with_stream_msgs_clock_snapshots:
47 msg = self._create_stream_beginning_message(
48 test_obj._stream, default_clock_snapshot=self._at
49 )
50 else:
51 msg = self._create_stream_beginning_message(
52 test_obj._stream
53 )
54 test_obj.assertIs(type(msg), bt2._StreamBeginningMessage)
55 elif self._at == 1:
56 msg = self._create_packet_beginning_message(
57 test_obj._packet, self._at
58 )
59 test_obj.assertIs(type(msg), bt2._PacketBeginningMessage)
60 elif self._at == 2:
61 msg = self._create_event_message(
62 test_obj._event_class, test_obj._packet, self._at
63 )
64 test_obj.assertIs(type(msg), bt2._EventMessage)
65 elif self._at == 3:
66 msg = self._create_message_iterator_inactivity_message(
67 test_obj._clock_class, self._at
68 )
69 elif self._at == 4:
70 msg = self._create_discarded_events_message(
71 test_obj._stream, 890, self._at, self._at
72 )
73 test_obj.assertIs(type(msg), bt2._DiscardedEventsMessage)
74 elif self._at == 5:
75 msg = self._create_packet_end_message(
76 test_obj._packet, self._at
77 )
78 test_obj.assertIs(type(msg), bt2._PacketEndMessage)
79 elif self._at == 6:
80 msg = self._create_discarded_packets_message(
81 test_obj._stream, 678, self._at, self._at
82 )
83 test_obj.assertIs(type(msg), bt2._DiscardedPacketsMessage)
84 elif self._at == 7:
85 if self._with_stream_msgs_clock_snapshots:
86 msg = self._create_stream_end_message(
87 test_obj._stream, default_clock_snapshot=self._at
88 )
89 else:
90 msg = self._create_stream_end_message(test_obj._stream)
91 test_obj.assertIs(type(msg), bt2._StreamEndMessage)
92 elif self._at >= 8:
93 raise bt2.Stop
94 else:
95 if self._at == 0:
96 msg = self._create_stream_beginning_message(test_obj._stream)
97 elif self._at == 1:
98 msg = self._create_packet_beginning_message(test_obj._packet)
99 elif self._at == 2:
100 msg = self._create_event_message(
101 test_obj._event_class, test_obj._packet
102 )
103 elif self._at == 3:
104 msg = self._create_discarded_events_message(
105 test_obj._stream, 890
106 )
107 elif self._at == 4:
108 msg = self._create_packet_end_message(test_obj._packet)
109 elif self._at == 5:
110 msg = self._create_discarded_packets_message(
111 test_obj._stream, 678
112 )
113 elif self._at == 6:
114 msg = self._create_stream_end_message(test_obj._stream)
115 elif self._at >= 7:
116 raise bt2.Stop
117
118 self._at += 1
119 return msg
120
121 class MySrc(bt2._UserSourceComponent, message_iterator_class=MyIter):
122 def __init__(self, config, params, obj):
123 self._add_output_port('out', params)
124
125 with_cc = bool(params['with_cc'])
126 tc = self._create_trace_class()
127 if with_cc:
128 cc = self._create_clock_class()
129 else:
130 cc = None
131
132 sc = tc.create_stream_class(
133 default_clock_class=cc,
134 supports_packets=True,
135 packets_have_beginning_default_clock_snapshot=with_cc,
136 packets_have_end_default_clock_snapshot=with_cc,
137 supports_discarded_events=True,
138 discarded_events_have_default_clock_snapshots=with_cc,
139 supports_discarded_packets=True,
140 discarded_packets_have_default_clock_snapshots=with_cc,
141 )
142
143 # Create payload field class
144 my_int_fc = tc.create_signed_integer_field_class(32)
145 payload_fc = tc.create_structure_field_class()
146 payload_fc += [('my_int', my_int_fc)]
147
148 # Create specific context field class
149 my_int_fc = tc.create_signed_integer_field_class(32)
150 specific_fc = tc.create_structure_field_class()
151 specific_fc += [('my_int', my_int_fc)]
152
153 ec = sc.create_event_class(
154 name='salut',
155 payload_field_class=payload_fc,
156 specific_context_field_class=specific_fc,
157 )
158
159 trace = tc()
160 stream = trace.create_stream(sc)
161 packet = stream.create_packet()
162
163 test_obj._trace = trace
164 test_obj._stream = stream
165 test_obj._packet = packet
166 test_obj._event_class = ec
167 test_obj._clock_class = cc
168
169 test_obj = self
170 self._graph = bt2.Graph()
171 self._src = MySrc
172 self._iter = MyIter
173
174 def test_all_msg_with_cc(self):
175 params = {'with_cc': True}
176 self._src_comp = self._graph.add_component(self._src, 'my_source', params)
177 self._msg_iter = TestOutputPortMessageIterator(
178 self._graph, self._src_comp.output_ports['out']
179 )
180
181 for i, msg in enumerate(self._msg_iter):
182 if i == 0:
183 self.assertIs(type(msg), bt2._StreamBeginningMessageConst)
184 self.assertIs(type(msg.stream), bt2_stream._StreamConst)
185 self.assertEqual(msg.stream.addr, self._stream.addr)
186 self.assertIsInstance(
187 msg.default_clock_snapshot, bt2._UnknownClockSnapshot
188 )
189 elif i == 1:
190 self.assertIs(type(msg), bt2._PacketBeginningMessageConst)
191 self.assertIs(type(msg.packet), bt2_packet._PacketConst)
192 self.assertIs(
193 type(msg.default_clock_snapshot),
194 bt2_clock_snapshot._ClockSnapshotConst,
195 )
196 self.assertEqual(msg.packet.addr, self._packet.addr)
197 self.assertEqual(msg.default_clock_snapshot.value, i)
198 elif i == 2:
199 self.assertIs(type(msg), bt2._EventMessageConst)
200 self.assertIs(type(msg.event), bt2_event._EventConst)
201 self.assertIs(
202 type(msg.default_clock_snapshot),
203 bt2_clock_snapshot._ClockSnapshotConst,
204 )
205 self.assertIs(
206 type(msg.event.payload_field), bt2_field._StructureFieldConst
207 )
208 self.assertIs(
209 type(msg.event.payload_field['my_int']),
210 bt2_field._SignedIntegerFieldConst,
211 )
212
213 self.assertEqual(msg.event.cls.addr, self._event_class.addr)
214 self.assertEqual(msg.default_clock_snapshot.value, i)
215 elif i == 3:
216 self.assertIs(type(msg), bt2._MessageIteratorInactivityMessageConst)
217 self.assertIs(
218 type(msg.clock_snapshot), bt2_clock_snapshot._ClockSnapshotConst
219 )
220 self.assertEqual(msg.clock_snapshot.value, i)
221 elif i == 4:
222 self.assertIs(type(msg), bt2._DiscardedEventsMessageConst)
223 self.assertIs(type(msg.stream), bt2_stream._StreamConst)
224 self.assertIs(type(msg.stream.cls), bt2_stream_class._StreamClassConst)
225 self.assertIs(
226 type(msg.beginning_default_clock_snapshot),
227 bt2_clock_snapshot._ClockSnapshotConst,
228 )
229 self.assertIs(
230 type(msg.end_default_clock_snapshot),
231 bt2_clock_snapshot._ClockSnapshotConst,
232 )
233
234 self.assertEqual(msg.stream.addr, self._stream.addr)
235 self.assertEqual(msg.count, 890)
236 self.assertEqual(
237 msg.stream.cls.default_clock_class.addr, self._clock_class.addr
238 )
239 self.assertEqual(msg.beginning_default_clock_snapshot.value, i)
240 self.assertEqual(msg.end_default_clock_snapshot.value, i)
241 elif i == 5:
242 self.assertIs(type(msg), bt2._PacketEndMessageConst)
243 self.assertIs(type(msg.packet), bt2_packet._PacketConst)
244 self.assertIs(
245 type(msg.default_clock_snapshot),
246 bt2_clock_snapshot._ClockSnapshotConst,
247 )
248 self.assertEqual(msg.packet.addr, self._packet.addr)
249 self.assertEqual(msg.default_clock_snapshot.value, i)
250 elif i == 6:
251 self.assertIs(type(msg), bt2._DiscardedPacketsMessageConst)
252 self.assertIs(type(msg.stream), bt2_stream._StreamConst)
253 self.assertIs(type(msg.stream.trace), bt2_trace._TraceConst)
254 self.assertIs(
255 type(msg.stream.trace.cls), bt2_trace_class._TraceClassConst
256 )
257 self.assertIs(
258 type(msg.beginning_default_clock_snapshot),
259 bt2_clock_snapshot._ClockSnapshotConst,
260 )
261 self.assertIs(
262 type(msg.end_default_clock_snapshot),
263 bt2_clock_snapshot._ClockSnapshotConst,
264 )
265 self.assertEqual(msg.stream.addr, self._stream.addr)
266 self.assertEqual(msg.count, 678)
267 self.assertEqual(
268 msg.stream.cls.default_clock_class.addr, self._clock_class.addr
269 )
270 self.assertEqual(msg.beginning_default_clock_snapshot.value, i)
271 self.assertEqual(msg.end_default_clock_snapshot.value, i)
272 elif i == 7:
273 self.assertIs(type(msg), bt2._StreamEndMessageConst)
274 self.assertIs(type(msg.stream), bt2_stream._StreamConst)
275 self.assertEqual(msg.stream.addr, self._stream.addr)
276 self.assertIs(
277 type(msg.default_clock_snapshot), bt2._UnknownClockSnapshot
278 )
279 else:
280 raise Exception
281
282 def test_all_msg_without_cc(self):
283 params = {'with_cc': False}
284 self._src_comp = self._graph.add_component(self._src, 'my_source', params)
285 self._msg_iter = TestOutputPortMessageIterator(
286 self._graph, self._src_comp.output_ports['out']
287 )
288
289 for i, msg in enumerate(self._msg_iter):
290 if i == 0:
291 self.assertIsInstance(msg, bt2._StreamBeginningMessageConst)
292 self.assertIs(type(msg.stream), bt2_stream._StreamConst)
293 self.assertEqual(msg.stream.addr, self._stream.addr)
294 with self.assertRaisesRegex(
295 ValueError, 'stream class has no default clock class'
296 ):
297 msg.default_clock_snapshot
298 elif i == 1:
299 self.assertIsInstance(msg, bt2._PacketBeginningMessageConst)
300 self.assertIs(type(msg.packet), bt2_packet._PacketConst)
301 self.assertEqual(msg.packet.addr, self._packet.addr)
302 elif i == 2:
303 self.assertIsInstance(msg, bt2._EventMessageConst)
304 self.assertIs(type(msg.event), bt2_event._EventConst)
305 self.assertIs(type(msg.event.cls), bt2_event_class._EventClassConst)
306 self.assertEqual(msg.event.cls.addr, self._event_class.addr)
307 with self.assertRaisesRegex(
308 ValueError, 'stream class has no default clock class'
309 ):
310 msg.default_clock_snapshot
311 elif i == 3:
312 self.assertIsInstance(msg, bt2._DiscardedEventsMessageConst)
313 self.assertIs(type(msg.stream), bt2_stream._StreamConst)
314 self.assertIs(type(msg.stream.cls), bt2_stream_class._StreamClassConst)
315 self.assertEqual(msg.stream.addr, self._stream.addr)
316 self.assertEqual(msg.count, 890)
317 self.assertIsNone(msg.stream.cls.default_clock_class)
318 with self.assertRaisesRegex(
319 ValueError,
320 'such a message has no clock snapshots for this stream class',
321 ):
322 msg.beginning_default_clock_snapshot
323 with self.assertRaisesRegex(
324 ValueError,
325 'such a message has no clock snapshots for this stream class',
326 ):
327 msg.end_default_clock_snapshot
328 elif i == 4:
329 self.assertIsInstance(msg, bt2._PacketEndMessageConst)
330 self.assertEqual(msg.packet.addr, self._packet.addr)
331 self.assertIs(type(msg.packet), bt2_packet._PacketConst)
332 elif i == 5:
333 self.assertIsInstance(msg, bt2._DiscardedPacketsMessageConst)
334 self.assertIs(type(msg.stream), bt2_stream._StreamConst)
335 self.assertIs(type(msg.stream.cls), bt2_stream_class._StreamClassConst)
336 self.assertIs(
337 type(msg.stream.cls.trace_class), bt2_trace_class._TraceClassConst
338 )
339 self.assertEqual(msg.stream.addr, self._stream.addr)
340 self.assertEqual(msg.count, 678)
341 self.assertIsNone(msg.stream.cls.default_clock_class)
342 with self.assertRaisesRegex(
343 ValueError,
344 'such a message has no clock snapshots for this stream class',
345 ):
346 msg.beginning_default_clock_snapshot
347 with self.assertRaisesRegex(
348 ValueError,
349 'such a message has no clock snapshots for this stream class',
350 ):
351 msg.end_default_clock_snapshot
352 elif i == 6:
353 self.assertIsInstance(msg, bt2._StreamEndMessageConst)
354 self.assertIs(type(msg.stream), bt2_stream._StreamConst)
355 self.assertEqual(msg.stream.addr, self._stream.addr)
356 with self.assertRaisesRegex(
357 ValueError, 'stream class has no default clock class'
358 ):
359 msg.default_clock_snapshot
360 else:
361 raise Exception
362
363 def test_msg_stream_with_clock_snapshots(self):
364 params = {'with_cc': True, 'with_stream_msgs_clock_snapshots': True}
365
366 self._src_comp = self._graph.add_component(self._src, 'my_source', params)
367 self._msg_iter = TestOutputPortMessageIterator(
368 self._graph, self._src_comp.output_ports['out']
369 )
370 msgs = list(self._msg_iter)
371
372 msg_stream_beg = msgs[0]
373 self.assertIsInstance(msg_stream_beg, bt2._StreamBeginningMessageConst)
374 self.assertIs(
375 type(msg_stream_beg.default_clock_snapshot),
376 bt2_clock_snapshot._ClockSnapshotConst,
377 )
378 self.assertEqual(msg_stream_beg.default_clock_snapshot.value, 0)
379
380 msg_stream_end = msgs[7]
381 self.assertIsInstance(msg_stream_end, bt2._StreamEndMessageConst)
382 self.assertIs(
383 type(msg_stream_end.default_clock_snapshot),
384 bt2_clock_snapshot._ClockSnapshotConst,
385 )
386 self.assertEqual(msg_stream_end.default_clock_snapshot.value, 7)
387
388 def test_stream_beg_msg(self):
389 msg = utils.get_stream_beginning_message()
390 self.assertIs(type(msg.stream), bt2_stream._Stream)
391
392 def test_stream_end_msg(self):
393 msg = utils.get_stream_end_message()
394 self.assertIs(type(msg.stream), bt2_stream._Stream)
395
396 def test_packet_beg_msg(self):
397 msg = utils.get_packet_beginning_message()
398 self.assertIs(type(msg.packet), bt2_packet._Packet)
399
400 def test_packet_end_msg(self):
401 msg = utils.get_packet_end_message()
402 self.assertIs(type(msg.packet), bt2_packet._Packet)
403
404 def test_event_msg(self):
405 msg = utils.get_event_message()
406 self.assertIs(type(msg.event), bt2_event._Event)
407
408
409 if __name__ == '__main__':
410 unittest.main()
This page took 0.042622 seconds and 3 git commands to generate.