ab65e0fa4c270e97be9632e08ce8a30186e106ff
[babeltrace.git] / tests / bindings / python / bt2 / test_event.py
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
19 from collections import OrderedDict
20 import unittest
21 import bt2
22
23
24 class EventTestCase(unittest.TestCase):
25 def _create_test_event_message(
26 self,
27 packet_fields_config=None,
28 event_fields_config=None,
29 with_clockclass=False,
30 with_cc=False,
31 with_sc=False,
32 with_ep=False,
33 with_packet=False,
34 ):
35 class MyIter(bt2._UserMessageIterator):
36 def __init__(self, self_output_port):
37 self._at = 0
38 self._msgs = [self._create_stream_beginning_message(test_obj.stream)]
39
40 if with_packet:
41 assert test_obj.packet
42 self._msgs.append(
43 self._create_packet_beginning_message(test_obj.packet)
44 )
45
46 default_clock_snapshot = 789 if with_clockclass else None
47
48 if with_packet:
49 assert test_obj.packet
50 ev_parent = test_obj.packet
51 else:
52 assert test_obj.stream
53 ev_parent = test_obj.stream
54
55 msg = self._create_event_message(
56 test_obj.event_class, ev_parent, default_clock_snapshot
57 )
58
59 if event_fields_config is not None:
60 event_fields_config(msg.event)
61
62 self._msgs.append(msg)
63
64 if with_packet:
65 self._msgs.append(self._create_packet_end_message(test_obj.packet))
66
67 self._msgs.append(self._create_stream_end_message(test_obj.stream))
68
69 def __next__(self):
70 if self._at == len(self._msgs):
71 raise bt2.Stop
72
73 msg = self._msgs[self._at]
74 self._at += 1
75 return msg
76
77 class MySrc(bt2._UserSourceComponent, message_iterator_class=MyIter):
78 def __init__(self, params):
79 self._add_output_port('out')
80 tc = self._create_trace_class()
81
82 clock_class = None
83 if with_clockclass:
84 clock_class = self._create_clock_class(frequency=1000)
85
86 # event common context (stream-class-defined)
87 cc = None
88 if with_cc:
89 cc = tc.create_structure_field_class()
90 cc += [
91 ('cpu_id', tc.create_signed_integer_field_class(8)),
92 ('stuff', tc.create_real_field_class()),
93 ]
94
95 # packet context (stream-class-defined)
96 pc = None
97
98 if with_packet:
99 pc = tc.create_structure_field_class()
100 pc += [
101 ('something', tc.create_unsigned_integer_field_class(8)),
102 ('something_else', tc.create_real_field_class()),
103 ]
104
105 stream_class = tc.create_stream_class(
106 default_clock_class=clock_class,
107 event_common_context_field_class=cc,
108 packet_context_field_class=pc,
109 supports_packets=with_packet,
110 )
111
112 # specific context (event-class-defined)
113 sc = None
114 if with_sc:
115 sc = tc.create_structure_field_class()
116 sc += [
117 ('ant', tc.create_signed_integer_field_class(16)),
118 ('msg', tc.create_string_field_class()),
119 ]
120
121 # event payload
122 ep = None
123 if with_ep:
124 ep = tc.create_structure_field_class()
125 ep += [
126 ('giraffe', tc.create_signed_integer_field_class(32)),
127 ('gnu', tc.create_signed_integer_field_class(8)),
128 ('mosquito', tc.create_signed_integer_field_class(8)),
129 ]
130
131 event_class = stream_class.create_event_class(
132 name='garou',
133 specific_context_field_class=sc,
134 payload_field_class=ep,
135 )
136
137 trace = tc()
138 stream = trace.create_stream(stream_class)
139
140 if with_packet:
141 packet = stream.create_packet()
142
143 if packet_fields_config is not None:
144 assert packet
145 packet_fields_config(packet)
146
147 if with_packet:
148 test_obj.packet = packet
149
150 test_obj.stream = stream
151 test_obj.event_class = event_class
152
153 test_obj = self
154 self._graph = bt2.Graph()
155 self._src_comp = self._graph.add_component(MySrc, 'my_source')
156 self._msg_iter = self._graph.create_output_port_message_iterator(
157 self._src_comp.output_ports['out']
158 )
159
160 for msg in self._msg_iter:
161 if type(msg) is bt2._EventMessage:
162 return msg
163
164 def test_attr_event_class(self):
165 msg = self._create_test_event_message()
166 self.assertEqual(msg.event.cls.addr, self.event_class.addr)
167
168 def test_attr_name(self):
169 msg = self._create_test_event_message()
170 self.assertEqual(msg.event.name, self.event_class.name)
171
172 def test_attr_id(self):
173 msg = self._create_test_event_message()
174 self.assertEqual(msg.event.id, self.event_class.id)
175
176 def test_get_common_context_field(self):
177 def event_fields_config(event):
178 event.common_context_field['cpu_id'] = 1
179 event.common_context_field['stuff'] = 13.194
180
181 msg = self._create_test_event_message(
182 event_fields_config=event_fields_config, with_cc=True
183 )
184
185 self.assertEqual(msg.event.common_context_field['cpu_id'], 1)
186 self.assertEqual(msg.event.common_context_field['stuff'], 13.194)
187
188 def test_no_common_context_field(self):
189 msg = self._create_test_event_message(with_cc=False)
190 self.assertIsNone(msg.event.common_context_field)
191
192 def test_get_specific_context_field(self):
193 def event_fields_config(event):
194 event.specific_context_field['ant'] = -1
195 event.specific_context_field['msg'] = 'hellooo'
196
197 msg = self._create_test_event_message(
198 event_fields_config=event_fields_config, with_sc=True
199 )
200
201 self.assertEqual(msg.event.specific_context_field['ant'], -1)
202 self.assertEqual(msg.event.specific_context_field['msg'], 'hellooo')
203
204 def test_no_specific_context_field(self):
205 msg = self._create_test_event_message(with_sc=False)
206 self.assertIsNone(msg.event.specific_context_field)
207
208 def test_get_event_payload_field(self):
209 def event_fields_config(event):
210 event.payload_field['giraffe'] = 1
211 event.payload_field['gnu'] = 23
212 event.payload_field['mosquito'] = 42
213
214 msg = self._create_test_event_message(
215 event_fields_config=event_fields_config, with_ep=True
216 )
217
218 self.assertEqual(msg.event.payload_field['giraffe'], 1)
219 self.assertEqual(msg.event.payload_field['gnu'], 23)
220 self.assertEqual(msg.event.payload_field['mosquito'], 42)
221
222 def test_no_payload_field(self):
223 msg = self._create_test_event_message(with_ep=False)
224 self.assertIsNone(msg.event.payload_field)
225
226 def test_clock_value(self):
227 msg = self._create_test_event_message(with_clockclass=True)
228 self.assertEqual(msg.default_clock_snapshot.value, 789)
229
230 def test_no_clock_value(self):
231 msg = self._create_test_event_message(with_clockclass=False)
232 with self.assertRaisesRegex(
233 ValueError, 'stream class has no default clock class'
234 ):
235 msg.default_clock_snapshot
236
237 def test_stream(self):
238 msg = self._create_test_event_message()
239 self.assertEqual(msg.event.stream.addr, self.stream.addr)
240
241 def test_getitem(self):
242 def event_fields_config(event):
243 event.payload_field['giraffe'] = 1
244 event.payload_field['gnu'] = 23
245 event.payload_field['mosquito'] = 42
246 event.specific_context_field['ant'] = -1
247 event.specific_context_field['msg'] = 'hellooo'
248 event.common_context_field['cpu_id'] = 1
249 event.common_context_field['stuff'] = 13.194
250
251 def packet_fields_config(packet):
252 packet.context_field['something'] = 154
253 packet.context_field['something_else'] = 17.2
254
255 msg = self._create_test_event_message(
256 packet_fields_config=packet_fields_config,
257 event_fields_config=event_fields_config,
258 with_cc=True,
259 with_sc=True,
260 with_ep=True,
261 with_packet=True,
262 )
263 ev = msg.event
264
265 # Test event fields
266 self.assertEqual(ev['giraffe'], 1)
267 self.assertEqual(ev['gnu'], 23)
268 self.assertEqual(ev['mosquito'], 42)
269 self.assertEqual(ev['ant'], -1)
270 self.assertEqual(ev['msg'], 'hellooo')
271 self.assertEqual(ev['cpu_id'], 1)
272 self.assertEqual(ev['stuff'], 13.194)
273
274 # Test packet fields
275 self.assertEqual(ev['something'], 154)
276 self.assertEqual(ev['something_else'], 17.2)
277
278 with self.assertRaises(KeyError):
279 ev['yes']
280
281
282 if __name__ == "__main__":
283 unittest.main()
This page took 0.038106 seconds and 3 git commands to generate.