lib: make can_seek_ns_from_origin logic use `can_seek_forward` property of iterator
[babeltrace.git] / tests / bindings / python / bt2 / test_stream_class.py
CommitLineData
d2d857a8
MJ
1#
2# Copyright (C) 2019 EfficiOS Inc.
3#
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
7# of the License.
8#
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.
13#
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.
17#
18
9cf643d1 19import unittest
3cdfbaea 20from utils import run_in_component_init
f0a42b33
FD
21from bt2 import stream_class as bt2_stream_class
22from bt2 import trace_class as bt2_trace_class
23from bt2 import clock_class as bt2_clock_class
24from bt2 import event_class as bt2_event_class
25from bt2 import field_class as bt2_field_class
9cf643d1
PP
26
27
28class StreamClassTestCase(unittest.TestCase):
29 def setUp(self):
3cdfbaea
SM
30 def f(comp_self):
31 tc = comp_self._create_trace_class(assigns_automatic_stream_class_id=True)
32 cc = comp_self._create_clock_class()
33 return tc, cc
34
35 self._tc, self._cc = run_in_component_init(f)
36 self._trace = self._tc()
37
38 def test_create_default(self):
39 sc = self._tc.create_stream_class()
40
f0a42b33 41 self.assertIs(type(sc), bt2_stream_class._StreamClass)
3cdfbaea
SM
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)
26fc5aed 48 self.assertFalse(sc.supports_packets)
9b24b6aa
PP
49 self.assertFalse(sc.packets_have_beginning_default_clock_snapshot)
50 self.assertFalse(sc.packets_have_end_default_clock_snapshot)
2e90378a
PP
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)
5783664e 55 self.assertEqual(len(sc.user_attributes), 0)
3cdfbaea
SM
56
57 def test_create_name(self):
58 sc = self._tc.create_stream_class(name='bozo')
59 self.assertEqual(sc.name, 'bozo')
60
61 def test_create_invalid_name(self):
9cf643d1 62 with self.assertRaises(TypeError):
3cdfbaea 63 self._tc.create_stream_class(name=17)
9cf643d1 64
3cdfbaea
SM
65 def test_create_packet_context_field_class(self):
66 fc = self._tc.create_structure_field_class()
cfbd7cf3
FD
67 sc = self._tc.create_stream_class(
68 packet_context_field_class=fc, supports_packets=True
69 )
3cdfbaea 70 self.assertEqual(sc.packet_context_field_class, fc)
f0a42b33
FD
71 self.assertIs(
72 type(sc.packet_context_field_class), bt2_field_class._StructureFieldClass
73 )
9cf643d1 74
3cdfbaea 75 def test_create_invalid_packet_context_field_class(self):
9cf643d1 76 with self.assertRaises(TypeError):
3cdfbaea 77 self._tc.create_stream_class(packet_context_field_class=22)
9cf643d1 78
26fc5aed
PP
79 def test_create_invalid_packet_context_field_class_no_packets(self):
80 fc = self._tc.create_structure_field_class()
81
82 with self.assertRaises(ValueError):
83 self._tc.create_stream_class(packet_context_field_class=fc)
84
3cdfbaea
SM
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)
f0a42b33
FD
89 self.assertIs(
90 type(sc.event_common_context_field_class),
91 bt2_field_class._StructureFieldClass,
92 )
811644b8 93
3cdfbaea
SM
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)
9cf643d1 97
3cdfbaea
SM
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)
f0a42b33 101 self.assertIs(type(sc.default_clock_class), bt2_clock_class._ClockClass)
9cf643d1 102
3cdfbaea 103 def test_create_invalid_default_clock_class(self):
9cf643d1 104 with self.assertRaises(TypeError):
3cdfbaea 105 self._tc.create_stream_class(default_clock_class=12)
9cf643d1 106
5783664e
PP
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})
110
111 def test_create_invalid_user_attributes(self):
112 with self.assertRaises(TypeError):
113 self._tc.create_stream_class(user_attributes=object())
114
115 def test_create_invalid_user_attributes_value_type(self):
116 with self.assertRaises(TypeError):
117 self._tc.create_stream_class(user_attributes=23)
118
3cdfbaea
SM
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)
9cf643d1 122
3cdfbaea
SM
123 stream = self._trace.create_stream(sc)
124 self.assertIsNotNone(stream.id)
9cf643d1 125
3cdfbaea
SM
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)
129
4430bc80 130 with self.assertRaises(ValueError):
3cdfbaea
SM
131 self._trace.create_stream(sc, id=123)
132
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)
136
137 stream = self._trace.create_stream(sc, id=333)
138 self.assertEqual(stream.id, 333)
139
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)
143
4430bc80 144 with self.assertRaises(ValueError):
3cdfbaea
SM
145 self._trace.create_stream(sc)
146
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)
150
151 ec = sc.create_event_class()
152 self.assertIsNotNone(ec.id)
153
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)
157
4430bc80 158 with self.assertRaises(ValueError):
3cdfbaea
SM
159 sc.create_event_class(id=123)
160
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)
9cf643d1 164
3cdfbaea
SM
165 ec = sc.create_event_class(id=333)
166 self.assertEqual(ec.id, 333)
9cf643d1 167
3cdfbaea
SM
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)
9cf643d1 171
4430bc80 172 with self.assertRaises(ValueError):
3cdfbaea
SM
173 sc.create_event_class()
174
26fc5aed 175 def test_supports_packets_without_cs(self):
cfbd7cf3
FD
176 sc = self._tc.create_stream_class(
177 default_clock_class=self._cc, supports_packets=True
178 )
26fc5aed
PP
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)
182
183 def test_supports_packets_with_begin_cs(self):
cfbd7cf3
FD
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,
188 )
26fc5aed 189 self.assertTrue(sc.supports_packets)
9b24b6aa 190 self.assertTrue(sc.packets_have_beginning_default_clock_snapshot)
26fc5aed 191 self.assertFalse(sc.packets_have_end_default_clock_snapshot)
3cdfbaea 192
26fc5aed 193 def test_supports_packets_with_end_cs(self):
cfbd7cf3
FD
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,
198 )
26fc5aed
PP
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)
202
203 def test_supports_packets_raises_type_error(self):
3cdfbaea 204 with self.assertRaises(TypeError):
082db648 205 self._tc.create_stream_class(
cfbd7cf3
FD
206 default_clock_class=self._cc, supports_packets=23
207 )
3cdfbaea 208
26fc5aed
PP
209 def test_packets_have_begin_default_cs_raises_type_error(self):
210 with self.assertRaises(TypeError):
082db648 211 self._tc.create_stream_class(
cfbd7cf3
FD
212 default_clock_class=self._cc,
213 packets_have_beginning_default_clock_snapshot=23,
214 )
3cdfbaea 215
26fc5aed 216 def test_packets_have_end_default_cs_raises_type_error(self):
9cf643d1 217 with self.assertRaises(TypeError):
082db648 218 self._tc.create_stream_class(
cfbd7cf3
FD
219 default_clock_class=self._cc, packets_have_end_default_clock_snapshot=23
220 )
26fc5aed
PP
221
222 def test_does_not_support_packets_raises_with_begin_cs(self):
223 with self.assertRaises(ValueError):
082db648 224 self._tc.create_stream_class(
cfbd7cf3
FD
225 default_clock_class=self._cc,
226 packets_have_beginning_default_clock_snapshot=True,
227 )
26fc5aed
PP
228
229 def test_does_not_support_packets_raises_with_end_cs(self):
230 with self.assertRaises(ValueError):
082db648 231 self._tc.create_stream_class(
cfbd7cf3
FD
232 default_clock_class=self._cc,
233 packets_have_end_default_clock_snapshot=True,
234 )
3cdfbaea 235
2e90378a 236 def test_supports_discarded_events_without_cs(self):
cfbd7cf3
FD
237 sc = self._tc.create_stream_class(
238 default_clock_class=self._cc, supports_discarded_events=True
239 )
2e90378a
PP
240 self.assertTrue(sc.supports_discarded_events)
241 self.assertFalse(sc.discarded_events_have_default_clock_snapshots)
242
243 def test_supports_discarded_events_with_cs(self):
cfbd7cf3
FD
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,
248 )
2e90378a
PP
249 self.assertTrue(sc.supports_discarded_events)
250 self.assertTrue(sc.discarded_events_have_default_clock_snapshots)
251
252 def test_supports_discarded_events_raises_type_error(self):
253 with self.assertRaises(TypeError):
082db648 254 self._tc.create_stream_class(
cfbd7cf3
FD
255 default_clock_class=self._cc, supports_discarded_events=23
256 )
2e90378a
PP
257
258 def test_discarded_events_have_default_cs_raises_type_error(self):
259 with self.assertRaises(TypeError):
082db648 260 self._tc.create_stream_class(
cfbd7cf3
FD
261 default_clock_class=self._cc,
262 discarded_events_have_default_clock_snapshots=23,
263 )
2e90378a
PP
264
265 def test_does_not_support_discarded_events_raises_with_cs(self):
266 with self.assertRaises(ValueError):
082db648 267 self._tc.create_stream_class(
cfbd7cf3
FD
268 default_clock_class=self._cc,
269 discarded_events_have_default_clock_snapshots=True,
270 )
2e90378a
PP
271
272 def test_supports_discarded_packets_without_cs(self):
cfbd7cf3
FD
273 sc = self._tc.create_stream_class(
274 default_clock_class=self._cc,
275 supports_discarded_packets=True,
276 supports_packets=True,
277 )
2e90378a
PP
278 self.assertTrue(sc.supports_discarded_packets)
279 self.assertFalse(sc.discarded_packets_have_default_clock_snapshots)
280
281 def test_supports_discarded_packets_with_cs(self):
cfbd7cf3
FD
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,
287 )
2e90378a
PP
288 self.assertTrue(sc.supports_discarded_packets)
289 self.assertTrue(sc.discarded_packets_have_default_clock_snapshots)
290
26fc5aed
PP
291 def test_supports_discarded_packets_raises_without_packet_support(self):
292 with self.assertRaises(ValueError):
082db648 293 self._tc.create_stream_class(
cfbd7cf3
FD
294 default_clock_class=self._cc, supports_discarded_packets=True
295 )
26fc5aed 296
2e90378a
PP
297 def test_supports_discarded_packets_raises_type_error(self):
298 with self.assertRaises(TypeError):
082db648 299 self._tc.create_stream_class(
cfbd7cf3
FD
300 default_clock_class=self._cc,
301 supports_discarded_packets=23,
302 supports_packets=True,
303 )
2e90378a
PP
304
305 def test_discarded_packets_have_default_cs_raises_type_error(self):
306 with self.assertRaises(TypeError):
082db648 307 self._tc.create_stream_class(
cfbd7cf3
FD
308 default_clock_class=self._cc,
309 discarded_packets_have_default_clock_snapshots=23,
310 supports_packets=True,
311 )
2e90378a
PP
312
313 def test_does_not_support_discarded_packets_raises_with_cs(self):
314 with self.assertRaises(ValueError):
082db648 315 self._tc.create_stream_class(
cfbd7cf3
FD
316 default_clock_class=self._cc,
317 discarded_packets_have_default_clock_snapshots=True,
318 supports_packets=True,
319 )
2e90378a 320
3cdfbaea
SM
321 def test_trace_class(self):
322 sc = self._tc.create_stream_class()
323 self.assertEqual(sc.trace_class.addr, self._tc.addr)
f0a42b33 324 self.assertIs(type(sc.trace_class), bt2_trace_class._TraceClass)
3cdfbaea
SM
325
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)
330 return sc, ec1, ec2
9cf643d1
PP
331
332 def test_getitem(self):
3cdfbaea
SM
333 sc, ec1, ec2 = self._create_stream_class_with_event_classes()
334
335 self.assertEqual(sc[23].addr, ec1.addr)
f0a42b33 336 self.assertEqual(type(sc[23]), bt2_event_class._EventClass)
3cdfbaea 337 self.assertEqual(sc[17].addr, ec2.addr)
f0a42b33 338 self.assertEqual(type(sc[17]), bt2_event_class._EventClass)
9cf643d1
PP
339
340 def test_getitem_wrong_key_type(self):
3cdfbaea
SM
341 sc, _, _ = self._create_stream_class_with_event_classes()
342
9cf643d1 343 with self.assertRaises(TypeError):
3cdfbaea 344 sc['event23']
9cf643d1
PP
345
346 def test_getitem_wrong_key(self):
3cdfbaea
SM
347 sc, _, _ = self._create_stream_class_with_event_classes()
348
9cf643d1 349 with self.assertRaises(KeyError):
3cdfbaea 350 sc[19]
9cf643d1
PP
351
352 def test_len(self):
3cdfbaea
SM
353 sc, _, _ = self._create_stream_class_with_event_classes()
354
355 self.assertEqual(len(sc), 2)
9cf643d1
PP
356
357 def test_iter(self):
3cdfbaea
SM
358 sc, _, _ = self._create_stream_class_with_event_classes()
359
360 ec_ids = sorted(sc)
361 self.assertEqual(ec_ids, [17, 23])
d14ddbba
SM
362
363
364if __name__ == '__main__':
365 unittest.main()
This page took 0.065704 seconds and 4 git commands to generate.