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
.assertRaises(TypeError):
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
.assertRaises(TypeError):
77 self
._tc
.create_stream_class(packet_context_field_class
=22)
79 def test_create_invalid_packet_context_field_class_no_packets(self
):
80 fc
= self
._tc
.create_structure_field_class()
82 with self
.assertRaises(ValueError):
83 self
._tc
.create_stream_class(packet_context_field_class
=fc
)
85 def test_create_event_common_context_field_class(self
):
86 fc
= self
._tc
.create_structure_field_class()
87 sc
= self
._tc
.create_stream_class(event_common_context_field_class
=fc
)
88 self
.assertEqual(sc
.event_common_context_field_class
, fc
)
90 type(sc
.event_common_context_field_class
),
91 bt2_field_class
._StructureFieldClass
,
94 def test_create_invalid_event_common_context_field_class(self
):
95 with self
.assertRaises(TypeError):
96 self
._tc
.create_stream_class(event_common_context_field_class
=22)
98 def test_create_default_clock_class(self
):
99 sc
= self
._tc
.create_stream_class(default_clock_class
=self
._cc
)
100 self
.assertEqual(sc
.default_clock_class
.addr
, self
._cc
.addr
)
101 self
.assertIs(type(sc
.default_clock_class
), bt2_clock_class
._ClockClass
)
103 def test_create_invalid_default_clock_class(self
):
104 with self
.assertRaises(TypeError):
105 self
._tc
.create_stream_class(default_clock_class
=12)
107 def test_create_user_attributes(self
):
108 sc
= self
._tc
.create_stream_class(user_attributes
={'salut': 23})
109 self
.assertEqual(sc
.user_attributes
, {'salut': 23})
111 def test_create_invalid_user_attributes(self
):
112 with self
.assertRaises(TypeError):
113 self
._tc
.create_stream_class(user_attributes
=object())
115 def test_create_invalid_user_attributes_value_type(self
):
116 with self
.assertRaises(TypeError):
117 self
._tc
.create_stream_class(user_attributes
=23)
119 def test_automatic_stream_ids(self
):
120 sc
= self
._tc
.create_stream_class(assigns_automatic_stream_id
=True)
121 self
.assertTrue(sc
.assigns_automatic_stream_id
)
123 stream
= self
._trace
.create_stream(sc
)
124 self
.assertIsNotNone(stream
.id)
126 def test_automatic_stream_ids_raises(self
):
127 sc
= self
._tc
.create_stream_class(assigns_automatic_stream_id
=True)
128 self
.assertTrue(sc
.assigns_automatic_stream_id
)
130 with self
.assertRaises(ValueError):
131 self
._trace
.create_stream(sc
, id=123)
133 def test_no_automatic_stream_ids(self
):
134 sc
= self
._tc
.create_stream_class(assigns_automatic_stream_id
=False)
135 self
.assertFalse(sc
.assigns_automatic_stream_id
)
137 stream
= self
._trace
.create_stream(sc
, id=333)
138 self
.assertEqual(stream
.id, 333)
140 def test_no_automatic_stream_ids_raises(self
):
141 sc
= self
._tc
.create_stream_class(assigns_automatic_stream_id
=False)
142 self
.assertFalse(sc
.assigns_automatic_stream_id
)
144 with self
.assertRaises(ValueError):
145 self
._trace
.create_stream(sc
)
147 def test_automatic_event_class_ids(self
):
148 sc
= self
._tc
.create_stream_class(assigns_automatic_event_class_id
=True)
149 self
.assertTrue(sc
.assigns_automatic_event_class_id
)
151 ec
= sc
.create_event_class()
152 self
.assertIsNotNone(ec
.id)
154 def test_automatic_event_class_ids_raises(self
):
155 sc
= self
._tc
.create_stream_class(assigns_automatic_event_class_id
=True)
156 self
.assertTrue(sc
.assigns_automatic_event_class_id
)
158 with self
.assertRaises(ValueError):
159 sc
.create_event_class(id=123)
161 def test_no_automatic_event_class_ids(self
):
162 sc
= self
._tc
.create_stream_class(assigns_automatic_event_class_id
=False)
163 self
.assertFalse(sc
.assigns_automatic_event_class_id
)
165 ec
= sc
.create_event_class(id=333)
166 self
.assertEqual(ec
.id, 333)
168 def test_no_automatic_event_class_ids_raises(self
):
169 sc
= self
._tc
.create_stream_class(assigns_automatic_event_class_id
=False)
170 self
.assertFalse(sc
.assigns_automatic_event_class_id
)
172 with self
.assertRaises(ValueError):
173 sc
.create_event_class()
175 def test_supports_packets_without_cs(self
):
176 sc
= self
._tc
.create_stream_class(
177 default_clock_class
=self
._cc
, supports_packets
=True
179 self
.assertTrue(sc
.supports_packets
)
180 self
.assertFalse(sc
.packets_have_beginning_default_clock_snapshot
)
181 self
.assertFalse(sc
.packets_have_end_default_clock_snapshot
)
183 def test_supports_packets_with_begin_cs(self
):
184 sc
= self
._tc
.create_stream_class(
185 default_clock_class
=self
._cc
,
186 supports_packets
=True,
187 packets_have_beginning_default_clock_snapshot
=True,
189 self
.assertTrue(sc
.supports_packets
)
190 self
.assertTrue(sc
.packets_have_beginning_default_clock_snapshot
)
191 self
.assertFalse(sc
.packets_have_end_default_clock_snapshot
)
193 def test_supports_packets_with_end_cs(self
):
194 sc
= self
._tc
.create_stream_class(
195 default_clock_class
=self
._cc
,
196 supports_packets
=True,
197 packets_have_end_default_clock_snapshot
=True,
199 self
.assertTrue(sc
.supports_packets
)
200 self
.assertFalse(sc
.packets_have_beginning_default_clock_snapshot
)
201 self
.assertTrue(sc
.packets_have_end_default_clock_snapshot
)
203 def test_supports_packets_raises_type_error(self
):
204 with self
.assertRaises(TypeError):
205 self
._tc
.create_stream_class(
206 default_clock_class
=self
._cc
, supports_packets
=23
209 def test_packets_have_begin_default_cs_raises_type_error(self
):
210 with self
.assertRaises(TypeError):
211 self
._tc
.create_stream_class(
212 default_clock_class
=self
._cc
,
213 packets_have_beginning_default_clock_snapshot
=23,
216 def test_packets_have_end_default_cs_raises_type_error(self
):
217 with self
.assertRaises(TypeError):
218 self
._tc
.create_stream_class(
219 default_clock_class
=self
._cc
, packets_have_end_default_clock_snapshot
=23
222 def test_does_not_support_packets_raises_with_begin_cs(self
):
223 with self
.assertRaises(ValueError):
224 self
._tc
.create_stream_class(
225 default_clock_class
=self
._cc
,
226 packets_have_beginning_default_clock_snapshot
=True,
229 def test_does_not_support_packets_raises_with_end_cs(self
):
230 with self
.assertRaises(ValueError):
231 self
._tc
.create_stream_class(
232 default_clock_class
=self
._cc
,
233 packets_have_end_default_clock_snapshot
=True,
236 def test_supports_discarded_events_without_cs(self
):
237 sc
= self
._tc
.create_stream_class(
238 default_clock_class
=self
._cc
, supports_discarded_events
=True
240 self
.assertTrue(sc
.supports_discarded_events
)
241 self
.assertFalse(sc
.discarded_events_have_default_clock_snapshots
)
243 def test_supports_discarded_events_with_cs(self
):
244 sc
= self
._tc
.create_stream_class(
245 default_clock_class
=self
._cc
,
246 supports_discarded_events
=True,
247 discarded_events_have_default_clock_snapshots
=True,
249 self
.assertTrue(sc
.supports_discarded_events
)
250 self
.assertTrue(sc
.discarded_events_have_default_clock_snapshots
)
252 def test_supports_discarded_events_raises_type_error(self
):
253 with self
.assertRaises(TypeError):
254 self
._tc
.create_stream_class(
255 default_clock_class
=self
._cc
, supports_discarded_events
=23
258 def test_discarded_events_have_default_cs_raises_type_error(self
):
259 with self
.assertRaises(TypeError):
260 self
._tc
.create_stream_class(
261 default_clock_class
=self
._cc
,
262 discarded_events_have_default_clock_snapshots
=23,
265 def test_does_not_support_discarded_events_raises_with_cs(self
):
266 with self
.assertRaises(ValueError):
267 self
._tc
.create_stream_class(
268 default_clock_class
=self
._cc
,
269 discarded_events_have_default_clock_snapshots
=True,
272 def test_supports_discarded_packets_without_cs(self
):
273 sc
= self
._tc
.create_stream_class(
274 default_clock_class
=self
._cc
,
275 supports_discarded_packets
=True,
276 supports_packets
=True,
278 self
.assertTrue(sc
.supports_discarded_packets
)
279 self
.assertFalse(sc
.discarded_packets_have_default_clock_snapshots
)
281 def test_supports_discarded_packets_with_cs(self
):
282 sc
= self
._tc
.create_stream_class(
283 default_clock_class
=self
._cc
,
284 supports_discarded_packets
=True,
285 discarded_packets_have_default_clock_snapshots
=True,
286 supports_packets
=True,
288 self
.assertTrue(sc
.supports_discarded_packets
)
289 self
.assertTrue(sc
.discarded_packets_have_default_clock_snapshots
)
291 def test_supports_discarded_packets_raises_without_packet_support(self
):
292 with self
.assertRaises(ValueError):
293 self
._tc
.create_stream_class(
294 default_clock_class
=self
._cc
, supports_discarded_packets
=True
297 def test_supports_discarded_packets_raises_type_error(self
):
298 with self
.assertRaises(TypeError):
299 self
._tc
.create_stream_class(
300 default_clock_class
=self
._cc
,
301 supports_discarded_packets
=23,
302 supports_packets
=True,
305 def test_discarded_packets_have_default_cs_raises_type_error(self
):
306 with self
.assertRaises(TypeError):
307 self
._tc
.create_stream_class(
308 default_clock_class
=self
._cc
,
309 discarded_packets_have_default_clock_snapshots
=23,
310 supports_packets
=True,
313 def test_does_not_support_discarded_packets_raises_with_cs(self
):
314 with self
.assertRaises(ValueError):
315 self
._tc
.create_stream_class(
316 default_clock_class
=self
._cc
,
317 discarded_packets_have_default_clock_snapshots
=True,
318 supports_packets
=True,
321 def test_trace_class(self
):
322 sc
= self
._tc
.create_stream_class()
323 self
.assertEqual(sc
.trace_class
.addr
, self
._tc
.addr
)
324 self
.assertIs(type(sc
.trace_class
), bt2_trace_class
._TraceClass
)
326 def _create_stream_class_with_event_classes(self
):
327 sc
= self
._tc
.create_stream_class(assigns_automatic_event_class_id
=False)
328 ec1
= sc
.create_event_class(id=23)
329 ec2
= sc
.create_event_class(id=17)
332 def test_getitem(self
):
333 sc
, ec1
, ec2
= self
._create
_stream
_class
_with
_event
_classes
()
335 self
.assertEqual(sc
[23].addr
, ec1
.addr
)
336 self
.assertEqual(type(sc
[23]), bt2_event_class
._EventClass
)
337 self
.assertEqual(sc
[17].addr
, ec2
.addr
)
338 self
.assertEqual(type(sc
[17]), bt2_event_class
._EventClass
)
340 def test_getitem_wrong_key_type(self
):
341 sc
, _
, _
= self
._create
_stream
_class
_with
_event
_classes
()
343 with self
.assertRaises(TypeError):
346 def test_getitem_wrong_key(self
):
347 sc
, _
, _
= self
._create
_stream
_class
_with
_event
_classes
()
349 with self
.assertRaises(KeyError):
353 sc
, _
, _
= self
._create
_stream
_class
_with
_event
_classes
()
355 self
.assertEqual(len(sc
), 2)
358 sc
, _
, _
= self
._create
_stream
_class
_with
_event
_classes
()
361 self
.assertEqual(ec_ids
, [17, 23])
364 if __name__
== '__main__':