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()),
101 # packet context (stream-class-defined)
105 pc
= tc
.create_structure_field_class()
107 ('something', tc
.create_unsigned_integer_field_class(8)),
110 tc
.create_double_precision_real_field_class(),
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
,
121 # specific context (event-class-defined)
124 sc
= tc
.create_structure_field_class()
126 ('ant', tc
.create_signed_integer_field_class(16)),
127 ('msg', tc
.create_string_field_class()),
133 ep
= tc
.create_structure_field_class()
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)),
140 event_class
= stream_class
.create_event_class(
142 specific_context_field_class
=sc
,
143 payload_field_class
=ep
,
147 stream
= trace
.create_stream(stream_class
)
150 packet
= stream
.create_packet()
152 if packet_fields_config
is not None:
154 packet_fields_config(packet
)
157 test_obj
.packet
= packet
159 test_obj
.stream
= stream
160 test_obj
.event_class
= event_class
163 self
._graph
= bt2
.Graph()
164 self
._src
_comp
= self
._graph
.add_component(MySrc
, 'my_source')
165 self
._msg
_iter
= TestOutputPortMessageIterator(
166 self
._graph
, self
._src
_comp
.output_ports
['out']
169 for msg
in self
._msg
_iter
:
170 if type(msg
) is bt2
._EventMessageConst
:
171 self
._event
_msg
= msg
174 def test_const_attr_event_class(self
):
175 msg
= self
._create
_test
_const
_event
_message
()
176 self
.assertEqual(msg
.event
.cls
.addr
, self
.event_class
.addr
)
177 self
.assertIs(type(msg
.event
.cls
), bt2_event_class
._EventClassConst
)
179 def test_attr_event_class(self
):
180 msg
= utils
.get_event_message()
181 self
.assertIs(type(msg
.event
.cls
), bt2_event_class
._EventClass
)
183 def test_const_attr_name(self
):
184 msg
= self
._create
_test
_const
_event
_message
()
185 self
.assertEqual(msg
.event
.name
, self
.event_class
.name
)
187 def test_const_attr_id(self
):
188 msg
= self
._create
_test
_const
_event
_message
()
189 self
.assertEqual(msg
.event
.id, self
.event_class
.id)
191 def test_const_get_common_context_field(self
):
192 def event_fields_config(event
):
193 event
.common_context_field
['cpu_id'] = 1
194 event
.common_context_field
['stuff'] = 13.194
196 msg
= self
._create
_test
_const
_event
_message
(
197 event_fields_config
=event_fields_config
, with_cc
=True
200 self
.assertEqual(msg
.event
.common_context_field
['cpu_id'], 1)
201 self
.assertEqual(msg
.event
.common_context_field
['stuff'], 13.194)
203 type(msg
.event
.common_context_field
), bt2_field
._StructureFieldConst
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
)
210 def test_const_no_common_context_field(self
):
211 msg
= self
._create
_test
_const
_event
_message
(with_cc
=False)
212 self
.assertIsNone(msg
.event
.common_context_field
)
214 def test_const_get_specific_context_field(self
):
215 def event_fields_config(event
):
216 event
.specific_context_field
['ant'] = -1
217 event
.specific_context_field
['msg'] = 'hellooo'
219 msg
= self
._create
_test
_const
_event
_message
(
220 event_fields_config
=event_fields_config
, with_sc
=True
223 self
.assertEqual(msg
.event
.specific_context_field
['ant'], -1)
224 self
.assertEqual(msg
.event
.specific_context_field
['msg'], 'hellooo')
226 type(msg
.event
.specific_context_field
), bt2_field
._StructureFieldConst
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
)
233 def test_const_no_specific_context_field(self
):
234 msg
= self
._create
_test
_const
_event
_message
(with_sc
=False)
235 self
.assertIsNone(msg
.event
.specific_context_field
)
237 def test_const_get_event_payload_field(self
):
238 def event_fields_config(event
):
239 event
.payload_field
['giraffe'] = 1
240 event
.payload_field
['gnu'] = 23
241 event
.payload_field
['mosquito'] = 42
243 msg
= self
._create
_test
_const
_event
_message
(
244 event_fields_config
=event_fields_config
, with_ep
=True
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)
250 self
.assertIs(type(msg
.event
.payload_field
), bt2_field
._StructureFieldConst
)
252 def test_attr_payload_field(self
):
253 msg
= utils
.get_event_message()
254 self
.assertIs(type(msg
.event
.payload_field
), bt2_field
._StructureField
)
256 def test_const_no_payload_field(self
):
257 msg
= self
._create
_test
_const
_event
_message
(with_ep
=False)
258 self
.assertIsNone(msg
.event
.payload_field
)
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)
264 type(msg
.default_clock_snapshot
), bt2_clock_snapshot
._ClockSnapshotConst
267 def test_clock_value(self
):
268 msg
= utils
.get_event_message()
269 self
.assertEqual(msg
.default_clock_snapshot
.value
, 789)
271 type(msg
.default_clock_snapshot
), bt2_clock_snapshot
._ClockSnapshotConst
274 def test_const_no_clock_value(self
):
275 msg
= self
._create
_test
_const
_event
_message
(with_clockclass
=False)
276 with self
.assertRaisesRegex(
277 ValueError, 'stream class has no default clock class'
279 msg
.default_clock_snapshot
281 def test_const_stream(self
):
282 msg
= self
._create
_test
_const
_event
_message
()
283 self
.assertEqual(msg
.event
.stream
.addr
, self
.stream
.addr
)
284 self
.assertIs(type(msg
.event
.stream
), bt2_stream
._StreamConst
)
286 def test_stream(self
):
287 msg
= utils
.get_event_message()
288 self
.assertIs(type(msg
.event
.stream
), bt2_stream
._Stream
)
290 def test_const_getitem(self
):
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
300 def packet_fields_config(packet
):
301 packet
.context_field
['something'] = 154
302 packet
.context_field
['something_else'] = 17.2
304 msg
= self
._create
_test
_const
_event
_message
(
305 packet_fields_config
=packet_fields_config
,
306 event_fields_config
=event_fields_config
,
315 self
.assertEqual(ev
['giraffe'], 1)
316 self
.assertIs(type(ev
['giraffe']), bt2_field
._SignedIntegerFieldConst
)
317 self
.assertEqual(ev
['gnu'], 23)
318 self
.assertEqual(ev
['mosquito'], 42)
319 self
.assertEqual(ev
['ant'], -1)
320 self
.assertIs(type(ev
['ant']), bt2_field
._SignedIntegerFieldConst
)
321 self
.assertEqual(ev
['msg'], 'hellooo')
322 self
.assertEqual(ev
['cpu_id'], 1)
323 self
.assertIs(type(ev
['cpu_id']), bt2_field
._SignedIntegerFieldConst
)
324 self
.assertEqual(ev
['stuff'], 13.194)
327 self
.assertEqual(ev
['something'], 154)
328 self
.assertIs(type(ev
['something']), bt2_field
._UnsignedIntegerFieldConst
)
329 self
.assertEqual(ev
['something_else'], 17.2)
331 with self
.assertRaises(KeyError):
334 def test_getitem(self
):
335 msg
= utils
.get_event_message()
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
)
347 if __name__
== "__main__":