2 # Copyright (C) 2019 EfficiOS Inc.
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
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.
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.
22 from utils
import TestOutputPortMessageIterator
24 from bt2
import field
as bt2_field
25 from bt2
import stream
as bt2_stream
26 from bt2
import event_class
as bt2_event_class
27 from bt2
import clock_snapshot
as bt2_clock_snapshot
30 class EventTestCase(unittest
.TestCase
):
31 def _create_test_const_event_message(
33 packet_fields_config
=None,
34 event_fields_config
=None,
35 with_clockclass
=False,
41 class MyIter(bt2
._UserMessageIterator
):
42 def __init__(self
, config
, self_output_port
):
44 self
._msgs
= [self
._create
_stream
_beginning
_message
(test_obj
.stream
)]
47 assert test_obj
.packet
49 self
._create
_packet
_beginning
_message
(test_obj
.packet
)
52 default_clock_snapshot
= 789 if with_clockclass
else None
55 assert test_obj
.packet
56 ev_parent
= test_obj
.packet
58 assert test_obj
.stream
59 ev_parent
= test_obj
.stream
61 msg
= self
._create
_event
_message
(
62 test_obj
.event_class
, ev_parent
, default_clock_snapshot
65 if event_fields_config
is not None:
66 event_fields_config(msg
.event
)
68 self
._msgs
.append(msg
)
71 self
._msgs
.append(self
._create
_packet
_end
_message
(test_obj
.packet
))
73 self
._msgs
.append(self
._create
_stream
_end
_message
(test_obj
.stream
))
76 if self
._at
== len(self
._msgs
):
79 msg
= self
._msgs
[self
._at
]
83 class MySrc(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
84 def __init__(self
, config
, params
, obj
):
85 self
._add
_output
_port
('out')
86 tc
= self
._create
_trace
_class
()
90 clock_class
= self
._create
_clock
_class
(frequency
=1000)
92 # event common context (stream-class-defined)
95 cc
= tc
.create_structure_field_class()
97 ('cpu_id', tc
.create_signed_integer_field_class(8)),
98 ('stuff', tc
.create_double_precision_real_field_class()),
99 ('gnu', tc
.create_string_field_class()),
102 # packet context (stream-class-defined)
106 pc
= tc
.create_structure_field_class()
108 ('something', tc
.create_unsigned_integer_field_class(8)),
111 tc
.create_double_precision_real_field_class(),
115 stream_class
= tc
.create_stream_class(
116 default_clock_class
=clock_class
,
117 event_common_context_field_class
=cc
,
118 packet_context_field_class
=pc
,
119 supports_packets
=with_packet
,
122 # specific context (event-class-defined)
125 sc
= tc
.create_structure_field_class()
127 ('ant', tc
.create_signed_integer_field_class(16)),
128 ('msg', tc
.create_string_field_class()),
134 ep
= tc
.create_structure_field_class()
136 ('giraffe', tc
.create_signed_integer_field_class(32)),
137 ('gnu', tc
.create_signed_integer_field_class(8)),
138 ('mosquito', tc
.create_signed_integer_field_class(8)),
141 event_class
= stream_class
.create_event_class(
143 specific_context_field_class
=sc
,
144 payload_field_class
=ep
,
148 stream
= trace
.create_stream(stream_class
)
151 packet
= stream
.create_packet()
153 if packet_fields_config
is not None:
155 packet_fields_config(packet
)
158 test_obj
.packet
= packet
160 test_obj
.stream
= stream
161 test_obj
.event_class
= event_class
164 self
._graph
= bt2
.Graph()
165 self
._src
_comp
= self
._graph
.add_component(MySrc
, 'my_source')
166 self
._msg
_iter
= TestOutputPortMessageIterator(
167 self
._graph
, self
._src
_comp
.output_ports
['out']
170 for msg
in self
._msg
_iter
:
171 if type(msg
) is bt2
._EventMessageConst
:
172 self
._event
_msg
= msg
175 def test_const_attr_event_class(self
):
176 msg
= self
._create
_test
_const
_event
_message
()
177 self
.assertEqual(msg
.event
.cls
.addr
, self
.event_class
.addr
)
178 self
.assertIs(type(msg
.event
.cls
), bt2_event_class
._EventClassConst
)
180 def test_attr_event_class(self
):
181 msg
= utils
.get_event_message()
182 self
.assertIs(type(msg
.event
.cls
), bt2_event_class
._EventClass
)
184 def test_const_attr_name(self
):
185 msg
= self
._create
_test
_const
_event
_message
()
186 self
.assertEqual(msg
.event
.name
, self
.event_class
.name
)
188 def test_const_attr_id(self
):
189 msg
= self
._create
_test
_const
_event
_message
()
190 self
.assertEqual(msg
.event
.id, self
.event_class
.id)
192 def test_const_get_common_context_field(self
):
193 def event_fields_config(event
):
194 event
.common_context_field
['cpu_id'] = 1
195 event
.common_context_field
['stuff'] = 13.194
196 event
.common_context_field
['gnu'] = 'salut'
198 msg
= self
._create
_test
_const
_event
_message
(
199 event_fields_config
=event_fields_config
, with_cc
=True
202 self
.assertEqual(msg
.event
.common_context_field
['cpu_id'], 1)
203 self
.assertEqual(msg
.event
.common_context_field
['stuff'], 13.194)
204 self
.assertEqual(msg
.event
.common_context_field
['gnu'], 'salut')
206 type(msg
.event
.common_context_field
), bt2_field
._StructureFieldConst
209 def test_attr_common_context_field(self
):
210 msg
= utils
.get_event_message()
211 self
.assertIs(type(msg
.event
.common_context_field
), bt2_field
._StructureField
)
213 def test_const_no_common_context_field(self
):
214 msg
= self
._create
_test
_const
_event
_message
(with_cc
=False)
215 self
.assertIsNone(msg
.event
.common_context_field
)
217 def test_const_get_specific_context_field(self
):
218 def event_fields_config(event
):
219 event
.specific_context_field
['ant'] = -1
220 event
.specific_context_field
['msg'] = 'hellooo'
222 msg
= self
._create
_test
_const
_event
_message
(
223 event_fields_config
=event_fields_config
, with_sc
=True
226 self
.assertEqual(msg
.event
.specific_context_field
['ant'], -1)
227 self
.assertEqual(msg
.event
.specific_context_field
['msg'], 'hellooo')
229 type(msg
.event
.specific_context_field
), bt2_field
._StructureFieldConst
232 def test_attr_specific_context_field(self
):
233 msg
= utils
.get_event_message()
234 self
.assertIs(type(msg
.event
.specific_context_field
), bt2_field
._StructureField
)
236 def test_const_no_specific_context_field(self
):
237 msg
= self
._create
_test
_const
_event
_message
(with_sc
=False)
238 self
.assertIsNone(msg
.event
.specific_context_field
)
240 def test_const_get_event_payload_field(self
):
241 def event_fields_config(event
):
242 event
.payload_field
['giraffe'] = 1
243 event
.payload_field
['gnu'] = 23
244 event
.payload_field
['mosquito'] = 42
246 msg
= self
._create
_test
_const
_event
_message
(
247 event_fields_config
=event_fields_config
, with_ep
=True
250 self
.assertEqual(msg
.event
.payload_field
['giraffe'], 1)
251 self
.assertEqual(msg
.event
.payload_field
['gnu'], 23)
252 self
.assertEqual(msg
.event
.payload_field
['mosquito'], 42)
253 self
.assertIs(type(msg
.event
.payload_field
), bt2_field
._StructureFieldConst
)
255 def test_attr_payload_field(self
):
256 msg
= utils
.get_event_message()
257 self
.assertIs(type(msg
.event
.payload_field
), bt2_field
._StructureField
)
259 def test_const_no_payload_field(self
):
260 msg
= self
._create
_test
_const
_event
_message
(with_ep
=False)
261 self
.assertIsNone(msg
.event
.payload_field
)
263 def test_const_clock_value(self
):
264 msg
= self
._create
_test
_const
_event
_message
(with_clockclass
=True)
265 self
.assertEqual(msg
.default_clock_snapshot
.value
, 789)
267 type(msg
.default_clock_snapshot
), bt2_clock_snapshot
._ClockSnapshotConst
270 def test_clock_value(self
):
271 msg
= utils
.get_event_message()
272 self
.assertEqual(msg
.default_clock_snapshot
.value
, 789)
274 type(msg
.default_clock_snapshot
), bt2_clock_snapshot
._ClockSnapshotConst
277 def test_const_no_clock_value(self
):
278 msg
= self
._create
_test
_const
_event
_message
(with_clockclass
=False)
279 with self
.assertRaisesRegex(
280 ValueError, 'stream class has no default clock class'
282 msg
.default_clock_snapshot
284 def test_const_stream(self
):
285 msg
= self
._create
_test
_const
_event
_message
()
286 self
.assertEqual(msg
.event
.stream
.addr
, self
.stream
.addr
)
287 self
.assertIs(type(msg
.event
.stream
), bt2_stream
._StreamConst
)
289 def test_stream(self
):
290 msg
= utils
.get_event_message()
291 self
.assertIs(type(msg
.event
.stream
), bt2_stream
._Stream
)
294 def _event_payload_fields_config(event
):
295 event
.payload_field
['giraffe'] = 1
296 event
.payload_field
['gnu'] = 23
297 event
.payload_field
['mosquito'] = 42
300 def _event_fields_config(event
):
301 EventTestCase
._event
_payload
_fields
_config
(event
)
302 event
.specific_context_field
['ant'] = -1
303 event
.specific_context_field
['msg'] = 'hellooo'
304 event
.common_context_field
['cpu_id'] = 1
305 event
.common_context_field
['stuff'] = 13.194
306 event
.common_context_field
['gnu'] = 'salut'
309 def _packet_fields_config(packet
):
310 packet
.context_field
['something'] = 154
311 packet
.context_field
['something_else'] = 17.2
313 def test_const_getitem(self
):
314 msg
= self
._create
_test
_const
_event
_message
(
315 packet_fields_config
=self
._packet
_fields
_config
,
316 event_fields_config
=self
._event
_fields
_config
,
325 self
.assertEqual(ev
['giraffe'], 1)
326 self
.assertIs(type(ev
['giraffe']), bt2_field
._SignedIntegerFieldConst
)
327 self
.assertEqual(ev
['gnu'], 23)
328 self
.assertEqual(ev
['mosquito'], 42)
329 self
.assertEqual(ev
['ant'], -1)
330 self
.assertIs(type(ev
['ant']), bt2_field
._SignedIntegerFieldConst
)
331 self
.assertEqual(ev
['msg'], 'hellooo')
332 self
.assertEqual(ev
['cpu_id'], 1)
333 self
.assertIs(type(ev
['cpu_id']), bt2_field
._SignedIntegerFieldConst
)
334 self
.assertEqual(ev
['stuff'], 13.194)
337 self
.assertEqual(ev
['something'], 154)
338 self
.assertIs(type(ev
['something']), bt2_field
._UnsignedIntegerFieldConst
)
339 self
.assertEqual(ev
['something_else'], 17.2)
341 with self
.assertRaises(KeyError):
344 def test_const_getitem_no_packet(self
):
345 msg
= self
._create
_test
_const
_event
_message
(
346 event_fields_config
=self
._event
_payload
_fields
_config
, with_ep
=True,
350 with self
.assertRaises(KeyError):
353 def test_getitem(self
):
354 msg
= utils
.get_event_message()
356 self
.assertEqual(ev
['giraffe'], 1)
357 self
.assertIs(type(ev
['giraffe']), bt2_field
._SignedIntegerField
)
358 self
.assertEqual(ev
['ant'], -1)
359 self
.assertIs(type(ev
['ant']), bt2_field
._SignedIntegerField
)
360 self
.assertEqual(ev
['cpu_id'], 1)
361 self
.assertIs(type(ev
['cpu_id']), bt2_field
._SignedIntegerField
)
362 self
.assertEqual(ev
['something'], 154)
363 self
.assertIs(type(ev
['something']), bt2_field
._UnsignedIntegerField
)
365 def test_iter_full(self
):
366 msg
= self
._create
_test
_const
_event
_message
(
367 packet_fields_config
=self
._packet
_fields
_config
,
368 event_fields_config
=self
._event
_fields
_config
,
374 expected_field_names
= [
389 self
.assertEqual(list(msg
.event
), expected_field_names
)
391 def test_iter_payload_only(self
):
392 msg
= self
._create
_test
_const
_event
_message
(
393 event_fields_config
=self
._event
_payload
_fields
_config
, with_ep
=True,
395 expected_field_names
= [
401 self
.assertEqual(list(msg
.event
), expected_field_names
)
403 def test_len_full(self
):
404 msg
= self
._create
_test
_const
_event
_message
(
405 packet_fields_config
=self
._packet
_fields
_config
,
406 event_fields_config
=self
._event
_fields
_config
,
412 self
.assertEqual(len(msg
.event
), 9)
414 def test_len_payload_only(self
):
415 msg
= self
._create
_test
_const
_event
_message
(
416 packet_fields_config
=None,
417 event_fields_config
=self
._event
_payload
_fields
_config
,
420 self
.assertEqual(len(msg
.event
), 3)
422 def test_in_full(self
):
423 msg
= self
._create
_test
_const
_event
_message
(
424 packet_fields_config
=self
._packet
_fields
_config
,
425 event_fields_config
=self
._event
_fields
_config
,
447 for field_name
in field_names
:
448 self
.assertTrue(field_name
in msg
.event
)
450 self
.assertFalse('lol' in msg
.event
)
452 def test_in_payload_only(self
):
453 msg
= self
._create
_test
_const
_event
_message
(
454 packet_fields_config
=None,
455 event_fields_config
=self
._event
_payload
_fields
_config
,
464 for field_name
in field_names
:
465 self
.assertTrue(field_name
in msg
.event
)
467 self
.assertFalse('lol' in msg
.event
)
470 if __name__
== "__main__":