1 from collections
import OrderedDict
8 class EventTestCase(unittest
.TestCase
):
10 self
._ec
= self
._create
_ec
()
15 def _create_ec(self
, with_eh
=True, with_sec
=True, with_ec
=True, with_ep
=True):
18 eh
= bt2
.StructureFieldType()
20 ('id', bt2
.IntegerFieldType(8)),
21 ('ts', bt2
.IntegerFieldType(32)),
26 # stream event context
28 sec
= bt2
.StructureFieldType()
30 ('cpu_id', bt2
.IntegerFieldType(8)),
31 ('stuff', bt2
.FloatingPointNumberFieldType()),
37 pc
= bt2
.StructureFieldType()
39 ('something', bt2
.IntegerFieldType(8)),
40 ('something_else', bt2
.FloatingPointNumberFieldType()),
44 sc
= bt2
.StreamClass()
45 sc
.packet_context_field_type
= pc
46 sc
.event_header_field_type
= eh
47 sc
.event_context_field_type
= sec
51 ec
= bt2
.StructureFieldType()
53 ('ant', bt2
.IntegerFieldType(16, is_signed
=True)),
54 ('msg', bt2
.StringFieldType()),
61 ep
= bt2
.StructureFieldType()
63 ('giraffe', bt2
.IntegerFieldType(32)),
64 ('gnu', bt2
.IntegerFieldType(8)),
65 ('mosquito', bt2
.IntegerFieldType(8)),
71 event_class
= bt2
.EventClass('ec')
72 event_class
.context_field_type
= ec
73 event_class
.payload_field_type
= ep
74 sc
.add_event_class(event_class
)
77 def test_attr_event_class(self
):
79 self
.assertEqual(ev
.event_class
.addr
, self
._ec
.addr
)
81 def test_attr_name(self
):
83 self
.assertEqual(ev
.name
, self
._ec
.name
)
85 def test_attr_id(self
):
87 self
.assertEqual(ev
.id, self
._ec
.id)
89 def test_get_event_header_field(self
):
91 ev
.header_field
['id'] = 23
92 ev
.header_field
['ts'] = 1234
93 self
.assertEqual(ev
.header_field
['id'], 23)
94 self
.assertEqual(ev
.header_field
['ts'], 1234)
96 def test_set_event_header_field(self
):
97 eh
= self
._ec
.stream_class
.event_header_field_type()
102 self
.assertEqual(ev
.header_field
['id'], 17)
103 self
.assertEqual(ev
.header_field
['ts'], 188)
105 def test_get_stream_event_context_field(self
):
107 ev
.stream_event_context_field
['cpu_id'] = 1
108 ev
.stream_event_context_field
['stuff'] = 13.194
109 self
.assertEqual(ev
.stream_event_context_field
['cpu_id'], 1)
110 self
.assertEqual(ev
.stream_event_context_field
['stuff'], 13.194)
112 def test_set_stream_event_context_field(self
):
113 sec
= self
._ec
.stream_class
.event_context_field_type()
117 ev
.stream_event_context_field
= sec
118 self
.assertEqual(ev
.stream_event_context_field
['cpu_id'], 2)
119 self
.assertEqual(ev
.stream_event_context_field
['stuff'], 19.19)
121 def test_no_stream_event_context(self
):
122 ec
= self
._create
_ec
(with_sec
=False)
124 self
.assertIsNone(ev
.stream_event_context_field
)
126 def test_get_event_context_field(self
):
128 ev
.context_field
['ant'] = -1
129 ev
.context_field
['msg'] = 'hellooo'
130 self
.assertEqual(ev
.context_field
['ant'], -1)
131 self
.assertEqual(ev
.context_field
['msg'], 'hellooo')
133 def test_set_event_context_field(self
):
134 ec
= self
._ec
.context_field_type()
136 ec
['msg'] = 'hi there'
138 ev
.context_field
= ec
139 self
.assertEqual(ev
.context_field
['ant'], 2)
140 self
.assertEqual(ev
.context_field
['msg'], 'hi there')
142 def test_no_event_context(self
):
143 ec
= self
._create
_ec
(with_ec
=False)
145 self
.assertIsNone(ev
.context_field
)
147 def test_get_event_payload_field(self
):
149 ev
.payload_field
['giraffe'] = 1
150 ev
.payload_field
['gnu'] = 23
151 ev
.payload_field
['mosquito'] = 42
152 self
.assertEqual(ev
.payload_field
['giraffe'], 1)
153 self
.assertEqual(ev
.payload_field
['gnu'], 23)
154 self
.assertEqual(ev
.payload_field
['mosquito'], 42)
156 def test_set_event_payload_field(self
):
157 ep
= self
._ec
.payload_field_type()
162 ev
.payload_field
= ep
163 self
.assertEqual(ev
.payload_field
['giraffe'], 2)
164 self
.assertEqual(ev
.payload_field
['gnu'], 124)
165 self
.assertEqual(ev
.payload_field
['mosquito'], 17)
167 def test_clock_value(self
):
169 tc
.add_stream_class(self
._ec
.stream_class
)
170 cc
= bt2
.ClockClass('hi', 1000)
171 tc
.add_clock_class(cc
)
173 ev
.add_clock_value(cc(177))
174 self
.assertEqual(ev
.clock_value(cc
).cycles
, 177)
176 def test_no_clock_value(self
):
178 tc
.add_stream_class(self
._ec
.stream_class
)
179 cc
= bt2
.ClockClass('hi', 1000)
180 tc
.add_clock_class(cc
)
182 self
.assertIsNone(ev
.clock_value(cc
))
184 def test_no_packet(self
):
186 self
.assertIsNone(ev
.packet
)
188 def test_packet(self
):
190 tc
.packet_header_field_type
= bt2
.StructureFieldType()
191 tc
.packet_header_field_type
.append_field('magic', bt2
.IntegerFieldType(32))
192 tc
.packet_header_field_type
.append_field('stream_id', bt2
.IntegerFieldType(16))
193 tc
.add_stream_class(self
._ec
.stream_class
)
196 stream
= self
._ec
.stream_class()
197 packet
= stream
.create_packet()
198 packet
.header_field
['magic'] = 0xc1fc1fc1
199 packet
.header_field
['stream_id'] = 0
200 packet
.context_field
['something'] = 154
201 packet
.context_field
['something_else'] = 17.2
203 self
.assertEqual(ev
.packet
.addr
, packet
.addr
)
205 def test_no_stream(self
):
207 self
.assertIsNone(ev
.stream
)
209 def test_stream(self
):
211 tc
.packet_header_field_type
= bt2
.StructureFieldType()
212 tc
.packet_header_field_type
.append_field('magic', bt2
.IntegerFieldType(32))
213 tc
.packet_header_field_type
.append_field('stream_id', bt2
.IntegerFieldType(16))
214 tc
.add_stream_class(self
._ec
.stream_class
)
217 stream
= self
._ec
.stream_class()
218 packet
= stream
.create_packet()
219 packet
.header_field
['magic'] = 0xc1fc1fc1
220 packet
.header_field
['stream_id'] = 0
221 packet
.context_field
['something'] = 154
222 packet
.context_field
['something_else'] = 17.2
224 self
.assertEqual(ev
.stream
.addr
, stream
.addr
)
226 def _fill_ev(self
, ev
):
227 ev
.header_field
['id'] = 23
228 ev
.header_field
['ts'] = 1234
229 ev
.stream_event_context_field
['cpu_id'] = 1
230 ev
.stream_event_context_field
['stuff'] = 13.194
231 ev
.context_field
['ant'] = -1
232 ev
.context_field
['msg'] = 'hellooo'
233 ev
.payload_field
['giraffe'] = 1
234 ev
.payload_field
['gnu'] = 23
235 ev
.payload_field
['mosquito'] = 42
237 def _get_full_ev(self
):
239 tc
.add_stream_class(self
._ec
.stream_class
)
240 cc
= bt2
.ClockClass('hi', 1000)
241 tc
.add_clock_class(cc
)
244 ev
.add_clock_value(cc(234))
247 def test_getitem(self
):
249 tc
.packet_header_field_type
= bt2
.StructureFieldType()
250 tc
.packet_header_field_type
.append_field('magic', bt2
.IntegerFieldType(32))
251 tc
.packet_header_field_type
.append_field('stream_id', bt2
.IntegerFieldType(16))
252 tc
.add_stream_class(self
._ec
.stream_class
)
255 stream
= self
._ec
.stream_class()
256 packet
= stream
.create_packet()
257 packet
.header_field
['magic'] = 0xc1fc1fc1
258 packet
.header_field
['stream_id'] = 0
259 packet
.context_field
['something'] = 154
260 packet
.context_field
['something_else'] = 17.2
262 with self
.assertRaises(KeyError):
266 self
.assertEqual(ev
['mosquito'], 42)
267 self
.assertEqual(ev
['gnu'], 23)
268 self
.assertEqual(ev
['giraffe'], 1)
269 self
.assertEqual(ev
['msg'], 'hellooo')
270 self
.assertEqual(ev
['ant'], -1)
271 self
.assertEqual(ev
['stuff'], 13.194)
272 self
.assertEqual(ev
['cpu_id'], 1)
273 self
.assertEqual(ev
['ts'], 1234)
274 self
.assertEqual(ev
['id'], 23)
275 self
.assertEqual(ev
['something_else'], 17.2)
276 self
.assertEqual(ev
['something'], 154)
277 self
.assertEqual(ev
['stream_id'], 0)
278 self
.assertEqual(ev
['magic'], 0xc1fc1fc1)
280 with self
.assertRaises(KeyError):
285 tc
.add_stream_class(self
._ec
.stream_class
)
286 cc
= bt2
.ClockClass('hi', 1000)
287 tc
.add_clock_class(cc
)
290 ev1
.add_clock_value(cc(234))
293 ev2
.add_clock_value(cc(234))
294 self
.assertEqual(ev1
, ev2
)
296 def test_ne_header_field(self
):
298 tc
.add_stream_class(self
._ec
.stream_class
)
299 cc
= bt2
.ClockClass('hi', 1000)
300 tc
.add_clock_class(cc
)
303 ev1
.header_field
['id'] = 19
304 ev1
.add_clock_value(cc(234))
307 ev2
.add_clock_value(cc(234))
308 self
.assertNotEqual(ev1
, ev2
)
310 def test_ne_stream_event_context_field(self
):
312 tc
.add_stream_class(self
._ec
.stream_class
)
313 cc
= bt2
.ClockClass('hi', 1000)
314 tc
.add_clock_class(cc
)
317 ev1
.stream_event_context_field
['cpu_id'] = 3
318 ev1
.add_clock_value(cc(234))
321 ev2
.add_clock_value(cc(234))
322 self
.assertNotEqual(ev1
, ev2
)
324 def test_ne_context_field(self
):
326 tc
.add_stream_class(self
._ec
.stream_class
)
327 cc
= bt2
.ClockClass('hi', 1000)
328 tc
.add_clock_class(cc
)
331 ev1
.context_field
['ant'] = -3
332 ev1
.add_clock_value(cc(234))
335 ev2
.add_clock_value(cc(234))
336 self
.assertNotEqual(ev1
, ev2
)
338 def test_ne_payload_field(self
):
340 tc
.add_stream_class(self
._ec
.stream_class
)
341 cc
= bt2
.ClockClass('hi', 1000)
342 tc
.add_clock_class(cc
)
345 ev1
.payload_field
['mosquito'] = 98
346 ev1
.add_clock_value(cc(234))
349 ev2
.add_clock_value(cc(234))
350 self
.assertNotEqual(ev1
, ev2
)
352 def test_eq_invalid(self
):
354 self
.assertFalse(ev
== 23)
356 def _test_copy(self
, func
):
358 tc
.add_stream_class(self
._ec
.stream_class
)
359 cc
= bt2
.ClockClass('hi', 1000)
360 tc
.add_clock_class(cc
)
363 ev
.add_clock_value(cc(234))
365 self
.assertIsNot(ev
, cpy
)
366 self
.assertNotEqual(ev
.addr
, cpy
.addr
)
367 self
.assertEqual(ev
, cpy
)
370 self
._test
_copy
(copy
.copy
)
372 def test_deepcopy(self
):
373 self
._test
_copy
(copy
.deepcopy
)