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