1 # SPDX-License-Identifier: GPL-2.0-only
3 # Copyright (C) 2019 EfficiOS Inc.
8 from bt2
import clock_class
as bt2_clock_class
9 from bt2
import event_class
as bt2_event_class
10 from bt2
import field_class
as bt2_field_class
11 from bt2
import trace_class
as bt2_trace_class
12 from bt2
import stream_class
as bt2_stream_class
13 from utils
import run_in_component_init
16 class StreamClassTestCase(unittest
.TestCase
):
19 tc
= comp_self
._create
_trace
_class
(assigns_automatic_stream_class_id
=True)
20 cc
= comp_self
._create
_clock
_class
()
23 self
._tc
, self
._cc
= run_in_component_init(f
)
24 self
._trace
= self
._tc
()
26 def test_create_default(self
):
27 sc
= self
._tc
.create_stream_class()
29 self
.assertIs(type(sc
), bt2_stream_class
._StreamClass
)
30 self
.assertIsNone(sc
.name
)
31 self
.assertIsNone(sc
.packet_context_field_class
)
32 self
.assertIsNone(sc
.event_common_context_field_class
)
33 self
.assertIsNone(sc
.default_clock_class
)
34 self
.assertTrue(sc
.assigns_automatic_event_class_id
)
35 self
.assertTrue(sc
.assigns_automatic_stream_id
)
36 self
.assertFalse(sc
.supports_packets
)
37 self
.assertFalse(sc
.packets_have_beginning_default_clock_snapshot
)
38 self
.assertFalse(sc
.packets_have_end_default_clock_snapshot
)
39 self
.assertFalse(sc
.supports_discarded_events
)
40 self
.assertFalse(sc
.discarded_events_have_default_clock_snapshots
)
41 self
.assertFalse(sc
.supports_discarded_packets
)
42 self
.assertFalse(sc
.discarded_packets_have_default_clock_snapshots
)
43 self
.assertEqual(len(sc
.user_attributes
), 0)
45 def test_create_name(self
):
46 sc
= self
._tc
.create_stream_class(name
="bozo")
47 self
.assertEqual(sc
.name
, "bozo")
49 def test_create_invalid_name(self
):
50 with self
.assertRaisesRegex(TypeError, "'int' is not a 'str' object"):
51 self
._tc
.create_stream_class(name
=17)
53 self
.assertEqual(len(self
._tc
), 0)
55 def test_create_packet_context_field_class(self
):
56 fc
= self
._tc
.create_structure_field_class()
57 sc
= self
._tc
.create_stream_class(
58 packet_context_field_class
=fc
, supports_packets
=True
60 self
.assertEqual(sc
.packet_context_field_class
, fc
)
62 type(sc
.packet_context_field_class
), bt2_field_class
._StructureFieldClass
65 def test_create_invalid_packet_context_field_class(self
):
66 with self
.assertRaisesRegex(
68 "'int' is not a '<class 'bt2.field_class._StructureFieldClass'>' object",
70 self
._tc
.create_stream_class(
71 packet_context_field_class
=22, supports_packets
=True
74 self
.assertEqual(len(self
._tc
), 0)
76 def test_create_invalid_packet_context_field_class_no_packets(self
):
77 fc
= self
._tc
.create_structure_field_class()
79 with self
.assertRaisesRegex(
81 "cannot have a packet context field class without supporting packets",
83 self
._tc
.create_stream_class(packet_context_field_class
=fc
)
85 self
.assertEqual(len(self
._tc
), 0)
87 def test_create_event_common_context_field_class(self
):
88 fc
= self
._tc
.create_structure_field_class()
89 sc
= self
._tc
.create_stream_class(event_common_context_field_class
=fc
)
90 self
.assertEqual(sc
.event_common_context_field_class
, fc
)
92 type(sc
.event_common_context_field_class
),
93 bt2_field_class
._StructureFieldClass
,
96 def test_create_invalid_event_common_context_field_class(self
):
97 with self
.assertRaisesRegex(
99 "'int' is not a '<class 'bt2.field_class._StructureFieldClass'>' object",
101 self
._tc
.create_stream_class(event_common_context_field_class
=22)
103 self
.assertEqual(len(self
._tc
), 0)
105 def test_create_default_clock_class(self
):
106 sc
= self
._tc
.create_stream_class(default_clock_class
=self
._cc
)
107 self
.assertEqual(sc
.default_clock_class
.addr
, self
._cc
.addr
)
108 self
.assertIs(type(sc
.default_clock_class
), bt2_clock_class
._ClockClass
)
110 def test_create_invalid_default_clock_class(self
):
111 with self
.assertRaisesRegex(
112 TypeError, "'int' is not a '<class 'bt2.clock_class._ClockClass'>' object"
114 self
._tc
.create_stream_class(default_clock_class
=12)
116 self
.assertEqual(len(self
._tc
), 0)
118 def test_create_user_attributes(self
):
119 sc
= self
._tc
.create_stream_class(user_attributes
={"salut": 23})
120 self
.assertEqual(sc
.user_attributes
, {"salut": 23})
122 def test_create_invalid_user_attributes(self
):
123 with self
.assertRaisesRegex(
124 TypeError, "cannot create value object from 'object' object"
126 self
._tc
.create_stream_class(user_attributes
=object())
128 self
.assertEqual(len(self
._tc
), 0)
130 def test_create_invalid_user_attributes_value_type(self
):
131 with self
.assertRaisesRegex(
133 "'SignedIntegerValue' is not a '<class 'bt2.value.MapValue'>' object",
135 self
._tc
.create_stream_class(user_attributes
=23)
137 self
.assertEqual(len(self
._tc
), 0)
139 def test_automatic_stream_ids(self
):
140 sc
= self
._tc
.create_stream_class(assigns_automatic_stream_id
=True)
141 self
.assertTrue(sc
.assigns_automatic_stream_id
)
143 stream
= self
._trace
.create_stream(sc
)
144 self
.assertIsNotNone(stream
.id)
146 def test_automatic_stream_ids_raises(self
):
147 sc
= self
._tc
.create_stream_class(assigns_automatic_stream_id
=True)
148 self
.assertTrue(sc
.assigns_automatic_stream_id
)
150 with self
.assertRaisesRegex(
151 ValueError, "id provided, but stream class assigns automatic stream ids"
153 self
._trace
.create_stream(sc
, id=123)
155 self
.assertEqual(len(self
._trace
), 0)
157 def test_automatic_stream_ids_wrong_type(self
):
158 with self
.assertRaisesRegex(TypeError, "str' is not a 'bool' object"):
159 self
._tc
.create_stream_class(assigns_automatic_stream_id
="True")
161 self
.assertEqual(len(self
._tc
), 0)
163 def test_no_automatic_stream_ids(self
):
164 sc
= self
._tc
.create_stream_class(assigns_automatic_stream_id
=False)
165 self
.assertFalse(sc
.assigns_automatic_stream_id
)
167 stream
= self
._trace
.create_stream(sc
, id=333)
168 self
.assertEqual(stream
.id, 333)
170 def test_no_automatic_stream_ids_raises(self
):
171 sc
= self
._tc
.create_stream_class(assigns_automatic_stream_id
=False)
172 self
.assertFalse(sc
.assigns_automatic_stream_id
)
174 with self
.assertRaisesRegex(
176 "id not provided, but stream class does not assign automatic stream ids",
178 self
._trace
.create_stream(sc
)
180 self
.assertEqual(len(self
._trace
), 0)
182 def test_automatic_event_class_ids(self
):
183 sc
= self
._tc
.create_stream_class(assigns_automatic_event_class_id
=True)
184 self
.assertTrue(sc
.assigns_automatic_event_class_id
)
186 ec
= sc
.create_event_class()
187 self
.assertIsNotNone(ec
.id)
189 def test_automatic_event_class_ids_raises(self
):
190 sc
= self
._tc
.create_stream_class(assigns_automatic_event_class_id
=True)
191 self
.assertTrue(sc
.assigns_automatic_event_class_id
)
193 with self
.assertRaisesRegex(
195 "id provided, but stream class assigns automatic event class ids",
197 sc
.create_event_class(id=123)
199 self
.assertEqual(len(sc
), 0)
201 def test_automatic_event_class_ids_wrong_type(self
):
202 with self
.assertRaisesRegex(TypeError, "'str' is not a 'bool' object"):
203 self
._tc
.create_stream_class(assigns_automatic_event_class_id
="True")
205 self
.assertEqual(len(self
._tc
), 0)
207 def test_no_automatic_event_class_ids(self
):
208 sc
= self
._tc
.create_stream_class(assigns_automatic_event_class_id
=False)
209 self
.assertFalse(sc
.assigns_automatic_event_class_id
)
211 ec
= sc
.create_event_class(id=333)
212 self
.assertEqual(ec
.id, 333)
214 def test_no_automatic_event_class_ids_raises(self
):
215 sc
= self
._tc
.create_stream_class(assigns_automatic_event_class_id
=False)
216 self
.assertFalse(sc
.assigns_automatic_event_class_id
)
218 with self
.assertRaisesRegex(
220 "id not provided, but stream class does not assign automatic event class ids",
222 sc
.create_event_class()
224 self
.assertEqual(len(sc
), 0)
226 def test_supports_packets_without_cs(self
):
227 sc
= self
._tc
.create_stream_class(
228 default_clock_class
=self
._cc
, supports_packets
=True
230 self
.assertTrue(sc
.supports_packets
)
231 self
.assertFalse(sc
.packets_have_beginning_default_clock_snapshot
)
232 self
.assertFalse(sc
.packets_have_end_default_clock_snapshot
)
234 def test_supports_packets_with_begin_cs(self
):
235 sc
= self
._tc
.create_stream_class(
236 default_clock_class
=self
._cc
,
237 supports_packets
=True,
238 packets_have_beginning_default_clock_snapshot
=True,
240 self
.assertTrue(sc
.supports_packets
)
241 self
.assertTrue(sc
.packets_have_beginning_default_clock_snapshot
)
242 self
.assertFalse(sc
.packets_have_end_default_clock_snapshot
)
244 def test_supports_packets_with_end_cs(self
):
245 sc
= self
._tc
.create_stream_class(
246 default_clock_class
=self
._cc
,
247 supports_packets
=True,
248 packets_have_end_default_clock_snapshot
=True,
250 self
.assertTrue(sc
.supports_packets
)
251 self
.assertFalse(sc
.packets_have_beginning_default_clock_snapshot
)
252 self
.assertTrue(sc
.packets_have_end_default_clock_snapshot
)
254 def test_supports_packets_raises_type_error(self
):
255 with self
.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
256 self
._tc
.create_stream_class(
257 default_clock_class
=self
._cc
, supports_packets
=23
260 self
.assertEqual(len(self
._tc
), 0)
262 def test_packets_have_begin_default_cs_raises_type_error(self
):
263 with self
.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
264 self
._tc
.create_stream_class(
265 default_clock_class
=self
._cc
,
266 packets_have_beginning_default_clock_snapshot
=23,
269 self
.assertEqual(len(self
._tc
), 0)
271 def test_packets_have_end_default_cs_raises_type_error(self
):
272 with self
.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
273 self
._tc
.create_stream_class(
274 default_clock_class
=self
._cc
, packets_have_end_default_clock_snapshot
=23
277 self
.assertEqual(len(self
._tc
), 0)
279 def test_does_not_support_packets_raises_with_begin_cs(self
):
280 with self
.assertRaisesRegex(
282 "cannot not support packets, but have packet beginning default clock snapshot",
284 self
._tc
.create_stream_class(
285 default_clock_class
=self
._cc
,
286 packets_have_beginning_default_clock_snapshot
=True,
289 self
.assertEqual(len(self
._tc
), 0)
291 def test_does_not_support_packets_raises_with_end_cs(self
):
292 with self
.assertRaisesRegex(
294 "cannot not support packets, but have packet end default clock snapshots",
296 self
._tc
.create_stream_class(
297 default_clock_class
=self
._cc
,
298 packets_have_end_default_clock_snapshot
=True,
301 self
.assertEqual(len(self
._tc
), 0)
303 def test_supports_discarded_events_without_cs(self
):
304 sc
= self
._tc
.create_stream_class(
305 default_clock_class
=self
._cc
, supports_discarded_events
=True
307 self
.assertTrue(sc
.supports_discarded_events
)
308 self
.assertFalse(sc
.discarded_events_have_default_clock_snapshots
)
310 def test_supports_discarded_events_with_cs(self
):
311 sc
= self
._tc
.create_stream_class(
312 default_clock_class
=self
._cc
,
313 supports_discarded_events
=True,
314 discarded_events_have_default_clock_snapshots
=True,
316 self
.assertTrue(sc
.supports_discarded_events
)
317 self
.assertTrue(sc
.discarded_events_have_default_clock_snapshots
)
319 def test_supports_discarded_events_raises_type_error(self
):
320 with self
.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
321 self
._tc
.create_stream_class(
322 default_clock_class
=self
._cc
, supports_discarded_events
=23
325 self
.assertEqual(len(self
._tc
), 0)
327 def test_discarded_events_have_default_cs_raises_type_error(self
):
328 with self
.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
329 self
._tc
.create_stream_class(
330 default_clock_class
=self
._cc
,
331 discarded_events_have_default_clock_snapshots
=23,
334 self
.assertEqual(len(self
._tc
), 0)
336 def test_does_not_support_discarded_events_raises_with_cs(self
):
337 with self
.assertRaisesRegex(
339 "cannot not support discarded events, but have default clock snapshots for discarded event messages",
341 self
._tc
.create_stream_class(
342 default_clock_class
=self
._cc
,
343 discarded_events_have_default_clock_snapshots
=True,
346 self
.assertEqual(len(self
._tc
), 0)
348 def test_supports_discarded_events_with_clock_snapshots_without_default_clock_class_raises(
351 with self
.assertRaisesRegex(
353 "cannot have no default clock class, but have default clock snapshots for discarded event messages",
355 self
._tc
.create_stream_class(
356 supports_discarded_events
=True,
357 discarded_events_have_default_clock_snapshots
=True,
360 self
.assertEqual(len(self
._tc
), 0)
362 def test_supports_discarded_packets_without_cs(self
):
363 sc
= self
._tc
.create_stream_class(
364 default_clock_class
=self
._cc
,
365 supports_discarded_packets
=True,
366 supports_packets
=True,
368 self
.assertTrue(sc
.supports_discarded_packets
)
369 self
.assertFalse(sc
.discarded_packets_have_default_clock_snapshots
)
371 def test_supports_discarded_packets_with_cs(self
):
372 sc
= self
._tc
.create_stream_class(
373 default_clock_class
=self
._cc
,
374 supports_discarded_packets
=True,
375 discarded_packets_have_default_clock_snapshots
=True,
376 supports_packets
=True,
378 self
.assertTrue(sc
.supports_discarded_packets
)
379 self
.assertTrue(sc
.discarded_packets_have_default_clock_snapshots
)
381 def test_supports_discarded_packets_raises_without_packet_support(self
):
382 with self
.assertRaisesRegex(
383 ValueError, "cannot support discarded packets, but not support packets"
385 self
._tc
.create_stream_class(
386 default_clock_class
=self
._cc
, supports_discarded_packets
=True
389 self
.assertEqual(len(self
._tc
), 0)
391 def test_supports_discarded_packets_raises_type_error(self
):
392 with self
.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
393 self
._tc
.create_stream_class(
394 default_clock_class
=self
._cc
,
395 supports_discarded_packets
=23,
396 supports_packets
=True,
399 self
.assertEqual(len(self
._tc
), 0)
401 def test_discarded_packets_have_default_cs_raises_type_error(self
):
402 with self
.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
403 self
._tc
.create_stream_class(
404 default_clock_class
=self
._cc
,
405 discarded_packets_have_default_clock_snapshots
=23,
406 supports_packets
=True,
409 self
.assertEqual(len(self
._tc
), 0)
411 def test_does_not_support_discarded_packets_raises_with_cs(self
):
412 with self
.assertRaisesRegex(
414 "cannot not support discarded packets, but have default clock snapshots for discarded packet messages",
416 self
._tc
.create_stream_class(
417 default_clock_class
=self
._cc
,
418 discarded_packets_have_default_clock_snapshots
=True,
419 supports_packets
=True,
422 self
.assertEqual(len(self
._tc
), 0)
424 def test_supports_discarded_packets_with_clock_snapshots_without_default_clock_class_raises(
427 with self
.assertRaisesRegex(
429 "cannot have no default clock class, but have default clock snapshots for discarded packet messages",
431 self
._tc
.create_stream_class(
432 supports_packets
=True,
433 supports_discarded_packets
=True,
434 discarded_packets_have_default_clock_snapshots
=True,
437 self
.assertEqual(len(self
._tc
), 0)
439 def test_trace_class(self
):
440 sc
= self
._tc
.create_stream_class()
441 self
.assertEqual(sc
.trace_class
.addr
, self
._tc
.addr
)
442 self
.assertIs(type(sc
.trace_class
), bt2_trace_class
._TraceClass
)
444 def _create_stream_class_with_event_classes(self
):
445 sc
= self
._tc
.create_stream_class(assigns_automatic_event_class_id
=False)
446 ec1
= sc
.create_event_class(id=23)
447 ec2
= sc
.create_event_class(id=17)
450 def test_getitem(self
):
451 sc
, ec1
, ec2
= self
._create
_stream
_class
_with
_event
_classes
()
453 self
.assertEqual(sc
[23].addr
, ec1
.addr
)
454 self
.assertEqual(type(sc
[23]), bt2_event_class
._EventClass
)
455 self
.assertEqual(sc
[17].addr
, ec2
.addr
)
456 self
.assertEqual(type(sc
[17]), bt2_event_class
._EventClass
)
458 def test_getitem_wrong_key_type(self
):
459 sc
, _
, _
= self
._create
_stream
_class
_with
_event
_classes
()
461 with self
.assertRaisesRegex(TypeError, "'str' is not an 'int' object"):
464 def test_getitem_wrong_key(self
):
465 sc
, _
, _
= self
._create
_stream
_class
_with
_event
_classes
()
467 with self
.assertRaisesRegex(KeyError, "19"):
471 sc
, _
, _
= self
._create
_stream
_class
_with
_event
_classes
()
473 self
.assertEqual(len(sc
), 2)
476 sc
, _
, _
= self
._create
_stream
_class
_with
_event
_classes
()
479 self
.assertEqual(ec_ids
, [17, 23])
482 if __name__
== "__main__":