Commit | Line | Data |
---|---|---|
9cf643d1 PP |
1 | from bt2 import values |
2 | import unittest | |
3 | import copy | |
4 | import uuid | |
5 | import bt2 | |
6 | ||
7 | ||
90cfc012 | 8 | @unittest.skip("this is broken") |
9cf643d1 PP |
9 | class TraceTestCase(unittest.TestCase): |
10 | def setUp(self): | |
11 | self._sc = self._create_stream_class('sc1', 3) | |
12 | self._tc = bt2.Trace() | |
13 | ||
f6a5e476 PP |
14 | def tearDown(self): |
15 | del self._sc | |
16 | del self._tc | |
17 | ||
9cf643d1 PP |
18 | def _create_stream_class(self, name, id): |
19 | ec1, ec2 = self._create_event_classes() | |
20 | packet_context_ft = bt2.StructureFieldType() | |
21 | packet_context_ft.append_field('menu', bt2.FloatingPointNumberFieldType()) | |
22 | packet_context_ft.append_field('sticker', bt2.StringFieldType()) | |
23 | event_header_ft = bt2.StructureFieldType() | |
24 | event_header_ft.append_field('id', bt2.IntegerFieldType(19)) | |
25 | event_context_ft = bt2.StructureFieldType() | |
26 | event_context_ft.append_field('msg', bt2.StringFieldType()) | |
27 | return bt2.StreamClass(name=name, id=id, | |
28 | packet_context_field_type=packet_context_ft, | |
29 | event_header_field_type=event_header_ft, | |
30 | event_context_field_type=event_context_ft, | |
31 | event_classes=(ec1, ec2)) | |
32 | ||
33 | def _create_event_classes(self): | |
34 | context_ft = bt2.StructureFieldType() | |
35 | context_ft.append_field('allo', bt2.StringFieldType()) | |
36 | context_ft.append_field('zola', bt2.IntegerFieldType(18)) | |
37 | payload_ft = bt2.StructureFieldType() | |
38 | payload_ft.append_field('zoom', bt2.StringFieldType()) | |
39 | ec1 = bt2.EventClass('event23', id=23, context_field_type=context_ft, | |
40 | payload_field_type=payload_ft) | |
41 | ec2 = bt2.EventClass('event17', id=17, context_field_type=payload_ft, | |
42 | payload_field_type=context_ft) | |
43 | return ec1, ec2 | |
44 | ||
45 | def test_create_default(self): | |
46 | self.assertEqual(len(self._tc), 0) | |
47 | ||
f6a5e476 | 48 | def _get_std_header(self): |
9cf643d1 PP |
49 | header_ft = bt2.StructureFieldType() |
50 | header_ft.append_field('magic', bt2.IntegerFieldType(32)) | |
f6a5e476 PP |
51 | header_ft.append_field('stream_id', bt2.IntegerFieldType(32)) |
52 | return header_ft | |
53 | ||
54 | def test_create_full(self): | |
55 | clock_classes = bt2.ClockClass('cc1', 1000), bt2.ClockClass('cc2', 30) | |
9cf643d1 PP |
56 | sc = self._create_stream_class('sc1', 3) |
57 | tc = bt2.Trace(name='my name', | |
58 | native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN, | |
59 | env={'the_string': 'value', 'the_int': 23}, | |
f6a5e476 | 60 | packet_header_field_type=self._get_std_header(), |
9cf643d1 PP |
61 | clock_classes=clock_classes, |
62 | stream_classes=(sc,)) | |
63 | self.assertEqual(tc.name, 'my name') | |
64 | self.assertEqual(tc.native_byte_order, bt2.ByteOrder.LITTLE_ENDIAN) | |
65 | self.assertEqual(tc.env['the_string'], 'value') | |
66 | self.assertEqual(tc.env['the_int'], 23) | |
f6a5e476 | 67 | self.assertEqual(tc.packet_header_field_type, self._get_std_header()) |
9cf643d1 PP |
68 | self.assertEqual(tc.clock_classes['cc1'], clock_classes[0]) |
69 | self.assertEqual(tc.clock_classes['cc2'], clock_classes[1]) | |
70 | self.assertEqual(tc[3], sc) | |
71 | ||
72 | def test_assign_name(self): | |
73 | self._tc.name = 'lel' | |
74 | self.assertEqual(self._tc.name, 'lel') | |
75 | ||
76 | def test_assign_invalid_name(self): | |
77 | with self.assertRaises(TypeError): | |
78 | self._tc.name = 17 | |
79 | ||
f6a5e476 PP |
80 | def test_assign_static(self): |
81 | self._tc.set_is_static() | |
82 | self.assertTrue(self._tc.is_static) | |
83 | ||
9cf643d1 PP |
84 | def test_assign_native_byte_order(self): |
85 | self._tc.native_byte_order = bt2.ByteOrder.BIG_ENDIAN | |
86 | self.assertEqual(self._tc.native_byte_order, bt2.ByteOrder.BIG_ENDIAN) | |
87 | ||
88 | def test_assign_invalid_native_byte_order(self): | |
89 | with self.assertRaises(TypeError): | |
90 | self._tc.native_byte_order = 'lel' | |
91 | ||
92 | def test_assign_packet_header_field_type(self): | |
93 | header_ft = bt2.StructureFieldType() | |
94 | header_ft.append_field('magic', bt2.IntegerFieldType(32)) | |
95 | self._tc.packet_header_field_type = header_ft | |
96 | self.assertEqual(self._tc.packet_header_field_type, header_ft) | |
97 | ||
98 | def test_assign_no_packet_header_field_type(self): | |
99 | self._tc.packet_header_field_type = None | |
100 | self.assertIsNone(self._tc.packet_header_field_type) | |
101 | ||
102 | def _test_copy(self, cpy): | |
103 | self.assertIsNot(cpy, self._tc) | |
104 | self.assertNotEqual(cpy.addr, self._tc.addr) | |
105 | self.assertEqual(cpy, self._tc) | |
106 | self.assertEqual(len(self._tc), len(cpy)) | |
107 | ||
108 | def _pre_copy(self): | |
f6a5e476 | 109 | self._tc.packet_header_field_type = self._get_std_header() |
9cf643d1 PP |
110 | self._tc.name = 'the trace class' |
111 | sc1 = self._create_stream_class('sc1', 3) | |
112 | sc2 = self._create_stream_class('sc2', 9) | |
113 | sc3 = self._create_stream_class('sc3', 17) | |
f6a5e476 PP |
114 | self._tc.add_clock_class(bt2.ClockClass('cc1', 1000)) |
115 | self._tc.add_clock_class(bt2.ClockClass('cc2', 30)) | |
9cf643d1 PP |
116 | self._tc.env['allo'] = 'bateau' |
117 | self._tc.env['bateau'] = 'cart' | |
118 | self._tc.add_stream_class(sc1) | |
119 | self._tc.add_stream_class(sc2) | |
120 | self._tc.add_stream_class(sc3) | |
121 | ||
122 | def test_copy(self): | |
123 | self._pre_copy() | |
124 | cpy = copy.copy(self._tc) | |
125 | self._test_copy(cpy) | |
126 | self.assertEqual(self._tc.packet_header_field_type.addr, cpy.packet_header_field_type.addr) | |
127 | self.assertEqual(self._tc.clock_classes['cc1'].addr, cpy.clock_classes['cc1'].addr) | |
128 | self.assertEqual(self._tc.clock_classes['cc2'].addr, cpy.clock_classes['cc2'].addr) | |
129 | self.assertEqual(self._tc.env['allo'].addr, cpy.env['allo'].addr) | |
130 | self.assertEqual(self._tc.env['bateau'].addr, cpy.env['bateau'].addr) | |
131 | ||
132 | def test_deepcopy(self): | |
133 | self._pre_copy() | |
134 | cpy = copy.deepcopy(self._tc) | |
135 | self._test_copy(cpy) | |
136 | self.assertNotEqual(self._tc.packet_header_field_type.addr, cpy.packet_header_field_type.addr) | |
137 | self.assertNotEqual(self._tc.clock_classes['cc1'].addr, cpy.clock_classes['cc1'].addr) | |
138 | self.assertNotEqual(self._tc.clock_classes['cc2'].addr, cpy.clock_classes['cc2'].addr) | |
139 | self.assertNotEqual(self._tc.env['allo'].addr, cpy.env['allo'].addr) | |
140 | self.assertNotEqual(self._tc.env['bateau'].addr, cpy.env['bateau'].addr) | |
141 | ||
142 | def test_getitem(self): | |
143 | self._tc.add_stream_class(self._sc) | |
144 | self.assertEqual(self._tc[3].addr, self._sc.addr) | |
145 | ||
146 | def test_getitem_wrong_key_type(self): | |
147 | self._tc.add_stream_class(self._sc) | |
148 | with self.assertRaises(TypeError): | |
149 | self._tc['hello'] | |
150 | ||
151 | def test_getitem_wrong_key(self): | |
152 | self._tc.add_stream_class(self._sc) | |
153 | with self.assertRaises(KeyError): | |
154 | self._tc[4] | |
155 | ||
156 | def test_len(self): | |
157 | self.assertEqual(len(self._tc), 0) | |
158 | self._tc.add_stream_class(self._sc) | |
159 | self.assertEqual(len(self._tc), 1) | |
160 | ||
161 | def test_iter(self): | |
f6a5e476 | 162 | self._tc.packet_header_field_type = self._get_std_header() |
9cf643d1 PP |
163 | sc1 = self._create_stream_class('sc1', 3) |
164 | sc2 = self._create_stream_class('sc2', 9) | |
165 | sc3 = self._create_stream_class('sc3', 17) | |
166 | self._tc.add_stream_class(sc1) | |
167 | self._tc.add_stream_class(sc2) | |
168 | self._tc.add_stream_class(sc3) | |
169 | ||
170 | for sid, stream_class in self._tc.items(): | |
171 | self.assertIsInstance(stream_class, bt2.StreamClass) | |
172 | ||
173 | if sid == 3: | |
174 | self.assertEqual(stream_class.addr, sc1.addr) | |
175 | elif sid == 9: | |
176 | self.assertEqual(stream_class.addr, sc2.addr) | |
177 | elif sid == 17: | |
178 | self.assertEqual(stream_class.addr, sc3.addr) | |
179 | ||
180 | def test_env_getitem_wrong_key(self): | |
181 | with self.assertRaises(KeyError): | |
182 | self._tc.env['lel'] | |
183 | ||
184 | def test_clock_classes_getitem_wrong_key(self): | |
185 | with self.assertRaises(KeyError): | |
186 | self._tc.clock_classes['lel'] | |
187 | ||
f6a5e476 PP |
188 | def test_streams_none(self): |
189 | self.assertEqual(len(self._tc.streams), 0) | |
190 | ||
191 | def test_streams_len(self): | |
192 | self._tc.add_stream_class(self._create_stream_class('sc1', 3)) | |
193 | stream0 = self._tc[3]() | |
194 | stream1 = self._tc[3]() | |
195 | stream2 = self._tc[3]() | |
196 | self.assertEqual(len(self._tc.streams), 3) | |
197 | ||
198 | def test_streams_iter(self): | |
199 | self._tc.add_stream_class(self._create_stream_class('sc1', 3)) | |
200 | stream0 = self._tc[3](id=12) | |
201 | stream1 = self._tc[3](id=15) | |
202 | stream2 = self._tc[3](id=17) | |
203 | sids = set() | |
204 | ||
205 | for stream in self._tc.streams: | |
206 | sids.add(stream.id) | |
207 | ||
208 | self.assertEqual(len(sids), 3) | |
209 | self.assertTrue(12 in sids and 15 in sids and 17 in sids) | |
210 | ||
9cf643d1 PP |
211 | def _test_eq_create_objects(self): |
212 | cc1_uuid = uuid.UUID('bc7f2f2d-2ee4-4e03-ab1f-2e0e1304e94f') | |
f6a5e476 | 213 | cc1 = bt2.ClockClass('cc1', 1000, uuid=cc1_uuid) |
9cf643d1 | 214 | cc2_uuid = uuid.UUID('da7d6b6f-3108-4706-89bd-ab554732611b') |
f6a5e476 | 215 | cc2 = bt2.ClockClass('cc2', 30, uuid=cc2_uuid) |
9cf643d1 PP |
216 | sc1 = self._create_stream_class('sc1', 3) |
217 | sc2 = self._create_stream_class('sc2', 9) | |
f6a5e476 | 218 | return cc1, cc2, sc1, sc2, self._get_std_header() |
9cf643d1 PP |
219 | |
220 | def test_eq(self): | |
221 | cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects() | |
222 | tc1 = bt2.Trace(name='my name', | |
223 | native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN, | |
224 | env={'the_string': 'value', 'the_int': 23}, | |
225 | packet_header_field_type=header_ft, | |
226 | clock_classes=(cc1, cc2), | |
227 | stream_classes=(sc1, sc2)) | |
228 | cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects() | |
229 | tc2 = bt2.Trace(name='my name', | |
230 | native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN, | |
231 | env={'the_string': 'value', 'the_int': 23}, | |
232 | packet_header_field_type=header_ft, | |
233 | clock_classes=(cc1, cc2), | |
234 | stream_classes=(sc1, sc2)) | |
235 | self.assertEqual(tc1, tc2) | |
236 | ||
237 | def test_ne_name(self): | |
238 | cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects() | |
239 | tc1 = bt2.Trace(name='my name2', | |
240 | native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN, | |
241 | env={'the_string': 'value', 'the_int': 23}, | |
242 | packet_header_field_type=header_ft, | |
243 | clock_classes=(cc1, cc2), | |
244 | stream_classes=(sc1, sc2)) | |
245 | cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects() | |
246 | tc2 = bt2.Trace(name='my name', | |
247 | native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN, | |
248 | env={'the_string': 'value', 'the_int': 23}, | |
249 | packet_header_field_type=header_ft, | |
250 | clock_classes=(cc1, cc2), | |
251 | stream_classes=(sc1, sc2)) | |
252 | self.assertNotEqual(tc1, tc2) | |
253 | ||
254 | def test_ne_packet_header_field_type(self): | |
255 | cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects() | |
256 | tc1 = bt2.Trace(name='my name', | |
257 | native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN, | |
258 | env={'the_string': 'value', 'the_int': 23}, | |
259 | packet_header_field_type=header_ft, | |
260 | clock_classes=(cc1, cc2), | |
261 | stream_classes=(sc1, sc2)) | |
262 | cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects() | |
263 | header_ft.append_field('yes', bt2.StringFieldType()) | |
264 | tc2 = bt2.Trace(name='my name', | |
265 | native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN, | |
266 | env={'the_string': 'value', 'the_int': 23}, | |
267 | packet_header_field_type=header_ft, | |
268 | clock_classes=(cc1, cc2), | |
269 | stream_classes=(sc1, sc2)) | |
270 | self.assertNotEqual(tc1, tc2) | |
271 | ||
272 | def test_ne_native_byte_order(self): | |
273 | cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects() | |
274 | tc1 = bt2.Trace(name='my name', | |
275 | native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN, | |
276 | env={'the_string': 'value', 'the_int': 23}, | |
277 | packet_header_field_type=header_ft, | |
278 | clock_classes=(cc1, cc2), | |
279 | stream_classes=(sc1, sc2)) | |
280 | cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects() | |
281 | tc2 = bt2.Trace(name='my name', | |
282 | native_byte_order=bt2.ByteOrder.BIG_ENDIAN, | |
283 | env={'the_string': 'value', 'the_int': 23}, | |
284 | packet_header_field_type=header_ft, | |
285 | clock_classes=(cc1, cc2), | |
286 | stream_classes=(sc1, sc2)) | |
287 | self.assertNotEqual(tc1, tc2) | |
288 | ||
289 | def test_ne_env(self): | |
290 | cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects() | |
291 | tc1 = bt2.Trace(name='my name', | |
292 | native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN, | |
293 | env={'the_string': 'value', 'the_int2': 23}, | |
294 | packet_header_field_type=header_ft, | |
295 | clock_classes=(cc1, cc2), | |
296 | stream_classes=(sc1, sc2)) | |
297 | cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects() | |
298 | tc2 = bt2.Trace(name='my name', | |
299 | native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN, | |
300 | env={'the_string': 'value', 'the_int': 23}, | |
301 | packet_header_field_type=header_ft, | |
302 | clock_classes=(cc1, cc2), | |
303 | stream_classes=(sc1, sc2)) | |
304 | self.assertNotEqual(tc1, tc2) | |
305 | ||
306 | def test_ne_clock_classes(self): | |
307 | cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects() | |
308 | tc1 = bt2.Trace(name='my name', | |
309 | native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN, | |
310 | env={'the_string': 'value', 'the_int': 23}, | |
311 | packet_header_field_type=header_ft, | |
312 | clock_classes=(cc1, cc2), | |
313 | stream_classes=(sc1, sc2)) | |
314 | cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects() | |
315 | cc2.frequency = 1234 | |
316 | tc2 = bt2.Trace(name='my name', | |
317 | native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN, | |
318 | env={'the_string': 'value', 'the_int': 23}, | |
319 | packet_header_field_type=header_ft, | |
320 | clock_classes=(cc1, cc2), | |
321 | stream_classes=(sc1, sc2)) | |
322 | self.assertNotEqual(tc1, tc2) | |
323 | ||
324 | def test_ne_stream_classes(self): | |
325 | cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects() | |
326 | tc1 = bt2.Trace(name='my name', | |
327 | native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN, | |
328 | env={'the_string': 'value', 'the_int': 23}, | |
329 | packet_header_field_type=header_ft, | |
330 | clock_classes=(cc1, cc2), | |
331 | stream_classes=(sc1, sc2)) | |
332 | cc1, cc2, sc1, sc2, header_ft = self._test_eq_create_objects() | |
333 | sc2.id = 72632 | |
334 | tc2 = bt2.Trace(name='my name', | |
335 | native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN, | |
336 | env={'the_string': 'value', 'the_int': 23}, | |
337 | packet_header_field_type=header_ft, | |
338 | clock_classes=(cc1, cc2), | |
339 | stream_classes=(sc1, sc2)) | |
340 | self.assertNotEqual(tc1, tc2) | |
341 | ||
342 | def test_eq_invalid(self): | |
343 | self.assertFalse(self._tc == 23) |