fix: test_message_iterator.py hangs on Python 3.12
[babeltrace.git] / tests / bindings / python / bt2 / test_stream_class.py
1 #
2 # Copyright (C) 2019 EfficiOS Inc.
3 #
4 # This program is free software; you can redistribute it and/or
5 # modify it under the terms of the GNU General Public License
6 # as published by the Free Software Foundation; only version 2
7 # of the License.
8 #
9 # This program is distributed in the hope that it will be useful,
10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 # GNU General Public License for more details.
13 #
14 # You should have received a copy of the GNU General Public License
15 # along with this program; if not, write to the Free Software
16 # Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
17 #
18
19 import unittest
20 from utils import run_in_component_init
21 from bt2 import stream_class as bt2_stream_class
22 from bt2 import trace_class as bt2_trace_class
23 from bt2 import clock_class as bt2_clock_class
24 from bt2 import event_class as bt2_event_class
25 from bt2 import field_class as bt2_field_class
26
27
28 class StreamClassTestCase(unittest.TestCase):
29 def setUp(self):
30 def f(comp_self):
31 tc = comp_self._create_trace_class(assigns_automatic_stream_class_id=True)
32 cc = comp_self._create_clock_class()
33 return tc, cc
34
35 self._tc, self._cc = run_in_component_init(f)
36 self._trace = self._tc()
37
38 def test_create_default(self):
39 sc = self._tc.create_stream_class()
40
41 self.assertIs(type(sc), bt2_stream_class._StreamClass)
42 self.assertIsNone(sc.name)
43 self.assertIsNone(sc.packet_context_field_class)
44 self.assertIsNone(sc.event_common_context_field_class)
45 self.assertIsNone(sc.default_clock_class)
46 self.assertTrue(sc.assigns_automatic_event_class_id)
47 self.assertTrue(sc.assigns_automatic_stream_id)
48 self.assertFalse(sc.supports_packets)
49 self.assertFalse(sc.packets_have_beginning_default_clock_snapshot)
50 self.assertFalse(sc.packets_have_end_default_clock_snapshot)
51 self.assertFalse(sc.supports_discarded_events)
52 self.assertFalse(sc.discarded_events_have_default_clock_snapshots)
53 self.assertFalse(sc.supports_discarded_packets)
54 self.assertFalse(sc.discarded_packets_have_default_clock_snapshots)
55 self.assertEqual(len(sc.user_attributes), 0)
56
57 def test_create_name(self):
58 sc = self._tc.create_stream_class(name='bozo')
59 self.assertEqual(sc.name, 'bozo')
60
61 def test_create_invalid_name(self):
62 with self.assertRaisesRegex(TypeError, "'int' is not a 'str' object"):
63 self._tc.create_stream_class(name=17)
64
65 self.assertEqual(len(self._tc), 0)
66
67 def test_create_packet_context_field_class(self):
68 fc = self._tc.create_structure_field_class()
69 sc = self._tc.create_stream_class(
70 packet_context_field_class=fc, supports_packets=True
71 )
72 self.assertEqual(sc.packet_context_field_class, fc)
73 self.assertIs(
74 type(sc.packet_context_field_class), bt2_field_class._StructureFieldClass
75 )
76
77 def test_create_invalid_packet_context_field_class(self):
78 with self.assertRaisesRegex(
79 TypeError,
80 "'int' is not a '<class 'bt2.field_class._StructureFieldClass'>' object",
81 ):
82 self._tc.create_stream_class(
83 packet_context_field_class=22, supports_packets=True
84 )
85
86 self.assertEqual(len(self._tc), 0)
87
88 def test_create_invalid_packet_context_field_class_no_packets(self):
89 fc = self._tc.create_structure_field_class()
90
91 with self.assertRaisesRegex(
92 ValueError,
93 "cannot have a packet context field class without supporting packets",
94 ):
95 self._tc.create_stream_class(packet_context_field_class=fc)
96
97 self.assertEqual(len(self._tc), 0)
98
99 def test_create_event_common_context_field_class(self):
100 fc = self._tc.create_structure_field_class()
101 sc = self._tc.create_stream_class(event_common_context_field_class=fc)
102 self.assertEqual(sc.event_common_context_field_class, fc)
103 self.assertIs(
104 type(sc.event_common_context_field_class),
105 bt2_field_class._StructureFieldClass,
106 )
107
108 def test_create_invalid_event_common_context_field_class(self):
109 with self.assertRaisesRegex(
110 TypeError,
111 "'int' is not a '<class 'bt2.field_class._StructureFieldClass'>' object",
112 ):
113 self._tc.create_stream_class(event_common_context_field_class=22)
114
115 self.assertEqual(len(self._tc), 0)
116
117 def test_create_default_clock_class(self):
118 sc = self._tc.create_stream_class(default_clock_class=self._cc)
119 self.assertEqual(sc.default_clock_class.addr, self._cc.addr)
120 self.assertIs(type(sc.default_clock_class), bt2_clock_class._ClockClass)
121
122 def test_create_invalid_default_clock_class(self):
123 with self.assertRaisesRegex(
124 TypeError, "'int' is not a '<class 'bt2.clock_class._ClockClass'>' object"
125 ):
126 self._tc.create_stream_class(default_clock_class=12)
127
128 self.assertEqual(len(self._tc), 0)
129
130 def test_create_user_attributes(self):
131 sc = self._tc.create_stream_class(user_attributes={'salut': 23})
132 self.assertEqual(sc.user_attributes, {'salut': 23})
133
134 def test_create_invalid_user_attributes(self):
135 with self.assertRaisesRegex(
136 TypeError, "cannot create value object from 'object' object"
137 ):
138 self._tc.create_stream_class(user_attributes=object())
139
140 self.assertEqual(len(self._tc), 0)
141
142 def test_create_invalid_user_attributes_value_type(self):
143 with self.assertRaisesRegex(
144 TypeError,
145 "'SignedIntegerValue' is not a '<class 'bt2.value.MapValue'>' object",
146 ):
147 self._tc.create_stream_class(user_attributes=23)
148
149 self.assertEqual(len(self._tc), 0)
150
151 def test_automatic_stream_ids(self):
152 sc = self._tc.create_stream_class(assigns_automatic_stream_id=True)
153 self.assertTrue(sc.assigns_automatic_stream_id)
154
155 stream = self._trace.create_stream(sc)
156 self.assertIsNotNone(stream.id)
157
158 def test_automatic_stream_ids_raises(self):
159 sc = self._tc.create_stream_class(assigns_automatic_stream_id=True)
160 self.assertTrue(sc.assigns_automatic_stream_id)
161
162 with self.assertRaisesRegex(
163 ValueError, "id provided, but stream class assigns automatic stream ids"
164 ):
165 self._trace.create_stream(sc, id=123)
166
167 self.assertEqual(len(self._trace), 0)
168
169 def test_automatic_stream_ids_wrong_type(self):
170 with self.assertRaisesRegex(TypeError, "str' is not a 'bool' object"):
171 self._tc.create_stream_class(assigns_automatic_stream_id='True')
172
173 self.assertEqual(len(self._tc), 0)
174
175 def test_no_automatic_stream_ids(self):
176 sc = self._tc.create_stream_class(assigns_automatic_stream_id=False)
177 self.assertFalse(sc.assigns_automatic_stream_id)
178
179 stream = self._trace.create_stream(sc, id=333)
180 self.assertEqual(stream.id, 333)
181
182 def test_no_automatic_stream_ids_raises(self):
183 sc = self._tc.create_stream_class(assigns_automatic_stream_id=False)
184 self.assertFalse(sc.assigns_automatic_stream_id)
185
186 with self.assertRaisesRegex(
187 ValueError,
188 "id not provided, but stream class does not assign automatic stream ids",
189 ):
190 self._trace.create_stream(sc)
191
192 self.assertEqual(len(self._trace), 0)
193
194 def test_automatic_event_class_ids(self):
195 sc = self._tc.create_stream_class(assigns_automatic_event_class_id=True)
196 self.assertTrue(sc.assigns_automatic_event_class_id)
197
198 ec = sc.create_event_class()
199 self.assertIsNotNone(ec.id)
200
201 def test_automatic_event_class_ids_raises(self):
202 sc = self._tc.create_stream_class(assigns_automatic_event_class_id=True)
203 self.assertTrue(sc.assigns_automatic_event_class_id)
204
205 with self.assertRaisesRegex(
206 ValueError,
207 "id provided, but stream class assigns automatic event class ids",
208 ):
209 sc.create_event_class(id=123)
210
211 self.assertEqual(len(sc), 0)
212
213 def test_automatic_event_class_ids_wrong_type(self):
214 with self.assertRaisesRegex(TypeError, "'str' is not a 'bool' object"):
215 self._tc.create_stream_class(assigns_automatic_event_class_id='True')
216
217 self.assertEqual(len(self._tc), 0)
218
219 def test_no_automatic_event_class_ids(self):
220 sc = self._tc.create_stream_class(assigns_automatic_event_class_id=False)
221 self.assertFalse(sc.assigns_automatic_event_class_id)
222
223 ec = sc.create_event_class(id=333)
224 self.assertEqual(ec.id, 333)
225
226 def test_no_automatic_event_class_ids_raises(self):
227 sc = self._tc.create_stream_class(assigns_automatic_event_class_id=False)
228 self.assertFalse(sc.assigns_automatic_event_class_id)
229
230 with self.assertRaisesRegex(
231 ValueError,
232 "id not provided, but stream class does not assign automatic event class ids",
233 ):
234 sc.create_event_class()
235
236 self.assertEqual(len(sc), 0)
237
238 def test_supports_packets_without_cs(self):
239 sc = self._tc.create_stream_class(
240 default_clock_class=self._cc, supports_packets=True
241 )
242 self.assertTrue(sc.supports_packets)
243 self.assertFalse(sc.packets_have_beginning_default_clock_snapshot)
244 self.assertFalse(sc.packets_have_end_default_clock_snapshot)
245
246 def test_supports_packets_with_begin_cs(self):
247 sc = self._tc.create_stream_class(
248 default_clock_class=self._cc,
249 supports_packets=True,
250 packets_have_beginning_default_clock_snapshot=True,
251 )
252 self.assertTrue(sc.supports_packets)
253 self.assertTrue(sc.packets_have_beginning_default_clock_snapshot)
254 self.assertFalse(sc.packets_have_end_default_clock_snapshot)
255
256 def test_supports_packets_with_end_cs(self):
257 sc = self._tc.create_stream_class(
258 default_clock_class=self._cc,
259 supports_packets=True,
260 packets_have_end_default_clock_snapshot=True,
261 )
262 self.assertTrue(sc.supports_packets)
263 self.assertFalse(sc.packets_have_beginning_default_clock_snapshot)
264 self.assertTrue(sc.packets_have_end_default_clock_snapshot)
265
266 def test_supports_packets_raises_type_error(self):
267 with self.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
268 self._tc.create_stream_class(
269 default_clock_class=self._cc, supports_packets=23
270 )
271
272 self.assertEqual(len(self._tc), 0)
273
274 def test_packets_have_begin_default_cs_raises_type_error(self):
275 with self.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
276 self._tc.create_stream_class(
277 default_clock_class=self._cc,
278 packets_have_beginning_default_clock_snapshot=23,
279 )
280
281 self.assertEqual(len(self._tc), 0)
282
283 def test_packets_have_end_default_cs_raises_type_error(self):
284 with self.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
285 self._tc.create_stream_class(
286 default_clock_class=self._cc, packets_have_end_default_clock_snapshot=23
287 )
288
289 self.assertEqual(len(self._tc), 0)
290
291 def test_does_not_support_packets_raises_with_begin_cs(self):
292 with self.assertRaisesRegex(
293 ValueError,
294 "cannot not support packets, but have packet beginning default clock snapshot",
295 ):
296 self._tc.create_stream_class(
297 default_clock_class=self._cc,
298 packets_have_beginning_default_clock_snapshot=True,
299 )
300
301 self.assertEqual(len(self._tc), 0)
302
303 def test_does_not_support_packets_raises_with_end_cs(self):
304 with self.assertRaisesRegex(
305 ValueError,
306 "cannot not support packets, but have packet end default clock snapshots",
307 ):
308 self._tc.create_stream_class(
309 default_clock_class=self._cc,
310 packets_have_end_default_clock_snapshot=True,
311 )
312
313 self.assertEqual(len(self._tc), 0)
314
315 def test_supports_discarded_events_without_cs(self):
316 sc = self._tc.create_stream_class(
317 default_clock_class=self._cc, supports_discarded_events=True
318 )
319 self.assertTrue(sc.supports_discarded_events)
320 self.assertFalse(sc.discarded_events_have_default_clock_snapshots)
321
322 def test_supports_discarded_events_with_cs(self):
323 sc = self._tc.create_stream_class(
324 default_clock_class=self._cc,
325 supports_discarded_events=True,
326 discarded_events_have_default_clock_snapshots=True,
327 )
328 self.assertTrue(sc.supports_discarded_events)
329 self.assertTrue(sc.discarded_events_have_default_clock_snapshots)
330
331 def test_supports_discarded_events_raises_type_error(self):
332 with self.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
333 self._tc.create_stream_class(
334 default_clock_class=self._cc, supports_discarded_events=23
335 )
336
337 self.assertEqual(len(self._tc), 0)
338
339 def test_discarded_events_have_default_cs_raises_type_error(self):
340 with self.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
341 self._tc.create_stream_class(
342 default_clock_class=self._cc,
343 discarded_events_have_default_clock_snapshots=23,
344 )
345
346 self.assertEqual(len(self._tc), 0)
347
348 def test_does_not_support_discarded_events_raises_with_cs(self):
349 with self.assertRaisesRegex(
350 ValueError,
351 "cannot not support discarded events, but have default clock snapshots for discarded event messages",
352 ):
353 self._tc.create_stream_class(
354 default_clock_class=self._cc,
355 discarded_events_have_default_clock_snapshots=True,
356 )
357
358 self.assertEqual(len(self._tc), 0)
359
360 def test_supports_discarded_events_with_clock_snapshots_without_default_clock_class_raises(
361 self,
362 ):
363 with self.assertRaisesRegex(
364 ValueError,
365 'cannot have no default clock class, but have default clock snapshots for discarded event messages',
366 ):
367 self._tc.create_stream_class(
368 supports_discarded_events=True,
369 discarded_events_have_default_clock_snapshots=True,
370 )
371
372 self.assertEqual(len(self._tc), 0)
373
374 def test_supports_discarded_packets_without_cs(self):
375 sc = self._tc.create_stream_class(
376 default_clock_class=self._cc,
377 supports_discarded_packets=True,
378 supports_packets=True,
379 )
380 self.assertTrue(sc.supports_discarded_packets)
381 self.assertFalse(sc.discarded_packets_have_default_clock_snapshots)
382
383 def test_supports_discarded_packets_with_cs(self):
384 sc = self._tc.create_stream_class(
385 default_clock_class=self._cc,
386 supports_discarded_packets=True,
387 discarded_packets_have_default_clock_snapshots=True,
388 supports_packets=True,
389 )
390 self.assertTrue(sc.supports_discarded_packets)
391 self.assertTrue(sc.discarded_packets_have_default_clock_snapshots)
392
393 def test_supports_discarded_packets_raises_without_packet_support(self):
394 with self.assertRaisesRegex(
395 ValueError, "cannot support discarded packets, but not support packets"
396 ):
397 self._tc.create_stream_class(
398 default_clock_class=self._cc, supports_discarded_packets=True
399 )
400
401 self.assertEqual(len(self._tc), 0)
402
403 def test_supports_discarded_packets_raises_type_error(self):
404 with self.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
405 self._tc.create_stream_class(
406 default_clock_class=self._cc,
407 supports_discarded_packets=23,
408 supports_packets=True,
409 )
410
411 self.assertEqual(len(self._tc), 0)
412
413 def test_discarded_packets_have_default_cs_raises_type_error(self):
414 with self.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
415 self._tc.create_stream_class(
416 default_clock_class=self._cc,
417 discarded_packets_have_default_clock_snapshots=23,
418 supports_packets=True,
419 )
420
421 self.assertEqual(len(self._tc), 0)
422
423 def test_does_not_support_discarded_packets_raises_with_cs(self):
424 with self.assertRaisesRegex(
425 ValueError,
426 "cannot not support discarded packets, but have default clock snapshots for discarded packet messages",
427 ):
428 self._tc.create_stream_class(
429 default_clock_class=self._cc,
430 discarded_packets_have_default_clock_snapshots=True,
431 supports_packets=True,
432 )
433
434 self.assertEqual(len(self._tc), 0)
435
436 def test_supports_discarded_packets_with_clock_snapshots_without_default_clock_class_raises(
437 self,
438 ):
439 with self.assertRaisesRegex(
440 ValueError,
441 'cannot have no default clock class, but have default clock snapshots for discarded packet messages',
442 ):
443 self._tc.create_stream_class(
444 supports_packets=True,
445 supports_discarded_packets=True,
446 discarded_packets_have_default_clock_snapshots=True,
447 )
448
449 self.assertEqual(len(self._tc), 0)
450
451 def test_trace_class(self):
452 sc = self._tc.create_stream_class()
453 self.assertEqual(sc.trace_class.addr, self._tc.addr)
454 self.assertIs(type(sc.trace_class), bt2_trace_class._TraceClass)
455
456 def _create_stream_class_with_event_classes(self):
457 sc = self._tc.create_stream_class(assigns_automatic_event_class_id=False)
458 ec1 = sc.create_event_class(id=23)
459 ec2 = sc.create_event_class(id=17)
460 return sc, ec1, ec2
461
462 def test_getitem(self):
463 sc, ec1, ec2 = self._create_stream_class_with_event_classes()
464
465 self.assertEqual(sc[23].addr, ec1.addr)
466 self.assertEqual(type(sc[23]), bt2_event_class._EventClass)
467 self.assertEqual(sc[17].addr, ec2.addr)
468 self.assertEqual(type(sc[17]), bt2_event_class._EventClass)
469
470 def test_getitem_wrong_key_type(self):
471 sc, _, _ = self._create_stream_class_with_event_classes()
472
473 with self.assertRaisesRegex(TypeError, "'str' is not an 'int' object"):
474 sc['event23']
475
476 def test_getitem_wrong_key(self):
477 sc, _, _ = self._create_stream_class_with_event_classes()
478
479 with self.assertRaisesRegex(KeyError, '19'):
480 sc[19]
481
482 def test_len(self):
483 sc, _, _ = self._create_stream_class_with_event_classes()
484
485 self.assertEqual(len(sc), 2)
486
487 def test_iter(self):
488 sc, _, _ = self._create_stream_class_with_event_classes()
489
490 ec_ids = sorted(sc)
491 self.assertEqual(ec_ids, [17, 23])
492
493
494 if __name__ == '__main__':
495 unittest.main()
This page took 0.052136 seconds and 4 git commands to generate.