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 def test_create_packet_context_field_class(self
):
66 fc
= self
._tc
.create_structure_field_class()
67 sc
= self
._tc
.create_stream_class(
68 packet_context_field_class
=fc
, supports_packets
=True
70 self
.assertEqual(sc
.packet_context_field_class
, fc
)
72 type(sc
.packet_context_field_class
), bt2_field_class
._StructureFieldClass
75 def test_create_invalid_packet_context_field_class(self
):
76 with self
.assertRaisesRegex(
78 "'int' is not a '<class 'bt2.field_class._StructureFieldClass'>' object",
80 self
._tc
.create_stream_class(packet_context_field_class
=22)
82 def test_create_invalid_packet_context_field_class_no_packets(self
):
83 fc
= self
._tc
.create_structure_field_class()
85 with self
.assertRaisesRegex(
87 "cannot have a packet context field class without supporting packets",
89 self
._tc
.create_stream_class(packet_context_field_class
=fc
)
91 def test_create_event_common_context_field_class(self
):
92 fc
= self
._tc
.create_structure_field_class()
93 sc
= self
._tc
.create_stream_class(event_common_context_field_class
=fc
)
94 self
.assertEqual(sc
.event_common_context_field_class
, fc
)
96 type(sc
.event_common_context_field_class
),
97 bt2_field_class
._StructureFieldClass
,
100 def test_create_invalid_event_common_context_field_class(self
):
101 with self
.assertRaisesRegex(
103 "'int' is not a '<class 'bt2.field_class._StructureFieldClass'>' object",
105 self
._tc
.create_stream_class(event_common_context_field_class
=22)
107 def test_create_default_clock_class(self
):
108 sc
= self
._tc
.create_stream_class(default_clock_class
=self
._cc
)
109 self
.assertEqual(sc
.default_clock_class
.addr
, self
._cc
.addr
)
110 self
.assertIs(type(sc
.default_clock_class
), bt2_clock_class
._ClockClass
)
112 def test_create_invalid_default_clock_class(self
):
113 with self
.assertRaisesRegex(
114 TypeError, "'int' is not a '<class 'bt2.clock_class._ClockClass'>' object"
116 self
._tc
.create_stream_class(default_clock_class
=12)
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 def test_create_invalid_user_attributes_value_type(self
):
129 with self
.assertRaisesRegex(
131 "'SignedIntegerValue' is not a '<class 'bt2.value.MapValue'>' object",
133 self
._tc
.create_stream_class(user_attributes
=23)
135 def test_automatic_stream_ids(self
):
136 sc
= self
._tc
.create_stream_class(assigns_automatic_stream_id
=True)
137 self
.assertTrue(sc
.assigns_automatic_stream_id
)
139 stream
= self
._trace
.create_stream(sc
)
140 self
.assertIsNotNone(stream
.id)
142 def test_automatic_stream_ids_raises(self
):
143 sc
= self
._tc
.create_stream_class(assigns_automatic_stream_id
=True)
144 self
.assertTrue(sc
.assigns_automatic_stream_id
)
146 with self
.assertRaisesRegex(
147 ValueError, "id provided, but stream class assigns automatic stream ids"
149 self
._trace
.create_stream(sc
, id=123)
151 def test_no_automatic_stream_ids(self
):
152 sc
= self
._tc
.create_stream_class(assigns_automatic_stream_id
=False)
153 self
.assertFalse(sc
.assigns_automatic_stream_id
)
155 stream
= self
._trace
.create_stream(sc
, id=333)
156 self
.assertEqual(stream
.id, 333)
158 def test_no_automatic_stream_ids_raises(self
):
159 sc
= self
._tc
.create_stream_class(assigns_automatic_stream_id
=False)
160 self
.assertFalse(sc
.assigns_automatic_stream_id
)
162 with self
.assertRaisesRegex(
164 "id not provided, but stream class does not assign automatic stream ids",
166 self
._trace
.create_stream(sc
)
168 def test_automatic_event_class_ids(self
):
169 sc
= self
._tc
.create_stream_class(assigns_automatic_event_class_id
=True)
170 self
.assertTrue(sc
.assigns_automatic_event_class_id
)
172 ec
= sc
.create_event_class()
173 self
.assertIsNotNone(ec
.id)
175 def test_automatic_event_class_ids_raises(self
):
176 sc
= self
._tc
.create_stream_class(assigns_automatic_event_class_id
=True)
177 self
.assertTrue(sc
.assigns_automatic_event_class_id
)
179 with self
.assertRaisesRegex(
181 "id provided, but stream class assigns automatic event class ids",
183 sc
.create_event_class(id=123)
185 def test_no_automatic_event_class_ids(self
):
186 sc
= self
._tc
.create_stream_class(assigns_automatic_event_class_id
=False)
187 self
.assertFalse(sc
.assigns_automatic_event_class_id
)
189 ec
= sc
.create_event_class(id=333)
190 self
.assertEqual(ec
.id, 333)
192 def test_no_automatic_event_class_ids_raises(self
):
193 sc
= self
._tc
.create_stream_class(assigns_automatic_event_class_id
=False)
194 self
.assertFalse(sc
.assigns_automatic_event_class_id
)
196 with self
.assertRaisesRegex(
198 "id not provided, but stream class does not assign automatic event class ids",
200 sc
.create_event_class()
202 def test_supports_packets_without_cs(self
):
203 sc
= self
._tc
.create_stream_class(
204 default_clock_class
=self
._cc
, supports_packets
=True
206 self
.assertTrue(sc
.supports_packets
)
207 self
.assertFalse(sc
.packets_have_beginning_default_clock_snapshot
)
208 self
.assertFalse(sc
.packets_have_end_default_clock_snapshot
)
210 def test_supports_packets_with_begin_cs(self
):
211 sc
= self
._tc
.create_stream_class(
212 default_clock_class
=self
._cc
,
213 supports_packets
=True,
214 packets_have_beginning_default_clock_snapshot
=True,
216 self
.assertTrue(sc
.supports_packets
)
217 self
.assertTrue(sc
.packets_have_beginning_default_clock_snapshot
)
218 self
.assertFalse(sc
.packets_have_end_default_clock_snapshot
)
220 def test_supports_packets_with_end_cs(self
):
221 sc
= self
._tc
.create_stream_class(
222 default_clock_class
=self
._cc
,
223 supports_packets
=True,
224 packets_have_end_default_clock_snapshot
=True,
226 self
.assertTrue(sc
.supports_packets
)
227 self
.assertFalse(sc
.packets_have_beginning_default_clock_snapshot
)
228 self
.assertTrue(sc
.packets_have_end_default_clock_snapshot
)
230 def test_supports_packets_raises_type_error(self
):
231 with self
.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
232 self
._tc
.create_stream_class(
233 default_clock_class
=self
._cc
, supports_packets
=23
236 def test_packets_have_begin_default_cs_raises_type_error(self
):
237 with self
.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
238 self
._tc
.create_stream_class(
239 default_clock_class
=self
._cc
,
240 packets_have_beginning_default_clock_snapshot
=23,
243 def test_packets_have_end_default_cs_raises_type_error(self
):
244 with self
.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
245 self
._tc
.create_stream_class(
246 default_clock_class
=self
._cc
, packets_have_end_default_clock_snapshot
=23
249 def test_does_not_support_packets_raises_with_begin_cs(self
):
250 with self
.assertRaisesRegex(
252 "cannot not support packets, but have packet beginning default clock snapshot",
254 self
._tc
.create_stream_class(
255 default_clock_class
=self
._cc
,
256 packets_have_beginning_default_clock_snapshot
=True,
259 def test_does_not_support_packets_raises_with_end_cs(self
):
260 with self
.assertRaisesRegex(
262 "cannot not support packets, but have packet end default clock snapshots",
264 self
._tc
.create_stream_class(
265 default_clock_class
=self
._cc
,
266 packets_have_end_default_clock_snapshot
=True,
269 def test_supports_discarded_events_without_cs(self
):
270 sc
= self
._tc
.create_stream_class(
271 default_clock_class
=self
._cc
, supports_discarded_events
=True
273 self
.assertTrue(sc
.supports_discarded_events
)
274 self
.assertFalse(sc
.discarded_events_have_default_clock_snapshots
)
276 def test_supports_discarded_events_with_cs(self
):
277 sc
= self
._tc
.create_stream_class(
278 default_clock_class
=self
._cc
,
279 supports_discarded_events
=True,
280 discarded_events_have_default_clock_snapshots
=True,
282 self
.assertTrue(sc
.supports_discarded_events
)
283 self
.assertTrue(sc
.discarded_events_have_default_clock_snapshots
)
285 def test_supports_discarded_events_raises_type_error(self
):
286 with self
.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
287 self
._tc
.create_stream_class(
288 default_clock_class
=self
._cc
, supports_discarded_events
=23
291 def test_discarded_events_have_default_cs_raises_type_error(self
):
292 with self
.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
293 self
._tc
.create_stream_class(
294 default_clock_class
=self
._cc
,
295 discarded_events_have_default_clock_snapshots
=23,
298 def test_does_not_support_discarded_events_raises_with_cs(self
):
299 with self
.assertRaisesRegex(
301 "cannot not support discarded events, but have default clock snapshots for discarded event messages",
303 self
._tc
.create_stream_class(
304 default_clock_class
=self
._cc
,
305 discarded_events_have_default_clock_snapshots
=True,
308 def test_supports_discarded_packets_without_cs(self
):
309 sc
= self
._tc
.create_stream_class(
310 default_clock_class
=self
._cc
,
311 supports_discarded_packets
=True,
312 supports_packets
=True,
314 self
.assertTrue(sc
.supports_discarded_packets
)
315 self
.assertFalse(sc
.discarded_packets_have_default_clock_snapshots
)
317 def test_supports_discarded_packets_with_cs(self
):
318 sc
= self
._tc
.create_stream_class(
319 default_clock_class
=self
._cc
,
320 supports_discarded_packets
=True,
321 discarded_packets_have_default_clock_snapshots
=True,
322 supports_packets
=True,
324 self
.assertTrue(sc
.supports_discarded_packets
)
325 self
.assertTrue(sc
.discarded_packets_have_default_clock_snapshots
)
327 def test_supports_discarded_packets_raises_without_packet_support(self
):
328 with self
.assertRaisesRegex(
329 ValueError, "cannot support discarded packets, but not support packets"
331 self
._tc
.create_stream_class(
332 default_clock_class
=self
._cc
, supports_discarded_packets
=True
335 def test_supports_discarded_packets_raises_type_error(self
):
336 with self
.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
337 self
._tc
.create_stream_class(
338 default_clock_class
=self
._cc
,
339 supports_discarded_packets
=23,
340 supports_packets
=True,
343 def test_discarded_packets_have_default_cs_raises_type_error(self
):
344 with self
.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
345 self
._tc
.create_stream_class(
346 default_clock_class
=self
._cc
,
347 discarded_packets_have_default_clock_snapshots
=23,
348 supports_packets
=True,
351 def test_does_not_support_discarded_packets_raises_with_cs(self
):
352 with self
.assertRaisesRegex(
354 "cannot not support discarded packets, but have default clock snapshots for discarded packet messages",
356 self
._tc
.create_stream_class(
357 default_clock_class
=self
._cc
,
358 discarded_packets_have_default_clock_snapshots
=True,
359 supports_packets
=True,
362 def test_trace_class(self
):
363 sc
= self
._tc
.create_stream_class()
364 self
.assertEqual(sc
.trace_class
.addr
, self
._tc
.addr
)
365 self
.assertIs(type(sc
.trace_class
), bt2_trace_class
._TraceClass
)
367 def _create_stream_class_with_event_classes(self
):
368 sc
= self
._tc
.create_stream_class(assigns_automatic_event_class_id
=False)
369 ec1
= sc
.create_event_class(id=23)
370 ec2
= sc
.create_event_class(id=17)
373 def test_getitem(self
):
374 sc
, ec1
, ec2
= self
._create
_stream
_class
_with
_event
_classes
()
376 self
.assertEqual(sc
[23].addr
, ec1
.addr
)
377 self
.assertEqual(type(sc
[23]), bt2_event_class
._EventClass
)
378 self
.assertEqual(sc
[17].addr
, ec2
.addr
)
379 self
.assertEqual(type(sc
[17]), bt2_event_class
._EventClass
)
381 def test_getitem_wrong_key_type(self
):
382 sc
, _
, _
= self
._create
_stream
_class
_with
_event
_classes
()
384 with self
.assertRaisesRegex(TypeError, "'str' is not an 'int' object"):
387 def test_getitem_wrong_key(self
):
388 sc
, _
, _
= self
._create
_stream
_class
_with
_event
_classes
()
390 with self
.assertRaisesRegex(KeyError, '19'):
394 sc
, _
, _
= self
._create
_stream
_class
_with
_event
_classes
()
396 self
.assertEqual(len(sc
), 2)
399 sc
, _
, _
= self
._create
_stream
_class
_with
_event
_classes
()
402 self
.assertEqual(ec_ids
, [17, 23])
405 if __name__
== '__main__':