1 from collections
import OrderedDict
8 class EventTestCase(unittest
.TestCase
):
10 self
._ec
= self
._create
_ec
()
12 def _create_ec(self
, with_eh
=True, with_sec
=True, with_ec
=True, with_ep
=True):
15 eh
= bt2
.StructureFieldType()
17 ('id', bt2
.IntegerFieldType(8)),
18 ('ts', bt2
.IntegerFieldType(32)),
23 # stream event context
25 sec
= bt2
.StructureFieldType()
27 ('cpu_id', bt2
.IntegerFieldType(8)),
28 ('stuff', bt2
.FloatingPointNumberFieldType()),
34 pc
= bt2
.StructureFieldType()
36 ('something', bt2
.IntegerFieldType(8)),
37 ('something_else', bt2
.FloatingPointNumberFieldType()),
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
48 ec
= bt2
.StructureFieldType()
50 ('ant', bt2
.IntegerFieldType(16, is_signed
=True)),
51 ('msg', bt2
.StringFieldType()),
58 ep
= bt2
.StructureFieldType()
60 ('giraffe', bt2
.IntegerFieldType(32)),
61 ('gnu', bt2
.IntegerFieldType(8)),
62 ('mosquito', bt2
.IntegerFieldType(8)),
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
)
74 def test_attr_event_class(self
):
76 self
.assertEqual(ev
.event_class
.addr
, self
._ec
.addr
)
78 def test_attr_name(self
):
80 self
.assertEqual(ev
.name
, self
._ec
.name
)
82 def test_attr_id(self
):
84 self
.assertEqual(ev
.id, self
._ec
.id)
86 def test_get_event_header_field(self
):
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)
93 def test_set_event_header_field(self
):
94 eh
= self
._ec
.stream_class
.event_header_field_type()
99 self
.assertEqual(ev
.header_field
['id'], 17)
100 self
.assertEqual(ev
.header_field
['ts'], 188)
102 def test_get_stream_event_context_field(self
):
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)
109 def test_set_stream_event_context_field(self
):
110 sec
= self
._ec
.stream_class
.event_context_field_type()
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)
118 def test_no_stream_event_context(self
):
119 ec
= self
._create
_ec
(with_sec
=False)
121 self
.assertIsNone(ev
.stream_event_context_field
)
123 def test_get_event_context_field(self
):
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')
130 def test_set_event_context_field(self
):
131 ec
= self
._ec
.context_field_type()
133 ec
['msg'] = 'hi there'
135 ev
.context_field
= ec
136 self
.assertEqual(ev
.context_field
['ant'], 2)
137 self
.assertEqual(ev
.context_field
['msg'], 'hi there')
139 def test_no_event_context(self
):
140 ec
= self
._create
_ec
(with_ec
=False)
142 self
.assertIsNone(ev
.context_field
)
144 def test_get_event_payload_field(self
):
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)
153 def test_set_event_payload_field(self
):
154 ep
= self
._ec
.payload_field_type()
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)
164 def test_clock_value(self
):
166 tc
.add_stream_class(self
._ec
.stream_class
)
167 cc
= bt2
.ClockClass('hi')
168 tc
.add_clock_class(cc
)
170 ev
.set_clock_value(cc
.create_clock_value(177))
171 self
.assertEqual(ev
.get_clock_value(cc
).cycles
, 177)
173 def test_no_clock_value(self
):
175 tc
.add_stream_class(self
._ec
.stream_class
)
176 cc
= bt2
.ClockClass('hi')
177 tc
.add_clock_class(cc
)
179 self
.assertIsNone(ev
.get_clock_value(cc
))
181 def test_no_packet(self
):
183 self
.assertIsNone(ev
.packet
)
185 def test_packet(self
):
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
)
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
200 self
.assertEqual(ev
.packet
.addr
, packet
.addr
)
202 def test_no_stream(self
):
204 self
.assertIsNone(ev
.stream
)
206 def test_stream(self
):
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
)
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
221 self
.assertEqual(ev
.stream
.addr
, stream
.addr
)
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
234 def _get_full_ev(self
):
236 tc
.add_stream_class(self
._ec
.stream_class
)
237 cc
= bt2
.ClockClass('hi')
238 tc
.add_clock_class(cc
)
241 ev
.clock_value(cc
, 234)
244 def test_getitem(self
):
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
)
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
259 with self
.assertRaises(KeyError):
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)
277 with self
.assertRaises(KeyError):
282 tc
.add_stream_class(self
._ec
.stream_class
)
283 cc
= bt2
.ClockClass('hi')
284 tc
.add_clock_class(cc
)
287 ev1
.set_clock_value(cc
.create_clock_value(234))
290 ev2
.set_clock_value(cc
.create_clock_value(234))
291 self
.assertEqual(ev1
, ev2
)
293 def test_ne_header_field(self
):
295 tc
.add_stream_class(self
._ec
.stream_class
)
296 cc
= bt2
.ClockClass('hi')
297 tc
.add_clock_class(cc
)
300 ev1
.header_field
['id'] = 19
301 ev1
.set_clock_value(cc
.create_clock_value(234))
304 ev2
.set_clock_value(cc
.create_clock_value(234))
305 self
.assertNotEqual(ev1
, ev2
)
307 def test_ne_stream_event_context_field(self
):
309 tc
.add_stream_class(self
._ec
.stream_class
)
310 cc
= bt2
.ClockClass('hi')
311 tc
.add_clock_class(cc
)
314 ev1
.stream_event_context_field
['cpu_id'] = 3
315 ev1
.set_clock_value(cc
.create_clock_value(234))
318 ev2
.set_clock_value(cc
.create_clock_value(234))
319 self
.assertNotEqual(ev1
, ev2
)
321 def test_ne_context_field(self
):
323 tc
.add_stream_class(self
._ec
.stream_class
)
324 cc
= bt2
.ClockClass('hi')
325 tc
.add_clock_class(cc
)
328 ev1
.context_field
['ant'] = -3
329 ev1
.set_clock_value(cc
.create_clock_value(234))
332 ev2
.set_clock_value(cc
.create_clock_value(234))
333 self
.assertNotEqual(ev1
, ev2
)
335 def test_ne_payload_field(self
):
337 tc
.add_stream_class(self
._ec
.stream_class
)
338 cc
= bt2
.ClockClass('hi')
339 tc
.add_clock_class(cc
)
342 ev1
.payload_field
['mosquito'] = 98
343 ev1
.set_clock_value(cc
.create_clock_value(234))
346 ev2
.set_clock_value(cc
.create_clock_value(234))
347 self
.assertNotEqual(ev1
, ev2
)
349 def test_eq_invalid(self
):
351 self
.assertFalse(ev
== 23)
353 def _test_copy(self
, func
):
355 tc
.add_stream_class(self
._ec
.stream_class
)
356 cc
= bt2
.ClockClass('hi')
357 tc
.add_clock_class(cc
)
360 ev
.set_clock_value(cc
.create_clock_value(234))
362 self
.assertIsNot(ev
, cpy
)
363 self
.assertNotEqual(ev
.addr
, cpy
.addr
)
364 self
.assertEqual(ev
, cpy
)
367 self
._test
_copy
(copy
.copy
)
369 def test_deepcopy(self
):
370 self
._test
_copy
(copy
.deepcopy
)