fix: test_message_iterator.py hangs on Python 3.12
[babeltrace.git] / tests / bindings / python / bt2 / test_message.py
CommitLineData
32d2d479
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
f6a5e476 19import unittest
f6a5e476 20import bt2
9cbe0c59 21import utils
fac7b25a 22from utils import TestOutputPortMessageIterator
9cbe0c59
FD
23from bt2 import clock_snapshot as bt2_clock_snapshot
24from bt2 import event as bt2_event
25from bt2 import event_class as bt2_event_class
26from bt2 import field as bt2_field
27from bt2 import packet as bt2_packet
28from bt2 import stream as bt2_stream
29from bt2 import stream_class as bt2_stream_class
30from bt2 import trace as bt2_trace
31from bt2 import trace_class as bt2_trace_class
f6a5e476
PP
32
33
0010c8b0 34class AllMessagesTestCase(unittest.TestCase):
f6a5e476 35 def setUp(self):
fa4c33e3 36 class MyIter(bt2._UserMessageIterator):
9415de1c 37 def __init__(self, config, self_port_output):
0010c8b0 38 self._at = 0
61d96b89
FD
39 self._with_stream_msgs_clock_snapshots = self_port_output.user_data.get(
40 'with_stream_msgs_clock_snapshots', False
41 )
f6a5e476
PP
42
43 def __next__(self):
0010c8b0
SM
44 if test_obj._clock_class:
45 if self._at == 0:
b7cbc799 46 if self._with_stream_msgs_clock_snapshots:
61d96b89
FD
47 msg = self._create_stream_beginning_message(
48 test_obj._stream, default_clock_snapshot=self._at
49 )
b7cbc799 50 else:
61d96b89
FD
51 msg = self._create_stream_beginning_message(
52 test_obj._stream
53 )
9cbe0c59 54 test_obj.assertIs(type(msg), bt2._StreamBeginningMessage)
0010c8b0 55 elif self._at == 1:
61d96b89
FD
56 msg = self._create_packet_beginning_message(
57 test_obj._packet, self._at
58 )
9cbe0c59 59 test_obj.assertIs(type(msg), bt2._PacketBeginningMessage)
b7cbc799 60 elif self._at == 2:
61d96b89
FD
61 msg = self._create_event_message(
62 test_obj._event_class, test_obj._packet, self._at
63 )
9cbe0c59 64 test_obj.assertIs(type(msg), bt2._EventMessage)
b7cbc799 65 elif self._at == 3:
61d96b89
FD
66 msg = self._create_message_iterator_inactivity_message(
67 test_obj._clock_class, self._at
68 )
b7cbc799 69 elif self._at == 4:
61d96b89
FD
70 msg = self._create_discarded_events_message(
71 test_obj._stream, 890, self._at, self._at
72 )
9cbe0c59 73 test_obj.assertIs(type(msg), bt2._DiscardedEventsMessage)
b7cbc799 74 elif self._at == 5:
61d96b89
FD
75 msg = self._create_packet_end_message(
76 test_obj._packet, self._at
77 )
9cbe0c59 78 test_obj.assertIs(type(msg), bt2._PacketEndMessage)
b7cbc799 79 elif self._at == 6:
61d96b89
FD
80 msg = self._create_discarded_packets_message(
81 test_obj._stream, 678, self._at, self._at
82 )
9cbe0c59 83 test_obj.assertIs(type(msg), bt2._DiscardedPacketsMessage)
b7cbc799
SM
84 elif self._at == 7:
85 if self._with_stream_msgs_clock_snapshots:
61d96b89
FD
86 msg = self._create_stream_end_message(
87 test_obj._stream, default_clock_snapshot=self._at
88 )
b7cbc799
SM
89 else:
90 msg = self._create_stream_end_message(test_obj._stream)
9cbe0c59 91 test_obj.assertIs(type(msg), bt2._StreamEndMessage)
b7cbc799 92 elif self._at >= 8:
0010c8b0 93 raise bt2.Stop
f6a5e476 94 else:
0010c8b0
SM
95 if self._at == 0:
96 msg = self._create_stream_beginning_message(test_obj._stream)
97 elif self._at == 1:
0010c8b0 98 msg = self._create_packet_beginning_message(test_obj._packet)
b7cbc799 99 elif self._at == 2:
61d96b89
FD
100 msg = self._create_event_message(
101 test_obj._event_class, test_obj._packet
102 )
b7cbc799 103 elif self._at == 3:
61d96b89
FD
104 msg = self._create_discarded_events_message(
105 test_obj._stream, 890
106 )
b7cbc799 107 elif self._at == 4:
0010c8b0 108 msg = self._create_packet_end_message(test_obj._packet)
b7cbc799 109 elif self._at == 5:
61d96b89
FD
110 msg = self._create_discarded_packets_message(
111 test_obj._stream, 678
112 )
b7cbc799 113 elif self._at == 6:
0010c8b0 114 msg = self._create_stream_end_message(test_obj._stream)
b7cbc799 115 elif self._at >= 7:
0010c8b0 116 raise bt2.Stop
f6a5e476
PP
117
118 self._at += 1
fa4c33e3 119 return msg
f6a5e476 120
0010c8b0 121 class MySrc(bt2._UserSourceComponent, message_iterator_class=MyIter):
e3250e61 122 def __init__(self, config, params, obj):
b7cbc799 123 self._add_output_port('out', params)
f6a5e476 124
77037b2b 125 with_cc = bool(params['with_cc'])
0010c8b0
SM
126 tc = self._create_trace_class()
127 if with_cc:
128 cc = self._create_clock_class()
f6a5e476 129 else:
0010c8b0 130 cc = None
0010c8b0 131
61d96b89
FD
132 sc = tc.create_stream_class(
133 default_clock_class=cc,
134 supports_packets=True,
135 packets_have_beginning_default_clock_snapshot=with_cc,
136 packets_have_end_default_clock_snapshot=with_cc,
137 supports_discarded_events=True,
138 discarded_events_have_default_clock_snapshots=with_cc,
139 supports_discarded_packets=True,
140 discarded_packets_have_default_clock_snapshots=with_cc,
141 )
0010c8b0
SM
142
143 # Create payload field class
144 my_int_fc = tc.create_signed_integer_field_class(32)
145 payload_fc = tc.create_structure_field_class()
61d96b89 146 payload_fc += [('my_int', my_int_fc)]
0010c8b0 147
9cbe0c59
FD
148 # Create specific context field class
149 my_int_fc = tc.create_signed_integer_field_class(32)
150 specific_fc = tc.create_structure_field_class()
151 specific_fc += [('my_int', my_int_fc)]
152
153 ec = sc.create_event_class(
154 name='salut',
155 payload_field_class=payload_fc,
156 specific_context_field_class=specific_fc,
157 )
0010c8b0
SM
158
159 trace = tc()
160 stream = trace.create_stream(sc)
161 packet = stream.create_packet()
162
163 test_obj._trace = trace
164 test_obj._stream = stream
165 test_obj._packet = packet
166 test_obj._event_class = ec
167 test_obj._clock_class = cc
168
169 test_obj = self
170 self._graph = bt2.Graph()
171 self._src = MySrc
172 self._iter = MyIter
173
174 def test_all_msg_with_cc(self):
175 params = {'with_cc': True}
176 self._src_comp = self._graph.add_component(self._src, 'my_source', params)
fac7b25a
PP
177 self._msg_iter = TestOutputPortMessageIterator(
178 self._graph, self._src_comp.output_ports['out']
61d96b89 179 )
0010c8b0
SM
180
181 for i, msg in enumerate(self._msg_iter):
182 if i == 0:
9cbe0c59
FD
183 self.assertIs(type(msg), bt2._StreamBeginningMessageConst)
184 self.assertIs(type(msg.stream), bt2_stream._StreamConst)
0010c8b0 185 self.assertEqual(msg.stream.addr, self._stream.addr)
61d96b89 186 self.assertIsInstance(
c946c9de 187 msg.default_clock_snapshot, bt2._UnknownClockSnapshot
61d96b89 188 )
0010c8b0 189 elif i == 1:
9cbe0c59
FD
190 self.assertIs(type(msg), bt2._PacketBeginningMessageConst)
191 self.assertIs(type(msg.packet), bt2_packet._PacketConst)
192 self.assertIs(
193 type(msg.default_clock_snapshot),
194 bt2_clock_snapshot._ClockSnapshotConst,
195 )
0010c8b0
SM
196 self.assertEqual(msg.packet.addr, self._packet.addr)
197 self.assertEqual(msg.default_clock_snapshot.value, i)
b7cbc799 198 elif i == 2:
9cbe0c59
FD
199 self.assertIs(type(msg), bt2._EventMessageConst)
200 self.assertIs(type(msg.event), bt2_event._EventConst)
201 self.assertIs(
202 type(msg.default_clock_snapshot),
203 bt2_clock_snapshot._ClockSnapshotConst,
204 )
205 self.assertIs(
206 type(msg.event.payload_field), bt2_field._StructureFieldConst
207 )
208 self.assertIs(
209 type(msg.event.payload_field['my_int']),
210 bt2_field._SignedIntegerFieldConst,
211 )
212
c88be1c8 213 self.assertEqual(msg.event.cls.addr, self._event_class.addr)
0010c8b0 214 self.assertEqual(msg.default_clock_snapshot.value, i)
b7cbc799 215 elif i == 3:
9cbe0c59
FD
216 self.assertIs(type(msg), bt2._MessageIteratorInactivityMessageConst)
217 self.assertIs(
62988c56 218 type(msg.clock_snapshot), bt2_clock_snapshot._ClockSnapshotConst
9cbe0c59 219 )
62988c56 220 self.assertEqual(msg.clock_snapshot.value, i)
b7cbc799 221 elif i == 4:
9cbe0c59
FD
222 self.assertIs(type(msg), bt2._DiscardedEventsMessageConst)
223 self.assertIs(type(msg.stream), bt2_stream._StreamConst)
224 self.assertIs(type(msg.stream.cls), bt2_stream_class._StreamClassConst)
225 self.assertIs(
226 type(msg.beginning_default_clock_snapshot),
227 bt2_clock_snapshot._ClockSnapshotConst,
228 )
229 self.assertIs(
230 type(msg.end_default_clock_snapshot),
231 bt2_clock_snapshot._ClockSnapshotConst,
232 )
233
0010c8b0
SM
234 self.assertEqual(msg.stream.addr, self._stream.addr)
235 self.assertEqual(msg.count, 890)
61d96b89
FD
236 self.assertEqual(
237 msg.stream.cls.default_clock_class.addr, self._clock_class.addr
238 )
0010c8b0
SM
239 self.assertEqual(msg.beginning_default_clock_snapshot.value, i)
240 self.assertEqual(msg.end_default_clock_snapshot.value, i)
b7cbc799 241 elif i == 5:
9cbe0c59
FD
242 self.assertIs(type(msg), bt2._PacketEndMessageConst)
243 self.assertIs(type(msg.packet), bt2_packet._PacketConst)
244 self.assertIs(
245 type(msg.default_clock_snapshot),
246 bt2_clock_snapshot._ClockSnapshotConst,
247 )
0010c8b0
SM
248 self.assertEqual(msg.packet.addr, self._packet.addr)
249 self.assertEqual(msg.default_clock_snapshot.value, i)
b7cbc799 250 elif i == 6:
9cbe0c59
FD
251 self.assertIs(type(msg), bt2._DiscardedPacketsMessageConst)
252 self.assertIs(type(msg.stream), bt2_stream._StreamConst)
253 self.assertIs(type(msg.stream.trace), bt2_trace._TraceConst)
254 self.assertIs(
255 type(msg.stream.trace.cls), bt2_trace_class._TraceClassConst
256 )
257 self.assertIs(
258 type(msg.beginning_default_clock_snapshot),
259 bt2_clock_snapshot._ClockSnapshotConst,
260 )
261 self.assertIs(
262 type(msg.end_default_clock_snapshot),
263 bt2_clock_snapshot._ClockSnapshotConst,
264 )
0010c8b0
SM
265 self.assertEqual(msg.stream.addr, self._stream.addr)
266 self.assertEqual(msg.count, 678)
61d96b89
FD
267 self.assertEqual(
268 msg.stream.cls.default_clock_class.addr, self._clock_class.addr
269 )
0010c8b0
SM
270 self.assertEqual(msg.beginning_default_clock_snapshot.value, i)
271 self.assertEqual(msg.end_default_clock_snapshot.value, i)
b7cbc799 272 elif i == 7:
9cbe0c59
FD
273 self.assertIs(type(msg), bt2._StreamEndMessageConst)
274 self.assertIs(type(msg.stream), bt2_stream._StreamConst)
0010c8b0 275 self.assertEqual(msg.stream.addr, self._stream.addr)
9cbe0c59
FD
276 self.assertIs(
277 type(msg.default_clock_snapshot), bt2._UnknownClockSnapshot
61d96b89 278 )
0010c8b0
SM
279 else:
280 raise Exception
281
282 def test_all_msg_without_cc(self):
283 params = {'with_cc': False}
284 self._src_comp = self._graph.add_component(self._src, 'my_source', params)
fac7b25a
PP
285 self._msg_iter = TestOutputPortMessageIterator(
286 self._graph, self._src_comp.output_ports['out']
61d96b89 287 )
0010c8b0
SM
288
289 for i, msg in enumerate(self._msg_iter):
290 if i == 0:
9cbe0c59
FD
291 self.assertIsInstance(msg, bt2._StreamBeginningMessageConst)
292 self.assertIs(type(msg.stream), bt2_stream._StreamConst)
0010c8b0 293 self.assertEqual(msg.stream.addr, self._stream.addr)
8b18e5b9
SM
294 with self.assertRaisesRegex(
295 ValueError, 'stream class has no default clock class'
296 ):
b7cbc799 297 msg.default_clock_snapshot
0010c8b0 298 elif i == 1:
9cbe0c59
FD
299 self.assertIsInstance(msg, bt2._PacketBeginningMessageConst)
300 self.assertIs(type(msg.packet), bt2_packet._PacketConst)
0010c8b0 301 self.assertEqual(msg.packet.addr, self._packet.addr)
b7cbc799 302 elif i == 2:
9cbe0c59
FD
303 self.assertIsInstance(msg, bt2._EventMessageConst)
304 self.assertIs(type(msg.event), bt2_event._EventConst)
305 self.assertIs(type(msg.event.cls), bt2_event_class._EventClassConst)
c88be1c8 306 self.assertEqual(msg.event.cls.addr, self._event_class.addr)
8b18e5b9
SM
307 with self.assertRaisesRegex(
308 ValueError, 'stream class has no default clock class'
309 ):
0010c8b0 310 msg.default_clock_snapshot
b7cbc799 311 elif i == 3:
9cbe0c59
FD
312 self.assertIsInstance(msg, bt2._DiscardedEventsMessageConst)
313 self.assertIs(type(msg.stream), bt2_stream._StreamConst)
314 self.assertIs(type(msg.stream.cls), bt2_stream_class._StreamClassConst)
0010c8b0
SM
315 self.assertEqual(msg.stream.addr, self._stream.addr)
316 self.assertEqual(msg.count, 890)
c88be1c8 317 self.assertIsNone(msg.stream.cls.default_clock_class)
8b18e5b9
SM
318 with self.assertRaisesRegex(
319 ValueError,
320 'such a message has no clock snapshots for this stream class',
321 ):
0010c8b0 322 msg.beginning_default_clock_snapshot
8b18e5b9
SM
323 with self.assertRaisesRegex(
324 ValueError,
325 'such a message has no clock snapshots for this stream class',
326 ):
0010c8b0 327 msg.end_default_clock_snapshot
b7cbc799 328 elif i == 4:
9cbe0c59 329 self.assertIsInstance(msg, bt2._PacketEndMessageConst)
0010c8b0 330 self.assertEqual(msg.packet.addr, self._packet.addr)
9cbe0c59 331 self.assertIs(type(msg.packet), bt2_packet._PacketConst)
b7cbc799 332 elif i == 5:
9cbe0c59
FD
333 self.assertIsInstance(msg, bt2._DiscardedPacketsMessageConst)
334 self.assertIs(type(msg.stream), bt2_stream._StreamConst)
335 self.assertIs(type(msg.stream.cls), bt2_stream_class._StreamClassConst)
336 self.assertIs(
337 type(msg.stream.cls.trace_class), bt2_trace_class._TraceClassConst
338 )
0010c8b0
SM
339 self.assertEqual(msg.stream.addr, self._stream.addr)
340 self.assertEqual(msg.count, 678)
c88be1c8 341 self.assertIsNone(msg.stream.cls.default_clock_class)
8b18e5b9
SM
342 with self.assertRaisesRegex(
343 ValueError,
344 'such a message has no clock snapshots for this stream class',
345 ):
0010c8b0 346 msg.beginning_default_clock_snapshot
8b18e5b9
SM
347 with self.assertRaisesRegex(
348 ValueError,
349 'such a message has no clock snapshots for this stream class',
350 ):
0010c8b0 351 msg.end_default_clock_snapshot
b7cbc799 352 elif i == 6:
9cbe0c59
FD
353 self.assertIsInstance(msg, bt2._StreamEndMessageConst)
354 self.assertIs(type(msg.stream), bt2_stream._StreamConst)
0010c8b0 355 self.assertEqual(msg.stream.addr, self._stream.addr)
8b18e5b9
SM
356 with self.assertRaisesRegex(
357 ValueError, 'stream class has no default clock class'
358 ):
b7cbc799 359 msg.default_clock_snapshot
0010c8b0
SM
360 else:
361 raise Exception
f6a5e476 362
b7cbc799 363 def test_msg_stream_with_clock_snapshots(self):
61d96b89 364 params = {'with_cc': True, 'with_stream_msgs_clock_snapshots': True}
dcd94213 365
b7cbc799 366 self._src_comp = self._graph.add_component(self._src, 'my_source', params)
fac7b25a
PP
367 self._msg_iter = TestOutputPortMessageIterator(
368 self._graph, self._src_comp.output_ports['out']
61d96b89 369 )
b7cbc799 370 msgs = list(self._msg_iter)
dcd94213 371
b7cbc799 372 msg_stream_beg = msgs[0]
9cbe0c59
FD
373 self.assertIsInstance(msg_stream_beg, bt2._StreamBeginningMessageConst)
374 self.assertIs(
375 type(msg_stream_beg.default_clock_snapshot),
376 bt2_clock_snapshot._ClockSnapshotConst,
377 )
b7cbc799 378 self.assertEqual(msg_stream_beg.default_clock_snapshot.value, 0)
dcd94213 379
b7cbc799 380 msg_stream_end = msgs[7]
9cbe0c59
FD
381 self.assertIsInstance(msg_stream_end, bt2._StreamEndMessageConst)
382 self.assertIs(
383 type(msg_stream_end.default_clock_snapshot),
384 bt2_clock_snapshot._ClockSnapshotConst,
385 )
b7cbc799 386 self.assertEqual(msg_stream_end.default_clock_snapshot.value, 7)
9cbe0c59
FD
387
388 def test_stream_beg_msg(self):
389 msg = utils.get_stream_beginning_message()
390 self.assertIs(type(msg.stream), bt2_stream._Stream)
391
392 def test_stream_end_msg(self):
393 msg = utils.get_stream_end_message()
394 self.assertIs(type(msg.stream), bt2_stream._Stream)
395
396 def test_packet_beg_msg(self):
397 msg = utils.get_packet_beginning_message()
398 self.assertIs(type(msg.packet), bt2_packet._Packet)
399
400 def test_packet_end_msg(self):
401 msg = utils.get_packet_end_message()
402 self.assertIs(type(msg.packet), bt2_packet._Packet)
403
404 def test_event_msg(self):
405 msg = utils.get_event_message()
406 self.assertIs(type(msg.event), bt2_event._Event)
3db06b1d
SM
407
408
356ada2b
SM
409class CreateDiscardedEventMessageTestCase(unittest.TestCase):
410 # Most basic case.
411 def test_create(self):
412 def create_stream_class(tc, cc):
413 return tc.create_stream_class(supports_discarded_events=True)
414
415 def msg_iter_next(msg_iter, stream):
416 return msg_iter._create_discarded_events_message(stream)
417
418 msg = utils.run_in_message_iterator_next(create_stream_class, msg_iter_next)
419 self.assertIs(type(msg), bt2._DiscardedEventsMessage)
31592d12 420 self.assertIs(msg.count, None)
356ada2b
SM
421
422 # With event count.
423 def test_create_with_count(self):
424 def create_stream_class(tc, cc):
425 return tc.create_stream_class(supports_discarded_events=True)
426
427 def msg_iter_next(msg_iter, stream):
428 return msg_iter._create_discarded_events_message(stream, count=242)
429
430 msg = utils.run_in_message_iterator_next(create_stream_class, msg_iter_next)
431 self.assertIs(type(msg), bt2._DiscardedEventsMessage)
31592d12 432 self.assertEqual(msg.count, 242)
356ada2b
SM
433
434 # With clock snapshots.
435 def test_create_with_clock_snapshots(self):
436 def create_stream_class(tc, cc):
437 return tc.create_stream_class(
438 default_clock_class=cc,
439 supports_discarded_events=True,
440 discarded_events_have_default_clock_snapshots=True,
441 )
442
443 def msg_iter_next(msg_iter, stream):
444 return msg_iter._create_discarded_events_message(
445 stream, beg_clock_snapshot=10, end_clock_snapshot=20
446 )
447
448 msg = utils.run_in_message_iterator_next(create_stream_class, msg_iter_next)
449 self.assertIs(type(msg), bt2._DiscardedEventsMessage)
31592d12
SM
450 self.assertEqual(msg.beginning_default_clock_snapshot, 10)
451 self.assertEqual(msg.end_default_clock_snapshot, 20)
356ada2b
SM
452
453 # Trying to create when the stream does not support discarded events.
454 def test_create_unsupported_raises(self):
455 def create_stream_class(tc, cc):
456 return tc.create_stream_class()
457
458 def msg_iter_next(msg_iter, stream):
459 with self.assertRaisesRegex(
460 ValueError, 'stream class does not support discarded events'
461 ):
462 msg_iter._create_discarded_events_message(stream)
463
464 return 123
465
466 res = utils.run_in_message_iterator_next(create_stream_class, msg_iter_next)
467 self.assertEqual(res, 123)
468
469 # Trying to create with clock snapshots when the stream does not support
470 # them.
471 def test_create_unsupported_clock_snapshots_raises(self):
472 def create_stream_class(tc, cc):
473 return tc.create_stream_class(supports_discarded_events=True)
474
475 def msg_iter_next(msg_iter, stream):
476 with self.assertRaisesRegex(
477 ValueError,
478 'discarded events have no default clock snapshots for this stream class',
479 ):
480 msg_iter._create_discarded_events_message(
481 stream, beg_clock_snapshot=10, end_clock_snapshot=20
482 )
483
484 return 123
485
486 res = utils.run_in_message_iterator_next(create_stream_class, msg_iter_next)
487 self.assertEqual(res, 123)
488
489 # Trying to create without clock snapshots when the stream requires them.
490 def test_create_missing_clock_snapshots_raises(self):
491 def create_stream_class(tc, cc):
492 return tc.create_stream_class(
493 default_clock_class=cc,
494 supports_discarded_events=True,
495 discarded_events_have_default_clock_snapshots=True,
496 )
497
498 def msg_iter_next(msg_iter, stream):
499 with self.assertRaisesRegex(
500 ValueError,
501 'discarded events have default clock snapshots for this stream class',
502 ):
503 msg_iter._create_discarded_events_message(stream)
504
505 return 123
506
507 res = utils.run_in_message_iterator_next(create_stream_class, msg_iter_next)
508 self.assertEqual(res, 123)
509
510
511class CreateDiscardedPacketMessageTestCase(unittest.TestCase):
512 # Most basic case.
513 def test_create(self):
514 def create_stream_class(tc, cc):
515 return tc.create_stream_class(
516 supports_packets=True, supports_discarded_packets=True
517 )
518
519 def msg_iter_next(msg_iter, stream):
520 return msg_iter._create_discarded_packets_message(stream)
521
522 msg = utils.run_in_message_iterator_next(create_stream_class, msg_iter_next)
523 self.assertIs(type(msg), bt2._DiscardedPacketsMessage)
524 self.assertIs(msg.count, None)
525
526 # With packet count.
527 def test_create_with_count(self):
528 def create_stream_class(tc, cc):
529 return tc.create_stream_class(
530 supports_packets=True, supports_discarded_packets=True
531 )
532
533 def msg_iter_next(msg_iter, stream):
534 return msg_iter._create_discarded_packets_message(stream, count=242)
535
536 msg = utils.run_in_message_iterator_next(create_stream_class, msg_iter_next)
537 self.assertIs(type(msg), bt2._DiscardedPacketsMessage)
538 self.assertEqual(msg.count, 242)
539
540 # With clock snapshots.
541 def test_create_with_clock_snapshots(self):
542 def create_stream_class(tc, cc):
543 return tc.create_stream_class(
544 default_clock_class=cc,
545 supports_packets=True,
546 supports_discarded_packets=True,
547 discarded_packets_have_default_clock_snapshots=True,
548 )
549
550 def msg_iter_next(msg_iter, stream):
551 return msg_iter._create_discarded_packets_message(
552 stream, beg_clock_snapshot=10, end_clock_snapshot=20
553 )
554
555 msg = utils.run_in_message_iterator_next(create_stream_class, msg_iter_next)
556 self.assertIs(type(msg), bt2._DiscardedPacketsMessage)
557 self.assertEqual(msg.beginning_default_clock_snapshot, 10)
558 self.assertEqual(msg.end_default_clock_snapshot, 20)
559
560 # Trying to create when the stream does not support discarded packets.
561 def test_create_unsupported_raises(self):
562 def create_stream_class(tc, cc):
e30908cf
PP
563 return tc.create_stream_class(
564 supports_packets=True,
565 )
356ada2b
SM
566
567 def msg_iter_next(msg_iter, stream):
568 with self.assertRaisesRegex(
569 ValueError, 'stream class does not support discarded packets'
570 ):
571 msg_iter._create_discarded_packets_message(stream)
572
573 return 123
574
575 res = utils.run_in_message_iterator_next(create_stream_class, msg_iter_next)
576 self.assertEqual(res, 123)
577
578 # Trying to create with clock snapshots when the stream does not support
579 # them.
580 def test_create_unsupported_clock_snapshots_raises(self):
581 def create_stream_class(tc, cc):
582 return tc.create_stream_class(
583 supports_packets=True, supports_discarded_packets=True
584 )
585
586 def msg_iter_next(msg_iter, stream):
587 with self.assertRaisesRegex(
588 ValueError,
589 'discarded packets have no default clock snapshots for this stream class',
590 ):
591 msg_iter._create_discarded_packets_message(
592 stream, beg_clock_snapshot=10, end_clock_snapshot=20
593 )
594
595 return 123
596
597 res = utils.run_in_message_iterator_next(create_stream_class, msg_iter_next)
598 self.assertEqual(res, 123)
599
600 # Trying to create without clock snapshots when the stream requires them.
601 def test_create_missing_clock_snapshots_raises(self):
602 def create_stream_class(tc, cc):
603 return tc.create_stream_class(
604 default_clock_class=cc,
605 supports_packets=True,
606 supports_discarded_packets=True,
607 discarded_packets_have_default_clock_snapshots=True,
608 )
609
610 def msg_iter_next(msg_iter, stream):
611 with self.assertRaisesRegex(
612 ValueError,
613 'discarded packets have default clock snapshots for this stream class',
614 ):
615 msg_iter._create_discarded_packets_message(stream)
616
617 return 123
618
619 res = utils.run_in_message_iterator_next(create_stream_class, msg_iter_next)
620 self.assertEqual(res, 123)
621
622
3db06b1d
SM
623if __name__ == '__main__':
624 unittest.main()
This page took 0.082301 seconds and 4 git commands to generate.