1 # SPDX-License-Identifier: GPL-2.0-only
3 # Copyright (C) 2019 EfficiOS Inc.
7 from utils
import run_in_component_init
8 from bt2
import stream_class
as bt2_stream_class
9 from bt2
import trace_class
as bt2_trace_class
10 from bt2
import clock_class
as bt2_clock_class
11 from bt2
import event_class
as bt2_event_class
12 from bt2
import field_class
as bt2_field_class
15 class StreamClassTestCase(unittest
.TestCase
):
18 tc
= comp_self
._create
_trace
_class
(assigns_automatic_stream_class_id
=True)
19 cc
= comp_self
._create
_clock
_class
()
22 self
._tc
, self
._cc
= run_in_component_init(f
)
23 self
._trace
= self
._tc
()
25 def test_create_default(self
):
26 sc
= self
._tc
.create_stream_class()
28 self
.assertIs(type(sc
), bt2_stream_class
._StreamClass
)
29 self
.assertIsNone(sc
.name
)
30 self
.assertIsNone(sc
.packet_context_field_class
)
31 self
.assertIsNone(sc
.event_common_context_field_class
)
32 self
.assertIsNone(sc
.default_clock_class
)
33 self
.assertTrue(sc
.assigns_automatic_event_class_id
)
34 self
.assertTrue(sc
.assigns_automatic_stream_id
)
35 self
.assertFalse(sc
.supports_packets
)
36 self
.assertFalse(sc
.packets_have_beginning_default_clock_snapshot
)
37 self
.assertFalse(sc
.packets_have_end_default_clock_snapshot
)
38 self
.assertFalse(sc
.supports_discarded_events
)
39 self
.assertFalse(sc
.discarded_events_have_default_clock_snapshots
)
40 self
.assertFalse(sc
.supports_discarded_packets
)
41 self
.assertFalse(sc
.discarded_packets_have_default_clock_snapshots
)
42 self
.assertEqual(len(sc
.user_attributes
), 0)
44 def test_create_name(self
):
45 sc
= self
._tc
.create_stream_class(name
='bozo')
46 self
.assertEqual(sc
.name
, 'bozo')
48 def test_create_invalid_name(self
):
49 with self
.assertRaisesRegex(TypeError, "'int' is not a 'str' object"):
50 self
._tc
.create_stream_class(name
=17)
52 self
.assertEqual(len(self
._tc
), 0)
54 def test_create_packet_context_field_class(self
):
55 fc
= self
._tc
.create_structure_field_class()
56 sc
= self
._tc
.create_stream_class(
57 packet_context_field_class
=fc
, supports_packets
=True
59 self
.assertEqual(sc
.packet_context_field_class
, fc
)
61 type(sc
.packet_context_field_class
), bt2_field_class
._StructureFieldClass
64 def test_create_invalid_packet_context_field_class(self
):
65 with self
.assertRaisesRegex(
67 "'int' is not a '<class 'bt2.field_class._StructureFieldClass'>' object",
69 self
._tc
.create_stream_class(
70 packet_context_field_class
=22, supports_packets
=True
73 self
.assertEqual(len(self
._tc
), 0)
75 def test_create_invalid_packet_context_field_class_no_packets(self
):
76 fc
= self
._tc
.create_structure_field_class()
78 with self
.assertRaisesRegex(
80 "cannot have a packet context field class without supporting packets",
82 self
._tc
.create_stream_class(packet_context_field_class
=fc
)
84 self
.assertEqual(len(self
._tc
), 0)
86 def test_create_event_common_context_field_class(self
):
87 fc
= self
._tc
.create_structure_field_class()
88 sc
= self
._tc
.create_stream_class(event_common_context_field_class
=fc
)
89 self
.assertEqual(sc
.event_common_context_field_class
, fc
)
91 type(sc
.event_common_context_field_class
),
92 bt2_field_class
._StructureFieldClass
,
95 def test_create_invalid_event_common_context_field_class(self
):
96 with self
.assertRaisesRegex(
98 "'int' is not a '<class 'bt2.field_class._StructureFieldClass'>' object",
100 self
._tc
.create_stream_class(event_common_context_field_class
=22)
102 self
.assertEqual(len(self
._tc
), 0)
104 def test_create_default_clock_class(self
):
105 sc
= self
._tc
.create_stream_class(default_clock_class
=self
._cc
)
106 self
.assertEqual(sc
.default_clock_class
.addr
, self
._cc
.addr
)
107 self
.assertIs(type(sc
.default_clock_class
), bt2_clock_class
._ClockClass
)
109 def test_create_invalid_default_clock_class(self
):
110 with self
.assertRaisesRegex(
111 TypeError, "'int' is not a '<class 'bt2.clock_class._ClockClass'>' object"
113 self
._tc
.create_stream_class(default_clock_class
=12)
115 self
.assertEqual(len(self
._tc
), 0)
117 def test_create_user_attributes(self
):
118 sc
= self
._tc
.create_stream_class(user_attributes
={'salut': 23})
119 self
.assertEqual(sc
.user_attributes
, {'salut': 23})
121 def test_create_invalid_user_attributes(self
):
122 with self
.assertRaisesRegex(
123 TypeError, "cannot create value object from 'object' object"
125 self
._tc
.create_stream_class(user_attributes
=object())
127 self
.assertEqual(len(self
._tc
), 0)
129 def test_create_invalid_user_attributes_value_type(self
):
130 with self
.assertRaisesRegex(
132 "'SignedIntegerValue' is not a '<class 'bt2.value.MapValue'>' object",
134 self
._tc
.create_stream_class(user_attributes
=23)
136 self
.assertEqual(len(self
._tc
), 0)
138 def test_automatic_stream_ids(self
):
139 sc
= self
._tc
.create_stream_class(assigns_automatic_stream_id
=True)
140 self
.assertTrue(sc
.assigns_automatic_stream_id
)
142 stream
= self
._trace
.create_stream(sc
)
143 self
.assertIsNotNone(stream
.id)
145 def test_automatic_stream_ids_raises(self
):
146 sc
= self
._tc
.create_stream_class(assigns_automatic_stream_id
=True)
147 self
.assertTrue(sc
.assigns_automatic_stream_id
)
149 with self
.assertRaisesRegex(
150 ValueError, "id provided, but stream class assigns automatic stream ids"
152 self
._trace
.create_stream(sc
, id=123)
154 self
.assertEqual(len(self
._trace
), 0)
156 def test_automatic_stream_ids_wrong_type(self
):
157 with self
.assertRaisesRegex(TypeError, "str' is not a 'bool' object"):
158 self
._tc
.create_stream_class(assigns_automatic_stream_id
='True')
160 self
.assertEqual(len(self
._tc
), 0)
162 def test_no_automatic_stream_ids(self
):
163 sc
= self
._tc
.create_stream_class(assigns_automatic_stream_id
=False)
164 self
.assertFalse(sc
.assigns_automatic_stream_id
)
166 stream
= self
._trace
.create_stream(sc
, id=333)
167 self
.assertEqual(stream
.id, 333)
169 def test_no_automatic_stream_ids_raises(self
):
170 sc
= self
._tc
.create_stream_class(assigns_automatic_stream_id
=False)
171 self
.assertFalse(sc
.assigns_automatic_stream_id
)
173 with self
.assertRaisesRegex(
175 "id not provided, but stream class does not assign automatic stream ids",
177 self
._trace
.create_stream(sc
)
179 self
.assertEqual(len(self
._trace
), 0)
181 def test_automatic_event_class_ids(self
):
182 sc
= self
._tc
.create_stream_class(assigns_automatic_event_class_id
=True)
183 self
.assertTrue(sc
.assigns_automatic_event_class_id
)
185 ec
= sc
.create_event_class()
186 self
.assertIsNotNone(ec
.id)
188 def test_automatic_event_class_ids_raises(self
):
189 sc
= self
._tc
.create_stream_class(assigns_automatic_event_class_id
=True)
190 self
.assertTrue(sc
.assigns_automatic_event_class_id
)
192 with self
.assertRaisesRegex(
194 "id provided, but stream class assigns automatic event class ids",
196 sc
.create_event_class(id=123)
198 self
.assertEqual(len(sc
), 0)
200 def test_automatic_event_class_ids_wrong_type(self
):
201 with self
.assertRaisesRegex(TypeError, "'str' is not a 'bool' object"):
202 self
._tc
.create_stream_class(assigns_automatic_event_class_id
='True')
204 self
.assertEqual(len(self
._tc
), 0)
206 def test_no_automatic_event_class_ids(self
):
207 sc
= self
._tc
.create_stream_class(assigns_automatic_event_class_id
=False)
208 self
.assertFalse(sc
.assigns_automatic_event_class_id
)
210 ec
= sc
.create_event_class(id=333)
211 self
.assertEqual(ec
.id, 333)
213 def test_no_automatic_event_class_ids_raises(self
):
214 sc
= self
._tc
.create_stream_class(assigns_automatic_event_class_id
=False)
215 self
.assertFalse(sc
.assigns_automatic_event_class_id
)
217 with self
.assertRaisesRegex(
219 "id not provided, but stream class does not assign automatic event class ids",
221 sc
.create_event_class()
223 self
.assertEqual(len(sc
), 0)
225 def test_supports_packets_without_cs(self
):
226 sc
= self
._tc
.create_stream_class(
227 default_clock_class
=self
._cc
, supports_packets
=True
229 self
.assertTrue(sc
.supports_packets
)
230 self
.assertFalse(sc
.packets_have_beginning_default_clock_snapshot
)
231 self
.assertFalse(sc
.packets_have_end_default_clock_snapshot
)
233 def test_supports_packets_with_begin_cs(self
):
234 sc
= self
._tc
.create_stream_class(
235 default_clock_class
=self
._cc
,
236 supports_packets
=True,
237 packets_have_beginning_default_clock_snapshot
=True,
239 self
.assertTrue(sc
.supports_packets
)
240 self
.assertTrue(sc
.packets_have_beginning_default_clock_snapshot
)
241 self
.assertFalse(sc
.packets_have_end_default_clock_snapshot
)
243 def test_supports_packets_with_end_cs(self
):
244 sc
= self
._tc
.create_stream_class(
245 default_clock_class
=self
._cc
,
246 supports_packets
=True,
247 packets_have_end_default_clock_snapshot
=True,
249 self
.assertTrue(sc
.supports_packets
)
250 self
.assertFalse(sc
.packets_have_beginning_default_clock_snapshot
)
251 self
.assertTrue(sc
.packets_have_end_default_clock_snapshot
)
253 def test_supports_packets_raises_type_error(self
):
254 with self
.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
255 self
._tc
.create_stream_class(
256 default_clock_class
=self
._cc
, supports_packets
=23
259 self
.assertEqual(len(self
._tc
), 0)
261 def test_packets_have_begin_default_cs_raises_type_error(self
):
262 with self
.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
263 self
._tc
.create_stream_class(
264 default_clock_class
=self
._cc
,
265 packets_have_beginning_default_clock_snapshot
=23,
268 self
.assertEqual(len(self
._tc
), 0)
270 def test_packets_have_end_default_cs_raises_type_error(self
):
271 with self
.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
272 self
._tc
.create_stream_class(
273 default_clock_class
=self
._cc
, packets_have_end_default_clock_snapshot
=23
276 self
.assertEqual(len(self
._tc
), 0)
278 def test_does_not_support_packets_raises_with_begin_cs(self
):
279 with self
.assertRaisesRegex(
281 "cannot not support packets, but have packet beginning default clock snapshot",
283 self
._tc
.create_stream_class(
284 default_clock_class
=self
._cc
,
285 packets_have_beginning_default_clock_snapshot
=True,
288 self
.assertEqual(len(self
._tc
), 0)
290 def test_does_not_support_packets_raises_with_end_cs(self
):
291 with self
.assertRaisesRegex(
293 "cannot not support packets, but have packet end default clock snapshots",
295 self
._tc
.create_stream_class(
296 default_clock_class
=self
._cc
,
297 packets_have_end_default_clock_snapshot
=True,
300 self
.assertEqual(len(self
._tc
), 0)
302 def test_supports_discarded_events_without_cs(self
):
303 sc
= self
._tc
.create_stream_class(
304 default_clock_class
=self
._cc
, supports_discarded_events
=True
306 self
.assertTrue(sc
.supports_discarded_events
)
307 self
.assertFalse(sc
.discarded_events_have_default_clock_snapshots
)
309 def test_supports_discarded_events_with_cs(self
):
310 sc
= self
._tc
.create_stream_class(
311 default_clock_class
=self
._cc
,
312 supports_discarded_events
=True,
313 discarded_events_have_default_clock_snapshots
=True,
315 self
.assertTrue(sc
.supports_discarded_events
)
316 self
.assertTrue(sc
.discarded_events_have_default_clock_snapshots
)
318 def test_supports_discarded_events_raises_type_error(self
):
319 with self
.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
320 self
._tc
.create_stream_class(
321 default_clock_class
=self
._cc
, supports_discarded_events
=23
324 self
.assertEqual(len(self
._tc
), 0)
326 def test_discarded_events_have_default_cs_raises_type_error(self
):
327 with self
.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
328 self
._tc
.create_stream_class(
329 default_clock_class
=self
._cc
,
330 discarded_events_have_default_clock_snapshots
=23,
333 self
.assertEqual(len(self
._tc
), 0)
335 def test_does_not_support_discarded_events_raises_with_cs(self
):
336 with self
.assertRaisesRegex(
338 "cannot not support discarded events, but have default clock snapshots for discarded event messages",
340 self
._tc
.create_stream_class(
341 default_clock_class
=self
._cc
,
342 discarded_events_have_default_clock_snapshots
=True,
345 self
.assertEqual(len(self
._tc
), 0)
347 def test_supports_discarded_packets_without_cs(self
):
348 sc
= self
._tc
.create_stream_class(
349 default_clock_class
=self
._cc
,
350 supports_discarded_packets
=True,
351 supports_packets
=True,
353 self
.assertTrue(sc
.supports_discarded_packets
)
354 self
.assertFalse(sc
.discarded_packets_have_default_clock_snapshots
)
356 def test_supports_discarded_packets_with_cs(self
):
357 sc
= self
._tc
.create_stream_class(
358 default_clock_class
=self
._cc
,
359 supports_discarded_packets
=True,
360 discarded_packets_have_default_clock_snapshots
=True,
361 supports_packets
=True,
363 self
.assertTrue(sc
.supports_discarded_packets
)
364 self
.assertTrue(sc
.discarded_packets_have_default_clock_snapshots
)
366 def test_supports_discarded_packets_raises_without_packet_support(self
):
367 with self
.assertRaisesRegex(
368 ValueError, "cannot support discarded packets, but not support packets"
370 self
._tc
.create_stream_class(
371 default_clock_class
=self
._cc
, supports_discarded_packets
=True
374 self
.assertEqual(len(self
._tc
), 0)
376 def test_supports_discarded_packets_raises_type_error(self
):
377 with self
.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
378 self
._tc
.create_stream_class(
379 default_clock_class
=self
._cc
,
380 supports_discarded_packets
=23,
381 supports_packets
=True,
384 self
.assertEqual(len(self
._tc
), 0)
386 def test_discarded_packets_have_default_cs_raises_type_error(self
):
387 with self
.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
388 self
._tc
.create_stream_class(
389 default_clock_class
=self
._cc
,
390 discarded_packets_have_default_clock_snapshots
=23,
391 supports_packets
=True,
394 self
.assertEqual(len(self
._tc
), 0)
396 def test_does_not_support_discarded_packets_raises_with_cs(self
):
397 with self
.assertRaisesRegex(
399 "cannot not support discarded packets, but have default clock snapshots for discarded packet messages",
401 self
._tc
.create_stream_class(
402 default_clock_class
=self
._cc
,
403 discarded_packets_have_default_clock_snapshots
=True,
404 supports_packets
=True,
407 self
.assertEqual(len(self
._tc
), 0)
409 def test_trace_class(self
):
410 sc
= self
._tc
.create_stream_class()
411 self
.assertEqual(sc
.trace_class
.addr
, self
._tc
.addr
)
412 self
.assertIs(type(sc
.trace_class
), bt2_trace_class
._TraceClass
)
414 def _create_stream_class_with_event_classes(self
):
415 sc
= self
._tc
.create_stream_class(assigns_automatic_event_class_id
=False)
416 ec1
= sc
.create_event_class(id=23)
417 ec2
= sc
.create_event_class(id=17)
420 def test_getitem(self
):
421 sc
, ec1
, ec2
= self
._create
_stream
_class
_with
_event
_classes
()
423 self
.assertEqual(sc
[23].addr
, ec1
.addr
)
424 self
.assertEqual(type(sc
[23]), bt2_event_class
._EventClass
)
425 self
.assertEqual(sc
[17].addr
, ec2
.addr
)
426 self
.assertEqual(type(sc
[17]), bt2_event_class
._EventClass
)
428 def test_getitem_wrong_key_type(self
):
429 sc
, _
, _
= self
._create
_stream
_class
_with
_event
_classes
()
431 with self
.assertRaisesRegex(TypeError, "'str' is not an 'int' object"):
434 def test_getitem_wrong_key(self
):
435 sc
, _
, _
= self
._create
_stream
_class
_with
_event
_classes
()
437 with self
.assertRaisesRegex(KeyError, '19'):
441 sc
, _
, _
= self
._create
_stream
_class
_with
_event
_classes
()
443 self
.assertEqual(len(sc
), 2)
446 sc
, _
, _
= self
._create
_stream
_class
_with
_event
_classes
()
449 self
.assertEqual(ec_ids
, [17, 23])
452 if __name__
== '__main__':