2 # Copyright (C) 2019 EfficiOS Inc.
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
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.
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.
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
28 class StreamClassTestCase(unittest
.TestCase
):
31 tc
= comp_self
._create
_trace
_class
(assigns_automatic_stream_class_id
=True)
32 cc
= comp_self
._create
_clock
_class
()
35 self
._tc
, self
._cc
= run_in_component_init(f
)
36 self
._trace
= self
._tc
()
38 def test_create_default(self
):
39 sc
= self
._tc
.create_stream_class()
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)
57 def test_create_name(self
):
58 sc
= self
._tc
.create_stream_class(name
='bozo')
59 self
.assertEqual(sc
.name
, 'bozo')
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)
65 self
.assertEqual(len(self
._tc
), 0)
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
72 self
.assertEqual(sc
.packet_context_field_class
, fc
)
74 type(sc
.packet_context_field_class
), bt2_field_class
._StructureFieldClass
77 def test_create_invalid_packet_context_field_class(self
):
78 with self
.assertRaisesRegex(
80 "'int' is not a '<class 'bt2.field_class._StructureFieldClass'>' object",
82 self
._tc
.create_stream_class(
83 packet_context_field_class
=22, supports_packets
=True
86 self
.assertEqual(len(self
._tc
), 0)
88 def test_create_invalid_packet_context_field_class_no_packets(self
):
89 fc
= self
._tc
.create_structure_field_class()
91 with self
.assertRaisesRegex(
93 "cannot have a packet context field class without supporting packets",
95 self
._tc
.create_stream_class(packet_context_field_class
=fc
)
97 self
.assertEqual(len(self
._tc
), 0)
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
)
104 type(sc
.event_common_context_field_class
),
105 bt2_field_class
._StructureFieldClass
,
108 def test_create_invalid_event_common_context_field_class(self
):
109 with self
.assertRaisesRegex(
111 "'int' is not a '<class 'bt2.field_class._StructureFieldClass'>' object",
113 self
._tc
.create_stream_class(event_common_context_field_class
=22)
115 self
.assertEqual(len(self
._tc
), 0)
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
)
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"
126 self
._tc
.create_stream_class(default_clock_class
=12)
128 self
.assertEqual(len(self
._tc
), 0)
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})
134 def test_create_invalid_user_attributes(self
):
135 with self
.assertRaisesRegex(
136 TypeError, "cannot create value object from 'object' object"
138 self
._tc
.create_stream_class(user_attributes
=object())
140 self
.assertEqual(len(self
._tc
), 0)
142 def test_create_invalid_user_attributes_value_type(self
):
143 with self
.assertRaisesRegex(
145 "'SignedIntegerValue' is not a '<class 'bt2.value.MapValue'>' object",
147 self
._tc
.create_stream_class(user_attributes
=23)
149 self
.assertEqual(len(self
._tc
), 0)
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
)
155 stream
= self
._trace
.create_stream(sc
)
156 self
.assertIsNotNone(stream
.id)
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
)
162 with self
.assertRaisesRegex(
163 ValueError, "id provided, but stream class assigns automatic stream ids"
165 self
._trace
.create_stream(sc
, id=123)
167 self
.assertEqual(len(self
._trace
), 0)
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')
173 self
.assertEqual(len(self
._tc
), 0)
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
)
179 stream
= self
._trace
.create_stream(sc
, id=333)
180 self
.assertEqual(stream
.id, 333)
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
)
186 with self
.assertRaisesRegex(
188 "id not provided, but stream class does not assign automatic stream ids",
190 self
._trace
.create_stream(sc
)
192 self
.assertEqual(len(self
._trace
), 0)
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
)
198 ec
= sc
.create_event_class()
199 self
.assertIsNotNone(ec
.id)
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
)
205 with self
.assertRaisesRegex(
207 "id provided, but stream class assigns automatic event class ids",
209 sc
.create_event_class(id=123)
211 self
.assertEqual(len(sc
), 0)
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')
217 self
.assertEqual(len(self
._tc
), 0)
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
)
223 ec
= sc
.create_event_class(id=333)
224 self
.assertEqual(ec
.id, 333)
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
)
230 with self
.assertRaisesRegex(
232 "id not provided, but stream class does not assign automatic event class ids",
234 sc
.create_event_class()
236 self
.assertEqual(len(sc
), 0)
238 def test_supports_packets_without_cs(self
):
239 sc
= self
._tc
.create_stream_class(
240 default_clock_class
=self
._cc
, supports_packets
=True
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
)
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,
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
)
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,
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
)
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
272 self
.assertEqual(len(self
._tc
), 0)
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,
281 self
.assertEqual(len(self
._tc
), 0)
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
289 self
.assertEqual(len(self
._tc
), 0)
291 def test_does_not_support_packets_raises_with_begin_cs(self
):
292 with self
.assertRaisesRegex(
294 "cannot not support packets, but have packet beginning default clock snapshot",
296 self
._tc
.create_stream_class(
297 default_clock_class
=self
._cc
,
298 packets_have_beginning_default_clock_snapshot
=True,
301 self
.assertEqual(len(self
._tc
), 0)
303 def test_does_not_support_packets_raises_with_end_cs(self
):
304 with self
.assertRaisesRegex(
306 "cannot not support packets, but have packet end default clock snapshots",
308 self
._tc
.create_stream_class(
309 default_clock_class
=self
._cc
,
310 packets_have_end_default_clock_snapshot
=True,
313 self
.assertEqual(len(self
._tc
), 0)
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
319 self
.assertTrue(sc
.supports_discarded_events
)
320 self
.assertFalse(sc
.discarded_events_have_default_clock_snapshots
)
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,
328 self
.assertTrue(sc
.supports_discarded_events
)
329 self
.assertTrue(sc
.discarded_events_have_default_clock_snapshots
)
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
337 self
.assertEqual(len(self
._tc
), 0)
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,
346 self
.assertEqual(len(self
._tc
), 0)
348 def test_does_not_support_discarded_events_raises_with_cs(self
):
349 with self
.assertRaisesRegex(
351 "cannot not support discarded events, but have default clock snapshots for discarded event messages",
353 self
._tc
.create_stream_class(
354 default_clock_class
=self
._cc
,
355 discarded_events_have_default_clock_snapshots
=True,
358 self
.assertEqual(len(self
._tc
), 0)
360 def test_supports_discarded_events_with_clock_snapshots_without_default_clock_class_raises(
363 with self
.assertRaisesRegex(
365 'cannot have no default clock class, but have default clock snapshots for discarded event messages',
367 self
._tc
.create_stream_class(
368 supports_discarded_events
=True,
369 discarded_events_have_default_clock_snapshots
=True,
372 self
.assertEqual(len(self
._tc
), 0)
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,
380 self
.assertTrue(sc
.supports_discarded_packets
)
381 self
.assertFalse(sc
.discarded_packets_have_default_clock_snapshots
)
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,
390 self
.assertTrue(sc
.supports_discarded_packets
)
391 self
.assertTrue(sc
.discarded_packets_have_default_clock_snapshots
)
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"
397 self
._tc
.create_stream_class(
398 default_clock_class
=self
._cc
, supports_discarded_packets
=True
401 self
.assertEqual(len(self
._tc
), 0)
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,
411 self
.assertEqual(len(self
._tc
), 0)
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,
421 self
.assertEqual(len(self
._tc
), 0)
423 def test_does_not_support_discarded_packets_raises_with_cs(self
):
424 with self
.assertRaisesRegex(
426 "cannot not support discarded packets, but have default clock snapshots for discarded packet messages",
428 self
._tc
.create_stream_class(
429 default_clock_class
=self
._cc
,
430 discarded_packets_have_default_clock_snapshots
=True,
431 supports_packets
=True,
434 self
.assertEqual(len(self
._tc
), 0)
436 def test_supports_discarded_packets_with_clock_snapshots_without_default_clock_class_raises(
439 with self
.assertRaisesRegex(
441 'cannot have no default clock class, but have default clock snapshots for discarded packet messages',
443 self
._tc
.create_stream_class(
444 supports_packets
=True,
445 supports_discarded_packets
=True,
446 discarded_packets_have_default_clock_snapshots
=True,
449 self
.assertEqual(len(self
._tc
), 0)
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
)
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)
462 def test_getitem(self
):
463 sc
, ec1
, ec2
= self
._create
_stream
_class
_with
_event
_classes
()
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
)
470 def test_getitem_wrong_key_type(self
):
471 sc
, _
, _
= self
._create
_stream
_class
_with
_event
_classes
()
473 with self
.assertRaisesRegex(TypeError, "'str' is not an 'int' object"):
476 def test_getitem_wrong_key(self
):
477 sc
, _
, _
= self
._create
_stream
_class
_with
_event
_classes
()
479 with self
.assertRaisesRegex(KeyError, '19'):
483 sc
, _
, _
= self
._create
_stream
_class
_with
_event
_classes
()
485 self
.assertEqual(len(sc
), 2)
488 sc
, _
, _
= self
._create
_stream
_class
_with
_event
_classes
()
491 self
.assertEqual(ec_ids
, [17, 23])
494 if __name__
== '__main__':