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.
19 from collections
import OrderedDict
22 from utils
import TestOutputPortMessageIterator
25 class EventTestCase(unittest
.TestCase
):
26 def _create_test_event_message(
28 packet_fields_config
=None,
29 event_fields_config
=None,
30 with_clockclass
=False,
36 class MyIter(bt2
._UserMessageIterator
):
37 def __init__(self
, self_output_port
):
39 self
._msgs
= [self
._create
_stream
_beginning
_message
(test_obj
.stream
)]
42 assert test_obj
.packet
44 self
._create
_packet
_beginning
_message
(test_obj
.packet
)
47 default_clock_snapshot
= 789 if with_clockclass
else None
50 assert test_obj
.packet
51 ev_parent
= test_obj
.packet
53 assert test_obj
.stream
54 ev_parent
= test_obj
.stream
56 msg
= self
._create
_event
_message
(
57 test_obj
.event_class
, ev_parent
, default_clock_snapshot
60 if event_fields_config
is not None:
61 event_fields_config(msg
.event
)
63 self
._msgs
.append(msg
)
66 self
._msgs
.append(self
._create
_packet
_end
_message
(test_obj
.packet
))
68 self
._msgs
.append(self
._create
_stream
_end
_message
(test_obj
.stream
))
71 if self
._at
== len(self
._msgs
):
74 msg
= self
._msgs
[self
._at
]
78 class MySrc(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
79 def __init__(self
, params
, obj
):
80 self
._add
_output
_port
('out')
81 tc
= self
._create
_trace
_class
()
85 clock_class
= self
._create
_clock
_class
(frequency
=1000)
87 # event common context (stream-class-defined)
90 cc
= tc
.create_structure_field_class()
92 ('cpu_id', tc
.create_signed_integer_field_class(8)),
93 ('stuff', tc
.create_real_field_class()),
96 # packet context (stream-class-defined)
100 pc
= tc
.create_structure_field_class()
102 ('something', tc
.create_unsigned_integer_field_class(8)),
103 ('something_else', tc
.create_real_field_class()),
106 stream_class
= tc
.create_stream_class(
107 default_clock_class
=clock_class
,
108 event_common_context_field_class
=cc
,
109 packet_context_field_class
=pc
,
110 supports_packets
=with_packet
,
113 # specific context (event-class-defined)
116 sc
= tc
.create_structure_field_class()
118 ('ant', tc
.create_signed_integer_field_class(16)),
119 ('msg', tc
.create_string_field_class()),
125 ep
= tc
.create_structure_field_class()
127 ('giraffe', tc
.create_signed_integer_field_class(32)),
128 ('gnu', tc
.create_signed_integer_field_class(8)),
129 ('mosquito', tc
.create_signed_integer_field_class(8)),
132 event_class
= stream_class
.create_event_class(
134 specific_context_field_class
=sc
,
135 payload_field_class
=ep
,
139 stream
= trace
.create_stream(stream_class
)
142 packet
= stream
.create_packet()
144 if packet_fields_config
is not None:
146 packet_fields_config(packet
)
149 test_obj
.packet
= packet
151 test_obj
.stream
= stream
152 test_obj
.event_class
= event_class
155 self
._graph
= bt2
.Graph()
156 self
._src
_comp
= self
._graph
.add_component(MySrc
, 'my_source')
157 self
._msg
_iter
= TestOutputPortMessageIterator(
158 self
._graph
, self
._src
_comp
.output_ports
['out']
161 for msg
in self
._msg
_iter
:
162 if type(msg
) is bt2
._EventMessage
:
165 def test_attr_event_class(self
):
166 msg
= self
._create
_test
_event
_message
()
167 self
.assertEqual(msg
.event
.cls
.addr
, self
.event_class
.addr
)
169 def test_attr_name(self
):
170 msg
= self
._create
_test
_event
_message
()
171 self
.assertEqual(msg
.event
.name
, self
.event_class
.name
)
173 def test_attr_id(self
):
174 msg
= self
._create
_test
_event
_message
()
175 self
.assertEqual(msg
.event
.id, self
.event_class
.id)
177 def test_get_common_context_field(self
):
178 def event_fields_config(event
):
179 event
.common_context_field
['cpu_id'] = 1
180 event
.common_context_field
['stuff'] = 13.194
182 msg
= self
._create
_test
_event
_message
(
183 event_fields_config
=event_fields_config
, with_cc
=True
186 self
.assertEqual(msg
.event
.common_context_field
['cpu_id'], 1)
187 self
.assertEqual(msg
.event
.common_context_field
['stuff'], 13.194)
189 def test_no_common_context_field(self
):
190 msg
= self
._create
_test
_event
_message
(with_cc
=False)
191 self
.assertIsNone(msg
.event
.common_context_field
)
193 def test_get_specific_context_field(self
):
194 def event_fields_config(event
):
195 event
.specific_context_field
['ant'] = -1
196 event
.specific_context_field
['msg'] = 'hellooo'
198 msg
= self
._create
_test
_event
_message
(
199 event_fields_config
=event_fields_config
, with_sc
=True
202 self
.assertEqual(msg
.event
.specific_context_field
['ant'], -1)
203 self
.assertEqual(msg
.event
.specific_context_field
['msg'], 'hellooo')
205 def test_no_specific_context_field(self
):
206 msg
= self
._create
_test
_event
_message
(with_sc
=False)
207 self
.assertIsNone(msg
.event
.specific_context_field
)
209 def test_get_event_payload_field(self
):
210 def event_fields_config(event
):
211 event
.payload_field
['giraffe'] = 1
212 event
.payload_field
['gnu'] = 23
213 event
.payload_field
['mosquito'] = 42
215 msg
= self
._create
_test
_event
_message
(
216 event_fields_config
=event_fields_config
, with_ep
=True
219 self
.assertEqual(msg
.event
.payload_field
['giraffe'], 1)
220 self
.assertEqual(msg
.event
.payload_field
['gnu'], 23)
221 self
.assertEqual(msg
.event
.payload_field
['mosquito'], 42)
223 def test_no_payload_field(self
):
224 msg
= self
._create
_test
_event
_message
(with_ep
=False)
225 self
.assertIsNone(msg
.event
.payload_field
)
227 def test_clock_value(self
):
228 msg
= self
._create
_test
_event
_message
(with_clockclass
=True)
229 self
.assertEqual(msg
.default_clock_snapshot
.value
, 789)
231 def test_no_clock_value(self
):
232 msg
= self
._create
_test
_event
_message
(with_clockclass
=False)
233 with self
.assertRaisesRegex(
234 ValueError, 'stream class has no default clock class'
236 msg
.default_clock_snapshot
238 def test_stream(self
):
239 msg
= self
._create
_test
_event
_message
()
240 self
.assertEqual(msg
.event
.stream
.addr
, self
.stream
.addr
)
242 def test_getitem(self
):
243 def event_fields_config(event
):
244 event
.payload_field
['giraffe'] = 1
245 event
.payload_field
['gnu'] = 23
246 event
.payload_field
['mosquito'] = 42
247 event
.specific_context_field
['ant'] = -1
248 event
.specific_context_field
['msg'] = 'hellooo'
249 event
.common_context_field
['cpu_id'] = 1
250 event
.common_context_field
['stuff'] = 13.194
252 def packet_fields_config(packet
):
253 packet
.context_field
['something'] = 154
254 packet
.context_field
['something_else'] = 17.2
256 msg
= self
._create
_test
_event
_message
(
257 packet_fields_config
=packet_fields_config
,
258 event_fields_config
=event_fields_config
,
267 self
.assertEqual(ev
['giraffe'], 1)
268 self
.assertEqual(ev
['gnu'], 23)
269 self
.assertEqual(ev
['mosquito'], 42)
270 self
.assertEqual(ev
['ant'], -1)
271 self
.assertEqual(ev
['msg'], 'hellooo')
272 self
.assertEqual(ev
['cpu_id'], 1)
273 self
.assertEqual(ev
['stuff'], 13.194)
276 self
.assertEqual(ev
['something'], 154)
277 self
.assertEqual(ev
['something_else'], 17.2)
279 with self
.assertRaises(KeyError):
283 if __name__
== "__main__":