Fix: log.h: missing defines of `_ERRNO()` macros to `_UNUSED()`
[babeltrace.git] / tests / bindings / python / bt2 / test_event.py
CommitLineData
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
9cf643d1 19import unittest
9cf643d1 20import bt2
f0a42b33 21import utils
6c373cc9 22from utils import TestOutputPortMessageIterator
9cf643d1 23
f0a42b33
FD
24from bt2 import field as bt2_field
25from bt2 import stream as bt2_stream
26from bt2 import event_class as bt2_event_class
27from bt2 import clock_snapshot as bt2_clock_snapshot
28
9cf643d1
PP
29
30class EventTestCase(unittest.TestCase):
f0a42b33 31 def _create_test_const_event_message(
cfbd7cf3
FD
32 self,
33 packet_fields_config=None,
34 event_fields_config=None,
35 with_clockclass=False,
36 with_cc=False,
37 with_sc=False,
38 with_ep=False,
39 with_packet=False,
40 ):
2ae9f48c 41 class MyIter(bt2._UserMessageIterator):
8d8b141d 42 def __init__(self, config, self_output_port):
2ae9f48c 43 self._at = 0
cfbd7cf3 44 self._msgs = [self._create_stream_beginning_message(test_obj.stream)]
2ae9f48c 45
26fc5aed 46 if with_packet:
2ae9f48c 47 assert test_obj.packet
cfbd7cf3
FD
48 self._msgs.append(
49 self._create_packet_beginning_message(test_obj.packet)
50 )
26fc5aed
PP
51
52 default_clock_snapshot = 789 if with_clockclass else None
53
54 if with_packet:
2ae9f48c 55 assert test_obj.packet
26fc5aed
PP
56 ev_parent = test_obj.packet
57 else:
58 assert test_obj.stream
59 ev_parent = test_obj.stream
60
cfbd7cf3
FD
61 msg = self._create_event_message(
62 test_obj.event_class, ev_parent, default_clock_snapshot
63 )
26fc5aed
PP
64
65 if event_fields_config is not None:
66 event_fields_config(msg.event)
67
68 self._msgs.append(msg)
69
70 if with_packet:
71 self._msgs.append(self._create_packet_end_message(test_obj.packet))
72
73 self._msgs.append(self._create_stream_end_message(test_obj.stream))
74
75 def __next__(self):
76 if self._at == len(self._msgs):
2ae9f48c
SM
77 raise bt2.Stop
78
26fc5aed 79 msg = self._msgs[self._at]
2ae9f48c
SM
80 self._at += 1
81 return msg
82
83 class MySrc(bt2._UserSourceComponent, message_iterator_class=MyIter):
59225a3e 84 def __init__(self, config, params, obj):
2ae9f48c
SM
85 self._add_output_port('out')
86 tc = self._create_trace_class()
87
88 clock_class = None
89 if with_clockclass:
90 clock_class = self._create_clock_class(frequency=1000)
91
92 # event common context (stream-class-defined)
93 cc = None
94 if with_cc:
95 cc = tc.create_structure_field_class()
45c51519 96 cc += [
2ae9f48c 97 ('cpu_id', tc.create_signed_integer_field_class(8)),
fe4df857 98 ('stuff', tc.create_double_precision_real_field_class()),
45c51519 99 ]
2ae9f48c
SM
100
101 # packet context (stream-class-defined)
26fc5aed
PP
102 pc = None
103
104 if with_packet:
105 pc = tc.create_structure_field_class()
45c51519 106 pc += [
26fc5aed 107 ('something', tc.create_unsigned_integer_field_class(8)),
fe4df857
FD
108 (
109 'something_else',
110 tc.create_double_precision_real_field_class(),
111 ),
45c51519 112 ]
2ae9f48c 113
cfbd7cf3
FD
114 stream_class = tc.create_stream_class(
115 default_clock_class=clock_class,
116 event_common_context_field_class=cc,
117 packet_context_field_class=pc,
118 supports_packets=with_packet,
119 )
2ae9f48c
SM
120
121 # specific context (event-class-defined)
122 sc = None
123 if with_sc:
124 sc = tc.create_structure_field_class()
45c51519 125 sc += [
2ae9f48c
SM
126 ('ant', tc.create_signed_integer_field_class(16)),
127 ('msg', tc.create_string_field_class()),
45c51519 128 ]
2ae9f48c
SM
129
130 # event payload
131 ep = None
132 if with_ep:
133 ep = tc.create_structure_field_class()
45c51519 134 ep += [
2ae9f48c
SM
135 ('giraffe', tc.create_signed_integer_field_class(32)),
136 ('gnu', tc.create_signed_integer_field_class(8)),
137 ('mosquito', tc.create_signed_integer_field_class(8)),
45c51519 138 ]
2ae9f48c 139
cfbd7cf3
FD
140 event_class = stream_class.create_event_class(
141 name='garou',
142 specific_context_field_class=sc,
143 payload_field_class=ep,
144 )
2ae9f48c
SM
145
146 trace = tc()
147 stream = trace.create_stream(stream_class)
26fc5aed
PP
148
149 if with_packet:
150 packet = stream.create_packet()
2ae9f48c
SM
151
152 if packet_fields_config is not None:
26fc5aed 153 assert packet
2ae9f48c
SM
154 packet_fields_config(packet)
155
26fc5aed
PP
156 if with_packet:
157 test_obj.packet = packet
158
2ae9f48c
SM
159 test_obj.stream = stream
160 test_obj.event_class = event_class
161
162 test_obj = self
163 self._graph = bt2.Graph()
164 self._src_comp = self._graph.add_component(MySrc, 'my_source')
6c373cc9
PP
165 self._msg_iter = TestOutputPortMessageIterator(
166 self._graph, self._src_comp.output_ports['out']
cfbd7cf3 167 )
2ae9f48c 168
26fc5aed 169 for msg in self._msg_iter:
f0a42b33
FD
170 if type(msg) is bt2._EventMessageConst:
171 self._event_msg = msg
2ae9f48c 172 return msg
9cf643d1 173
f0a42b33
FD
174 def test_const_attr_event_class(self):
175 msg = self._create_test_const_event_message()
e8ac1aae 176 self.assertEqual(msg.event.cls.addr, self.event_class.addr)
f0a42b33
FD
177 self.assertIs(type(msg.event.cls), bt2_event_class._EventClassConst)
178
179 def test_attr_event_class(self):
180 msg = utils.get_event_message()
181 self.assertIs(type(msg.event.cls), bt2_event_class._EventClass)
9cf643d1 182
f0a42b33
FD
183 def test_const_attr_name(self):
184 msg = self._create_test_const_event_message()
2ae9f48c 185 self.assertEqual(msg.event.name, self.event_class.name)
9cf643d1 186
f0a42b33
FD
187 def test_const_attr_id(self):
188 msg = self._create_test_const_event_message()
2ae9f48c
SM
189 self.assertEqual(msg.event.id, self.event_class.id)
190
f0a42b33 191 def test_const_get_common_context_field(self):
2ae9f48c
SM
192 def event_fields_config(event):
193 event.common_context_field['cpu_id'] = 1
194 event.common_context_field['stuff'] = 13.194
195
f0a42b33 196 msg = self._create_test_const_event_message(
cfbd7cf3
FD
197 event_fields_config=event_fields_config, with_cc=True
198 )
2ae9f48c
SM
199
200 self.assertEqual(msg.event.common_context_field['cpu_id'], 1)
201 self.assertEqual(msg.event.common_context_field['stuff'], 13.194)
f0a42b33
FD
202 self.assertIs(
203 type(msg.event.common_context_field), bt2_field._StructureFieldConst
204 )
205
206 def test_attr_common_context_field(self):
207 msg = utils.get_event_message()
208 self.assertIs(type(msg.event.common_context_field), bt2_field._StructureField)
2ae9f48c 209
f0a42b33
FD
210 def test_const_no_common_context_field(self):
211 msg = self._create_test_const_event_message(with_cc=False)
2ae9f48c
SM
212 self.assertIsNone(msg.event.common_context_field)
213
f0a42b33 214 def test_const_get_specific_context_field(self):
2ae9f48c
SM
215 def event_fields_config(event):
216 event.specific_context_field['ant'] = -1
217 event.specific_context_field['msg'] = 'hellooo'
218
f0a42b33 219 msg = self._create_test_const_event_message(
cfbd7cf3
FD
220 event_fields_config=event_fields_config, with_sc=True
221 )
2ae9f48c
SM
222
223 self.assertEqual(msg.event.specific_context_field['ant'], -1)
224 self.assertEqual(msg.event.specific_context_field['msg'], 'hellooo')
f0a42b33
FD
225 self.assertIs(
226 type(msg.event.specific_context_field), bt2_field._StructureFieldConst
227 )
228
229 def test_attr_specific_context_field(self):
230 msg = utils.get_event_message()
231 self.assertIs(type(msg.event.specific_context_field), bt2_field._StructureField)
2ae9f48c 232
f0a42b33
FD
233 def test_const_no_specific_context_field(self):
234 msg = self._create_test_const_event_message(with_sc=False)
2ae9f48c 235 self.assertIsNone(msg.event.specific_context_field)
9cf643d1 236
f0a42b33 237 def test_const_get_event_payload_field(self):
2ae9f48c
SM
238 def event_fields_config(event):
239 event.payload_field['giraffe'] = 1
240 event.payload_field['gnu'] = 23
241 event.payload_field['mosquito'] = 42
9cf643d1 242
f0a42b33 243 msg = self._create_test_const_event_message(
cfbd7cf3
FD
244 event_fields_config=event_fields_config, with_ep=True
245 )
9cf643d1 246
2ae9f48c
SM
247 self.assertEqual(msg.event.payload_field['giraffe'], 1)
248 self.assertEqual(msg.event.payload_field['gnu'], 23)
249 self.assertEqual(msg.event.payload_field['mosquito'], 42)
f0a42b33
FD
250 self.assertIs(type(msg.event.payload_field), bt2_field._StructureFieldConst)
251
252 def test_attr_payload_field(self):
253 msg = utils.get_event_message()
254 self.assertIs(type(msg.event.payload_field), bt2_field._StructureField)
9cf643d1 255
f0a42b33
FD
256 def test_const_no_payload_field(self):
257 msg = self._create_test_const_event_message(with_ep=False)
2ae9f48c 258 self.assertIsNone(msg.event.payload_field)
9cf643d1 259
f0a42b33
FD
260 def test_const_clock_value(self):
261 msg = self._create_test_const_event_message(with_clockclass=True)
262 self.assertEqual(msg.default_clock_snapshot.value, 789)
263 self.assertIs(
264 type(msg.default_clock_snapshot), bt2_clock_snapshot._ClockSnapshotConst
265 )
266
2ae9f48c 267 def test_clock_value(self):
f0a42b33 268 msg = utils.get_event_message()
2ae9f48c 269 self.assertEqual(msg.default_clock_snapshot.value, 789)
f0a42b33
FD
270 self.assertIs(
271 type(msg.default_clock_snapshot), bt2_clock_snapshot._ClockSnapshotConst
272 )
2ae9f48c 273
f0a42b33
FD
274 def test_const_no_clock_value(self):
275 msg = self._create_test_const_event_message(with_clockclass=False)
1153eccb
SM
276 with self.assertRaisesRegex(
277 ValueError, 'stream class has no default clock class'
278 ):
9ec609ec 279 msg.default_clock_snapshot
2ae9f48c 280
f0a42b33
FD
281 def test_const_stream(self):
282 msg = self._create_test_const_event_message()
2ae9f48c 283 self.assertEqual(msg.event.stream.addr, self.stream.addr)
f0a42b33 284 self.assertIs(type(msg.event.stream), bt2_stream._StreamConst)
2ae9f48c 285
f0a42b33
FD
286 def test_stream(self):
287 msg = utils.get_event_message()
288 self.assertIs(type(msg.event.stream), bt2_stream._Stream)
289
290 def test_const_getitem(self):
2ae9f48c
SM
291 def event_fields_config(event):
292 event.payload_field['giraffe'] = 1
293 event.payload_field['gnu'] = 23
294 event.payload_field['mosquito'] = 42
295 event.specific_context_field['ant'] = -1
296 event.specific_context_field['msg'] = 'hellooo'
297 event.common_context_field['cpu_id'] = 1
298 event.common_context_field['stuff'] = 13.194
299
300 def packet_fields_config(packet):
301 packet.context_field['something'] = 154
302 packet.context_field['something_else'] = 17.2
303
f0a42b33 304 msg = self._create_test_const_event_message(
cfbd7cf3
FD
305 packet_fields_config=packet_fields_config,
306 event_fields_config=event_fields_config,
307 with_cc=True,
308 with_sc=True,
309 with_ep=True,
310 with_packet=True,
311 )
2ae9f48c
SM
312 ev = msg.event
313
314 # Test event fields
9cf643d1 315 self.assertEqual(ev['giraffe'], 1)
f0a42b33 316 self.assertIs(type(ev['giraffe']), bt2_field._SignedIntegerFieldConst)
2ae9f48c
SM
317 self.assertEqual(ev['gnu'], 23)
318 self.assertEqual(ev['mosquito'], 42)
9cf643d1 319 self.assertEqual(ev['ant'], -1)
f0a42b33 320 self.assertIs(type(ev['ant']), bt2_field._SignedIntegerFieldConst)
2ae9f48c 321 self.assertEqual(ev['msg'], 'hellooo')
9cf643d1 322 self.assertEqual(ev['cpu_id'], 1)
f0a42b33 323 self.assertIs(type(ev['cpu_id']), bt2_field._SignedIntegerFieldConst)
2ae9f48c
SM
324 self.assertEqual(ev['stuff'], 13.194)
325
326 # Test packet fields
9cf643d1 327 self.assertEqual(ev['something'], 154)
f0a42b33 328 self.assertIs(type(ev['something']), bt2_field._UnsignedIntegerFieldConst)
2ae9f48c 329 self.assertEqual(ev['something_else'], 17.2)
9cf643d1
PP
330
331 with self.assertRaises(KeyError):
332 ev['yes']
333
f0a42b33
FD
334 def test_getitem(self):
335 msg = utils.get_event_message()
336 ev = msg.event
337 self.assertEqual(ev['giraffe'], 1)
338 self.assertIs(type(ev['giraffe']), bt2_field._SignedIntegerField)
339 self.assertEqual(ev['ant'], -1)
340 self.assertIs(type(ev['ant']), bt2_field._SignedIntegerField)
341 self.assertEqual(ev['cpu_id'], 1)
342 self.assertIs(type(ev['cpu_id']), bt2_field._SignedIntegerField)
343 self.assertEqual(ev['something'], 154)
344 self.assertIs(type(ev['something']), bt2_field._UnsignedIntegerField)
345
2ae9f48c
SM
346
347if __name__ == "__main__":
348 unittest.main()
This page took 0.07102 seconds and 4 git commands to generate.