Commit | Line | Data |
---|---|---|
9cf643d1 PP |
1 | from collections import OrderedDict |
2 | from bt2 import values | |
3 | import unittest | |
4 | import copy | |
5 | import bt2 | |
6 | ||
7 | ||
8 | class EventTestCase(unittest.TestCase): | |
9 | def setUp(self): | |
10 | self._ec = self._create_ec() | |
11 | ||
12 | def _create_ec(self, with_eh=True, with_sec=True, with_ec=True, with_ep=True): | |
13 | # event header | |
14 | if with_eh: | |
15 | eh = bt2.StructureFieldType() | |
16 | eh += OrderedDict(( | |
17 | ('id', bt2.IntegerFieldType(8)), | |
18 | ('ts', bt2.IntegerFieldType(32)), | |
19 | )) | |
20 | else: | |
21 | eh = None | |
22 | ||
23 | # stream event context | |
24 | if with_sec: | |
25 | sec = bt2.StructureFieldType() | |
26 | sec += OrderedDict(( | |
27 | ('cpu_id', bt2.IntegerFieldType(8)), | |
28 | ('stuff', bt2.FloatingPointNumberFieldType()), | |
29 | )) | |
30 | else: | |
31 | sec = None | |
32 | ||
33 | # packet context | |
34 | pc = bt2.StructureFieldType() | |
35 | pc += OrderedDict(( | |
36 | ('something', bt2.IntegerFieldType(8)), | |
37 | ('something_else', bt2.FloatingPointNumberFieldType()), | |
38 | )) | |
39 | ||
40 | # stream class | |
41 | sc = bt2.StreamClass() | |
42 | sc.packet_context_field_type = pc | |
43 | sc.event_header_field_type = eh | |
44 | sc.event_context_field_type = sec | |
45 | ||
46 | # event context | |
47 | if with_ec: | |
48 | ec = bt2.StructureFieldType() | |
49 | ec += OrderedDict(( | |
50 | ('ant', bt2.IntegerFieldType(16, is_signed=True)), | |
51 | ('msg', bt2.StringFieldType()), | |
52 | )) | |
53 | else: | |
54 | ec = None | |
55 | ||
56 | # event payload | |
57 | if with_ep: | |
58 | ep = bt2.StructureFieldType() | |
59 | ep += OrderedDict(( | |
60 | ('giraffe', bt2.IntegerFieldType(32)), | |
61 | ('gnu', bt2.IntegerFieldType(8)), | |
62 | ('mosquito', bt2.IntegerFieldType(8)), | |
63 | )) | |
64 | else: | |
65 | ep = None | |
66 | ||
67 | # event class | |
68 | event_class = bt2.EventClass('ec') | |
69 | event_class.context_field_type = ec | |
70 | event_class.payload_field_type = ep | |
71 | sc.add_event_class(event_class) | |
72 | return event_class | |
73 | ||
74 | def test_attr_event_class(self): | |
75 | ev = self._ec() | |
76 | self.assertEqual(ev.event_class.addr, self._ec.addr) | |
77 | ||
78 | def test_attr_name(self): | |
79 | ev = self._ec() | |
80 | self.assertEqual(ev.name, self._ec.name) | |
81 | ||
82 | def test_attr_id(self): | |
83 | ev = self._ec() | |
84 | self.assertEqual(ev.id, self._ec.id) | |
85 | ||
86 | def test_get_event_header_field(self): | |
87 | ev = self._ec() | |
88 | ev.header_field['id'] = 23 | |
89 | ev.header_field['ts'] = 1234 | |
90 | self.assertEqual(ev.header_field['id'], 23) | |
91 | self.assertEqual(ev.header_field['ts'], 1234) | |
92 | ||
93 | def test_set_event_header_field(self): | |
94 | eh = self._ec.stream_class.event_header_field_type() | |
95 | eh['id'] = 17 | |
96 | eh['ts'] = 188 | |
97 | ev = self._ec() | |
98 | ev.header_field = eh | |
99 | self.assertEqual(ev.header_field['id'], 17) | |
100 | self.assertEqual(ev.header_field['ts'], 188) | |
101 | ||
102 | def test_get_stream_event_context_field(self): | |
103 | ev = self._ec() | |
104 | ev.stream_event_context_field['cpu_id'] = 1 | |
105 | ev.stream_event_context_field['stuff'] = 13.194 | |
106 | self.assertEqual(ev.stream_event_context_field['cpu_id'], 1) | |
107 | self.assertEqual(ev.stream_event_context_field['stuff'], 13.194) | |
108 | ||
109 | def test_set_stream_event_context_field(self): | |
110 | sec = self._ec.stream_class.event_context_field_type() | |
111 | sec['cpu_id'] = 2 | |
112 | sec['stuff'] = 19.19 | |
113 | ev = self._ec() | |
114 | ev.stream_event_context_field = sec | |
115 | self.assertEqual(ev.stream_event_context_field['cpu_id'], 2) | |
116 | self.assertEqual(ev.stream_event_context_field['stuff'], 19.19) | |
117 | ||
118 | def test_no_stream_event_context(self): | |
119 | ec = self._create_ec(with_sec=False) | |
120 | ev = ec() | |
121 | self.assertIsNone(ev.stream_event_context_field) | |
122 | ||
123 | def test_get_event_context_field(self): | |
124 | ev = self._ec() | |
125 | ev.context_field['ant'] = -1 | |
126 | ev.context_field['msg'] = 'hellooo' | |
127 | self.assertEqual(ev.context_field['ant'], -1) | |
128 | self.assertEqual(ev.context_field['msg'], 'hellooo') | |
129 | ||
130 | def test_set_event_context_field(self): | |
131 | ec = self._ec.context_field_type() | |
132 | ec['ant'] = 2 | |
133 | ec['msg'] = 'hi there' | |
134 | ev = self._ec() | |
135 | ev.context_field = ec | |
136 | self.assertEqual(ev.context_field['ant'], 2) | |
137 | self.assertEqual(ev.context_field['msg'], 'hi there') | |
138 | ||
139 | def test_no_event_context(self): | |
140 | ec = self._create_ec(with_ec=False) | |
141 | ev = ec() | |
142 | self.assertIsNone(ev.context_field) | |
143 | ||
144 | def test_get_event_payload_field(self): | |
145 | ev = self._ec() | |
146 | ev.payload_field['giraffe'] = 1 | |
147 | ev.payload_field['gnu'] = 23 | |
148 | ev.payload_field['mosquito'] = 42 | |
149 | self.assertEqual(ev.payload_field['giraffe'], 1) | |
150 | self.assertEqual(ev.payload_field['gnu'], 23) | |
151 | self.assertEqual(ev.payload_field['mosquito'], 42) | |
152 | ||
153 | def test_set_event_payload_field(self): | |
154 | ep = self._ec.payload_field_type() | |
155 | ep['giraffe'] = 2 | |
156 | ep['gnu'] = 124 | |
157 | ep['mosquito'] = 17 | |
158 | ev = self._ec() | |
159 | ev.payload_field = ep | |
160 | self.assertEqual(ev.payload_field['giraffe'], 2) | |
161 | self.assertEqual(ev.payload_field['gnu'], 124) | |
162 | self.assertEqual(ev.payload_field['mosquito'], 17) | |
163 | ||
164 | def test_clock_value(self): | |
165 | tc = bt2.Trace() | |
166 | tc.add_stream_class(self._ec.stream_class) | |
167 | cc = bt2.ClockClass('hi') | |
168 | tc.add_clock_class(cc) | |
169 | ev = self._ec() | |
170 | ev.set_clock_value(cc.create_clock_value(177)) | |
171 | self.assertEqual(ev.get_clock_value(cc).cycles, 177) | |
172 | ||
173 | def test_no_clock_value(self): | |
174 | tc = bt2.Trace() | |
175 | tc.add_stream_class(self._ec.stream_class) | |
176 | cc = bt2.ClockClass('hi') | |
177 | tc.add_clock_class(cc) | |
178 | ev = self._ec() | |
179 | self.assertIsNone(ev.get_clock_value(cc)) | |
180 | ||
181 | def test_no_packet(self): | |
182 | ev = self._ec() | |
183 | self.assertIsNone(ev.packet) | |
184 | ||
185 | def test_packet(self): | |
186 | tc = bt2.Trace() | |
187 | tc.packet_header_field_type = bt2.StructureFieldType() | |
188 | tc.packet_header_field_type.append_field('magic', bt2.IntegerFieldType(32)) | |
189 | tc.packet_header_field_type.append_field('stream_id', bt2.IntegerFieldType(16)) | |
190 | tc.add_stream_class(self._ec.stream_class) | |
191 | ev = self._ec() | |
192 | self._fill_ev(ev) | |
193 | stream = self._ec.stream_class() | |
194 | packet = stream.create_packet() | |
195 | packet.header_field['magic'] = 0xc1fc1fc1 | |
196 | packet.header_field['stream_id'] = 0 | |
197 | packet.context_field['something'] = 154 | |
198 | packet.context_field['something_else'] = 17.2 | |
199 | ev.packet = packet | |
200 | self.assertEqual(ev.packet.addr, packet.addr) | |
201 | ||
202 | def test_no_stream(self): | |
203 | ev = self._ec() | |
204 | self.assertIsNone(ev.stream) | |
205 | ||
206 | def test_stream(self): | |
207 | tc = bt2.Trace() | |
208 | tc.packet_header_field_type = bt2.StructureFieldType() | |
209 | tc.packet_header_field_type.append_field('magic', bt2.IntegerFieldType(32)) | |
210 | tc.packet_header_field_type.append_field('stream_id', bt2.IntegerFieldType(16)) | |
211 | tc.add_stream_class(self._ec.stream_class) | |
212 | ev = self._ec() | |
213 | self._fill_ev(ev) | |
214 | stream = self._ec.stream_class() | |
215 | packet = stream.create_packet() | |
216 | packet.header_field['magic'] = 0xc1fc1fc1 | |
217 | packet.header_field['stream_id'] = 0 | |
218 | packet.context_field['something'] = 154 | |
219 | packet.context_field['something_else'] = 17.2 | |
220 | ev.packet = packet | |
221 | self.assertEqual(ev.stream.addr, stream.addr) | |
222 | ||
223 | def _fill_ev(self, ev): | |
224 | ev.header_field['id'] = 23 | |
225 | ev.header_field['ts'] = 1234 | |
226 | ev.stream_event_context_field['cpu_id'] = 1 | |
227 | ev.stream_event_context_field['stuff'] = 13.194 | |
228 | ev.context_field['ant'] = -1 | |
229 | ev.context_field['msg'] = 'hellooo' | |
230 | ev.payload_field['giraffe'] = 1 | |
231 | ev.payload_field['gnu'] = 23 | |
232 | ev.payload_field['mosquito'] = 42 | |
233 | ||
234 | def _get_full_ev(self): | |
235 | tc = bt2.Trace() | |
236 | tc.add_stream_class(self._ec.stream_class) | |
237 | cc = bt2.ClockClass('hi') | |
238 | tc.add_clock_class(cc) | |
239 | ev = self._ec() | |
240 | self._fill_ev(ev) | |
241 | ev.clock_value(cc, 234) | |
242 | return ev | |
243 | ||
244 | def test_getitem(self): | |
245 | tc = bt2.Trace() | |
246 | tc.packet_header_field_type = bt2.StructureFieldType() | |
247 | tc.packet_header_field_type.append_field('magic', bt2.IntegerFieldType(32)) | |
248 | tc.packet_header_field_type.append_field('stream_id', bt2.IntegerFieldType(16)) | |
249 | tc.add_stream_class(self._ec.stream_class) | |
250 | ev = self._ec() | |
251 | self._fill_ev(ev) | |
252 | stream = self._ec.stream_class() | |
253 | packet = stream.create_packet() | |
254 | packet.header_field['magic'] = 0xc1fc1fc1 | |
255 | packet.header_field['stream_id'] = 0 | |
256 | packet.context_field['something'] = 154 | |
257 | packet.context_field['something_else'] = 17.2 | |
258 | ||
259 | with self.assertRaises(KeyError): | |
260 | ev['magic'] | |
261 | ||
262 | ev.packet = packet | |
263 | self.assertEqual(ev['mosquito'], 42) | |
264 | self.assertEqual(ev['gnu'], 23) | |
265 | self.assertEqual(ev['giraffe'], 1) | |
266 | self.assertEqual(ev['msg'], 'hellooo') | |
267 | self.assertEqual(ev['ant'], -1) | |
268 | self.assertEqual(ev['stuff'], 13.194) | |
269 | self.assertEqual(ev['cpu_id'], 1) | |
270 | self.assertEqual(ev['ts'], 1234) | |
271 | self.assertEqual(ev['id'], 23) | |
272 | self.assertEqual(ev['something_else'], 17.2) | |
273 | self.assertEqual(ev['something'], 154) | |
274 | self.assertEqual(ev['stream_id'], 0) | |
275 | self.assertEqual(ev['magic'], 0xc1fc1fc1) | |
276 | ||
277 | with self.assertRaises(KeyError): | |
278 | ev['yes'] | |
279 | ||
280 | def test_eq(self): | |
281 | tc = bt2.Trace() | |
282 | tc.add_stream_class(self._ec.stream_class) | |
283 | cc = bt2.ClockClass('hi') | |
284 | tc.add_clock_class(cc) | |
285 | ev1 = self._ec() | |
286 | self._fill_ev(ev1) | |
287 | ev1.set_clock_value(cc.create_clock_value(234)) | |
288 | ev2 = self._ec() | |
289 | self._fill_ev(ev2) | |
290 | ev2.set_clock_value(cc.create_clock_value(234)) | |
291 | self.assertEqual(ev1, ev2) | |
292 | ||
293 | def test_ne_header_field(self): | |
294 | tc = bt2.Trace() | |
295 | tc.add_stream_class(self._ec.stream_class) | |
296 | cc = bt2.ClockClass('hi') | |
297 | tc.add_clock_class(cc) | |
298 | ev1 = self._ec() | |
299 | self._fill_ev(ev1) | |
300 | ev1.header_field['id'] = 19 | |
301 | ev1.set_clock_value(cc.create_clock_value(234)) | |
302 | ev2 = self._ec() | |
303 | self._fill_ev(ev2) | |
304 | ev2.set_clock_value(cc.create_clock_value(234)) | |
305 | self.assertNotEqual(ev1, ev2) | |
306 | ||
307 | def test_ne_stream_event_context_field(self): | |
308 | tc = bt2.Trace() | |
309 | tc.add_stream_class(self._ec.stream_class) | |
310 | cc = bt2.ClockClass('hi') | |
311 | tc.add_clock_class(cc) | |
312 | ev1 = self._ec() | |
313 | self._fill_ev(ev1) | |
314 | ev1.stream_event_context_field['cpu_id'] = 3 | |
315 | ev1.set_clock_value(cc.create_clock_value(234)) | |
316 | ev2 = self._ec() | |
317 | self._fill_ev(ev2) | |
318 | ev2.set_clock_value(cc.create_clock_value(234)) | |
319 | self.assertNotEqual(ev1, ev2) | |
320 | ||
321 | def test_ne_context_field(self): | |
322 | tc = bt2.Trace() | |
323 | tc.add_stream_class(self._ec.stream_class) | |
324 | cc = bt2.ClockClass('hi') | |
325 | tc.add_clock_class(cc) | |
326 | ev1 = self._ec() | |
327 | self._fill_ev(ev1) | |
328 | ev1.context_field['ant'] = -3 | |
329 | ev1.set_clock_value(cc.create_clock_value(234)) | |
330 | ev2 = self._ec() | |
331 | self._fill_ev(ev2) | |
332 | ev2.set_clock_value(cc.create_clock_value(234)) | |
333 | self.assertNotEqual(ev1, ev2) | |
334 | ||
335 | def test_ne_payload_field(self): | |
336 | tc = bt2.Trace() | |
337 | tc.add_stream_class(self._ec.stream_class) | |
338 | cc = bt2.ClockClass('hi') | |
339 | tc.add_clock_class(cc) | |
340 | ev1 = self._ec() | |
341 | self._fill_ev(ev1) | |
342 | ev1.payload_field['mosquito'] = 98 | |
343 | ev1.set_clock_value(cc.create_clock_value(234)) | |
344 | ev2 = self._ec() | |
345 | self._fill_ev(ev2) | |
346 | ev2.set_clock_value(cc.create_clock_value(234)) | |
347 | self.assertNotEqual(ev1, ev2) | |
348 | ||
349 | def test_eq_invalid(self): | |
350 | ev = self._ec() | |
351 | self.assertFalse(ev == 23) | |
352 | ||
353 | def _test_copy(self, func): | |
354 | tc = bt2.Trace() | |
355 | tc.add_stream_class(self._ec.stream_class) | |
356 | cc = bt2.ClockClass('hi') | |
357 | tc.add_clock_class(cc) | |
358 | ev = self._ec() | |
359 | self._fill_ev(ev) | |
360 | ev.set_clock_value(cc.create_clock_value(234)) | |
361 | cpy = func(ev) | |
362 | self.assertIsNot(ev, cpy) | |
363 | self.assertNotEqual(ev.addr, cpy.addr) | |
364 | self.assertEqual(ev, cpy) | |
365 | ||
366 | def test_copy(self): | |
367 | self._test_copy(copy.copy) | |
368 | ||
369 | def test_deepcopy(self): | |
370 | self._test_copy(copy.deepcopy) |