bt2: Mass field_types -> field_class rename
[babeltrace.git] / tests / bindings / python / bt2 / test_trace.py
1 from bt2 import values
2 import unittest
3 import copy
4 import uuid
5 import bt2
6
7
8 @unittest.skip("this is broken")
9 class TraceTestCase(unittest.TestCase):
10 def setUp(self):
11 self._sc = self._create_stream_class('sc1', 3)
12 self._tc = bt2.Trace()
13
14 def tearDown(self):
15 del self._sc
16 del self._tc
17
18 def _create_stream_class(self, name, id):
19 ec1, ec2 = self._create_event_classes()
20 packet_context_fc = bt2.StructureFieldClass()
21 packet_context_fc.append_field('menu', bt2.FloatingPointNumberFieldClass())
22 packet_context_fc.append_field('sticker', bt2.StringFieldClass())
23 event_header_fc = bt2.StructureFieldClass()
24 event_header_fc.append_field('id', bt2.IntegerFieldClass(19))
25 event_context_fc = bt2.StructureFieldClass()
26 event_context_fc.append_field('msg', bt2.StringFieldClass())
27 return bt2.StreamClass(name=name, id=id,
28 packet_context_field_class=packet_context_fc,
29 event_header_field_class=event_header_fc,
30 event_context_field_class=event_context_fc,
31 event_classes=(ec1, ec2))
32
33 def _create_event_classes(self):
34 context_fc = bt2.StructureFieldClass()
35 context_fc.append_field('allo', bt2.StringFieldClass())
36 context_fc.append_field('zola', bt2.IntegerFieldClass(18))
37 payload_fc = bt2.StructureFieldClass()
38 payload_fc.append_field('zoom', bt2.StringFieldClass())
39 ec1 = bt2.EventClass('event23', id=23, context_field_class=context_fc,
40 payload_field_class=payload_fc)
41 ec2 = bt2.EventClass('event17', id=17, context_field_class=payload_fc,
42 payload_field_class=context_fc)
43 return ec1, ec2
44
45 def test_create_default(self):
46 self.assertEqual(len(self._tc), 0)
47
48 def _get_std_header(self):
49 header_fc = bt2.StructureFieldClass()
50 header_fc.append_field('magic', bt2.IntegerFieldClass(32))
51 header_fc.append_field('stream_id', bt2.IntegerFieldClass(32))
52 return header_fc
53
54 def test_create_full(self):
55 clock_classes = bt2.ClockClass('cc1', 1000), bt2.ClockClass('cc2', 30)
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},
60 packet_header_field_class=self._get_std_header(),
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)
67 self.assertEqual(tc.packet_header_field_class, self._get_std_header())
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
80 def test_assign_static(self):
81 self._tc.set_is_static()
82 self.assertTrue(self._tc.is_static)
83
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_class(self):
93 header_fc = bt2.StructureFieldClass()
94 header_fc.append_field('magic', bt2.IntegerFieldClass(32))
95 self._tc.packet_header_field_class = header_fc
96 self.assertEqual(self._tc.packet_header_field_class, header_fc)
97
98 def test_assign_no_packet_header_field_class(self):
99 self._tc.packet_header_field_class = None
100 self.assertIsNone(self._tc.packet_header_field_class)
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):
109 self._tc.packet_header_field_class = self._get_std_header()
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)
114 self._tc.add_clock_class(bt2.ClockClass('cc1', 1000))
115 self._tc.add_clock_class(bt2.ClockClass('cc2', 30))
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_class.addr, cpy.packet_header_field_class.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_class.addr, cpy.packet_header_field_class.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):
162 self._tc.packet_header_field_class = self._get_std_header()
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
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
211 def _test_eq_create_objects(self):
212 cc1_uuid = uuid.UUID('bc7f2f2d-2ee4-4e03-ab1f-2e0e1304e94f')
213 cc1 = bt2.ClockClass('cc1', 1000, uuid=cc1_uuid)
214 cc2_uuid = uuid.UUID('da7d6b6f-3108-4706-89bd-ab554732611b')
215 cc2 = bt2.ClockClass('cc2', 30, uuid=cc2_uuid)
216 sc1 = self._create_stream_class('sc1', 3)
217 sc2 = self._create_stream_class('sc2', 9)
218 return cc1, cc2, sc1, sc2, self._get_std_header()
219
220 def test_eq(self):
221 cc1, cc2, sc1, sc2, header_fc = 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_class=header_fc,
226 clock_classes=(cc1, cc2),
227 stream_classes=(sc1, sc2))
228 cc1, cc2, sc1, sc2, header_fc = 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_class=header_fc,
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_fc = 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_class=header_fc,
243 clock_classes=(cc1, cc2),
244 stream_classes=(sc1, sc2))
245 cc1, cc2, sc1, sc2, header_fc = 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_class=header_fc,
250 clock_classes=(cc1, cc2),
251 stream_classes=(sc1, sc2))
252 self.assertNotEqual(tc1, tc2)
253
254 def test_ne_packet_header_field_class(self):
255 cc1, cc2, sc1, sc2, header_fc = 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_class=header_fc,
260 clock_classes=(cc1, cc2),
261 stream_classes=(sc1, sc2))
262 cc1, cc2, sc1, sc2, header_fc = self._test_eq_create_objects()
263 header_fc.append_field('yes', bt2.StringFieldClass())
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_class=header_fc,
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_fc = 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_class=header_fc,
278 clock_classes=(cc1, cc2),
279 stream_classes=(sc1, sc2))
280 cc1, cc2, sc1, sc2, header_fc = 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_class=header_fc,
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_fc = 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_class=header_fc,
295 clock_classes=(cc1, cc2),
296 stream_classes=(sc1, sc2))
297 cc1, cc2, sc1, sc2, header_fc = 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_class=header_fc,
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_fc = 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_class=header_fc,
312 clock_classes=(cc1, cc2),
313 stream_classes=(sc1, sc2))
314 cc1, cc2, sc1, sc2, header_fc = 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_class=header_fc,
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_fc = 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_class=header_fc,
330 clock_classes=(cc1, cc2),
331 stream_classes=(sc1, sc2))
332 cc1, cc2, sc1, sc2, header_fc = 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_class=header_fc,
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)
This page took 0.0479619999999999 seconds and 5 git commands to generate.