1 # SPDX-License-Identifier: GPL-2.0-only
3 # Copyright (C) 2019 EfficiOS Inc.
10 from bt2
import field
as bt2_field
11 from bt2
import stream
as bt2_stream
12 from bt2
import event_class
as bt2_event_class
13 from bt2
import clock_snapshot
as bt2_clock_snapshot
14 from utils
import TestOutputPortMessageIterator
17 class EventTestCase(unittest
.TestCase
):
18 def _create_test_const_event_message(
20 packet_fields_config
=None,
21 event_fields_config
=None,
22 with_clockclass
=False,
28 class MyIter(bt2
._UserMessageIterator
):
29 def __init__(self
, config
, self_output_port
):
31 self
._msgs
= [self
._create
_stream
_beginning
_message
(test_obj
.stream
)]
34 assert test_obj
.packet
36 self
._create
_packet
_beginning
_message
(test_obj
.packet
)
39 default_clock_snapshot
= 789 if with_clockclass
else None
42 assert test_obj
.packet
43 ev_parent
= test_obj
.packet
45 assert test_obj
.stream
46 ev_parent
= test_obj
.stream
48 msg
= self
._create
_event
_message
(
49 test_obj
.event_class
, ev_parent
, default_clock_snapshot
52 if event_fields_config
is not None:
53 event_fields_config(msg
.event
)
55 self
._msgs
.append(msg
)
58 self
._msgs
.append(self
._create
_packet
_end
_message
(test_obj
.packet
))
60 self
._msgs
.append(self
._create
_stream
_end
_message
(test_obj
.stream
))
63 if self
._at
== len(self
._msgs
):
66 msg
= self
._msgs
[self
._at
]
70 class MySrc(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
71 def __init__(self
, config
, params
, obj
):
72 self
._add
_output
_port
("out")
73 tc
= self
._create
_trace
_class
()
77 clock_class
= self
._create
_clock
_class
(frequency
=1000)
79 # event common context (stream-class-defined)
82 cc
= tc
.create_structure_field_class()
84 ("cpu_id", tc
.create_signed_integer_field_class(8)),
85 ("stuff", tc
.create_double_precision_real_field_class()),
86 ("gnu", tc
.create_string_field_class()),
89 # packet context (stream-class-defined)
93 pc
= tc
.create_structure_field_class()
95 ("something", tc
.create_unsigned_integer_field_class(8)),
98 tc
.create_double_precision_real_field_class(),
102 stream_class
= tc
.create_stream_class(
103 default_clock_class
=clock_class
,
104 event_common_context_field_class
=cc
,
105 packet_context_field_class
=pc
,
106 supports_packets
=with_packet
,
109 # specific context (event-class-defined)
112 sc
= tc
.create_structure_field_class()
114 ("ant", tc
.create_signed_integer_field_class(16)),
115 ("msg", tc
.create_string_field_class()),
121 ep
= tc
.create_structure_field_class()
123 ("giraffe", tc
.create_signed_integer_field_class(32)),
124 ("gnu", tc
.create_signed_integer_field_class(8)),
125 ("mosquito", tc
.create_signed_integer_field_class(8)),
128 event_class
= stream_class
.create_event_class(
130 specific_context_field_class
=sc
,
131 payload_field_class
=ep
,
135 stream
= trace
.create_stream(stream_class
)
138 packet
= stream
.create_packet()
140 if packet_fields_config
is not None:
142 packet_fields_config(packet
)
145 test_obj
.packet
= packet
147 test_obj
.stream
= stream
148 test_obj
.event_class
= event_class
151 self
._graph
= bt2
.Graph()
152 self
._src
_comp
= self
._graph
.add_component(MySrc
, "my_source")
153 self
._msg
_iter
= TestOutputPortMessageIterator(
154 self
._graph
, self
._src
_comp
.output_ports
["out"]
157 for msg
in self
._msg
_iter
:
158 if type(msg
) is bt2
._EventMessageConst
:
159 self
._event
_msg
= msg
162 def test_const_attr_event_class(self
):
163 msg
= self
._create
_test
_const
_event
_message
()
164 self
.assertEqual(msg
.event
.cls
.addr
, self
.event_class
.addr
)
165 self
.assertIs(type(msg
.event
.cls
), bt2_event_class
._EventClassConst
)
167 def test_attr_event_class(self
):
168 msg
= utils
.get_event_message()
169 self
.assertIs(type(msg
.event
.cls
), bt2_event_class
._EventClass
)
171 def test_const_attr_name(self
):
172 msg
= self
._create
_test
_const
_event
_message
()
173 self
.assertEqual(msg
.event
.name
, self
.event_class
.name
)
175 def test_const_attr_id(self
):
176 msg
= self
._create
_test
_const
_event
_message
()
177 self
.assertEqual(msg
.event
.id, self
.event_class
.id)
179 def test_const_get_common_context_field(self
):
180 def event_fields_config(event
):
181 event
.common_context_field
["cpu_id"] = 1
182 event
.common_context_field
["stuff"] = 13.194
183 event
.common_context_field
["gnu"] = "salut"
185 msg
= self
._create
_test
_const
_event
_message
(
186 event_fields_config
=event_fields_config
, with_cc
=True
189 self
.assertEqual(msg
.event
.common_context_field
["cpu_id"], 1)
190 self
.assertEqual(msg
.event
.common_context_field
["stuff"], 13.194)
191 self
.assertEqual(msg
.event
.common_context_field
["gnu"], "salut")
193 type(msg
.event
.common_context_field
), bt2_field
._StructureFieldConst
196 def test_attr_common_context_field(self
):
197 msg
= utils
.get_event_message()
198 self
.assertIs(type(msg
.event
.common_context_field
), bt2_field
._StructureField
)
200 def test_const_no_common_context_field(self
):
201 msg
= self
._create
_test
_const
_event
_message
(with_cc
=False)
202 self
.assertIsNone(msg
.event
.common_context_field
)
204 def test_const_get_specific_context_field(self
):
205 def event_fields_config(event
):
206 event
.specific_context_field
["ant"] = -1
207 event
.specific_context_field
["msg"] = "hellooo"
209 msg
= self
._create
_test
_const
_event
_message
(
210 event_fields_config
=event_fields_config
, with_sc
=True
213 self
.assertEqual(msg
.event
.specific_context_field
["ant"], -1)
214 self
.assertEqual(msg
.event
.specific_context_field
["msg"], "hellooo")
216 type(msg
.event
.specific_context_field
), bt2_field
._StructureFieldConst
219 def test_attr_specific_context_field(self
):
220 msg
= utils
.get_event_message()
221 self
.assertIs(type(msg
.event
.specific_context_field
), bt2_field
._StructureField
)
223 def test_const_no_specific_context_field(self
):
224 msg
= self
._create
_test
_const
_event
_message
(with_sc
=False)
225 self
.assertIsNone(msg
.event
.specific_context_field
)
227 def test_const_get_event_payload_field(self
):
228 def event_fields_config(event
):
229 event
.payload_field
["giraffe"] = 1
230 event
.payload_field
["gnu"] = 23
231 event
.payload_field
["mosquito"] = 42
233 msg
= self
._create
_test
_const
_event
_message
(
234 event_fields_config
=event_fields_config
, with_ep
=True
237 self
.assertEqual(msg
.event
.payload_field
["giraffe"], 1)
238 self
.assertEqual(msg
.event
.payload_field
["gnu"], 23)
239 self
.assertEqual(msg
.event
.payload_field
["mosquito"], 42)
240 self
.assertIs(type(msg
.event
.payload_field
), bt2_field
._StructureFieldConst
)
242 def test_attr_payload_field(self
):
243 msg
= utils
.get_event_message()
244 self
.assertIs(type(msg
.event
.payload_field
), bt2_field
._StructureField
)
246 def test_const_no_payload_field(self
):
247 msg
= self
._create
_test
_const
_event
_message
(with_ep
=False)
248 self
.assertIsNone(msg
.event
.payload_field
)
250 def test_const_clock_value(self
):
251 msg
= self
._create
_test
_const
_event
_message
(with_clockclass
=True)
252 self
.assertEqual(msg
.default_clock_snapshot
.value
, 789)
254 type(msg
.default_clock_snapshot
), bt2_clock_snapshot
._ClockSnapshotConst
257 def test_clock_value(self
):
258 msg
= utils
.get_event_message()
259 self
.assertEqual(msg
.default_clock_snapshot
.value
, 789)
261 type(msg
.default_clock_snapshot
), bt2_clock_snapshot
._ClockSnapshotConst
264 def test_const_no_clock_value(self
):
265 msg
= self
._create
_test
_const
_event
_message
(with_clockclass
=False)
266 with self
.assertRaisesRegex(
267 ValueError, "stream class has no default clock class"
269 msg
.default_clock_snapshot
271 def test_const_stream(self
):
272 msg
= self
._create
_test
_const
_event
_message
()
273 self
.assertEqual(msg
.event
.stream
.addr
, self
.stream
.addr
)
274 self
.assertIs(type(msg
.event
.stream
), bt2_stream
._StreamConst
)
276 def test_stream(self
):
277 msg
= utils
.get_event_message()
278 self
.assertIs(type(msg
.event
.stream
), bt2_stream
._Stream
)
281 def _event_payload_fields_config(event
):
282 event
.payload_field
["giraffe"] = 1
283 event
.payload_field
["gnu"] = 23
284 event
.payload_field
["mosquito"] = 42
287 def _event_fields_config(event
):
288 EventTestCase
._event
_payload
_fields
_config
(event
)
289 event
.specific_context_field
["ant"] = -1
290 event
.specific_context_field
["msg"] = "hellooo"
291 event
.common_context_field
["cpu_id"] = 1
292 event
.common_context_field
["stuff"] = 13.194
293 event
.common_context_field
["gnu"] = "salut"
296 def _packet_fields_config(packet
):
297 packet
.context_field
["something"] = 154
298 packet
.context_field
["something_else"] = 17.2
300 def test_const_getitem(self
):
301 msg
= self
._create
_test
_const
_event
_message
(
302 packet_fields_config
=self
._packet
_fields
_config
,
303 event_fields_config
=self
._event
_fields
_config
,
312 self
.assertEqual(ev
["giraffe"], 1)
313 self
.assertIs(type(ev
["giraffe"]), bt2_field
._SignedIntegerFieldConst
)
314 self
.assertEqual(ev
["gnu"], 23)
315 self
.assertEqual(ev
["mosquito"], 42)
316 self
.assertEqual(ev
["ant"], -1)
317 self
.assertIs(type(ev
["ant"]), bt2_field
._SignedIntegerFieldConst
)
318 self
.assertEqual(ev
["msg"], "hellooo")
319 self
.assertEqual(ev
["cpu_id"], 1)
320 self
.assertIs(type(ev
["cpu_id"]), bt2_field
._SignedIntegerFieldConst
)
321 self
.assertEqual(ev
["stuff"], 13.194)
324 self
.assertEqual(ev
["something"], 154)
325 self
.assertIs(type(ev
["something"]), bt2_field
._UnsignedIntegerFieldConst
)
326 self
.assertEqual(ev
["something_else"], 17.2)
328 with self
.assertRaises(KeyError):
331 def test_const_getitem_no_packet(self
):
332 msg
= self
._create
_test
_const
_event
_message
(
333 event_fields_config
=self
._event
_payload
_fields
_config
,
338 with self
.assertRaises(KeyError):
341 def test_getitem(self
):
342 msg
= utils
.get_event_message()
344 self
.assertEqual(ev
["giraffe"], 1)
345 self
.assertIs(type(ev
["giraffe"]), bt2_field
._SignedIntegerField
)
346 self
.assertEqual(ev
["ant"], -1)
347 self
.assertIs(type(ev
["ant"]), bt2_field
._SignedIntegerField
)
348 self
.assertEqual(ev
["cpu_id"], 1)
349 self
.assertIs(type(ev
["cpu_id"]), bt2_field
._SignedIntegerField
)
350 self
.assertEqual(ev
["something"], 154)
351 self
.assertIs(type(ev
["something"]), bt2_field
._UnsignedIntegerField
)
353 def test_iter_full(self
):
354 msg
= self
._create
_test
_const
_event
_message
(
355 packet_fields_config
=self
._packet
_fields
_config
,
356 event_fields_config
=self
._event
_fields
_config
,
362 expected_field_names
= [
377 self
.assertEqual(list(msg
.event
), expected_field_names
)
379 def test_iter_payload_only(self
):
380 msg
= self
._create
_test
_const
_event
_message
(
381 event_fields_config
=self
._event
_payload
_fields
_config
,
384 expected_field_names
= [
390 self
.assertEqual(list(msg
.event
), expected_field_names
)
392 def test_len_full(self
):
393 msg
= self
._create
_test
_const
_event
_message
(
394 packet_fields_config
=self
._packet
_fields
_config
,
395 event_fields_config
=self
._event
_fields
_config
,
401 self
.assertEqual(len(msg
.event
), 9)
403 def test_len_payload_only(self
):
404 msg
= self
._create
_test
_const
_event
_message
(
405 packet_fields_config
=None,
406 event_fields_config
=self
._event
_payload
_fields
_config
,
409 self
.assertEqual(len(msg
.event
), 3)
411 def test_in_full(self
):
412 msg
= self
._create
_test
_const
_event
_message
(
413 packet_fields_config
=self
._packet
_fields
_config
,
414 event_fields_config
=self
._event
_fields
_config
,
436 for field_name
in field_names
:
437 self
.assertTrue(field_name
in msg
.event
)
439 self
.assertFalse("lol" in msg
.event
)
441 def test_in_payload_only(self
):
442 msg
= self
._create
_test
_const
_event
_message
(
443 packet_fields_config
=None,
444 event_fields_config
=self
._event
_payload
_fields
_config
,
453 for field_name
in field_names
:
454 self
.assertTrue(field_name
in msg
.event
)
456 self
.assertFalse("lol" in msg
.event
)
459 if __name__
== "__main__":