tests/bindings/python/bt2/test_message.py

   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
  19 import unittest
  20 import bt2
  21 import utils
  22 from utils import TestOutputPortMessageIterator
  23 from bt2 import clock_snapshot as bt2_clock_snapshot
  24 from bt2 import event as bt2_event
  25 from bt2 import event_class as bt2_event_class
  26 from bt2 import field as bt2_field
  27 from bt2 import packet as bt2_packet
  28 from bt2 import stream as bt2_stream
  29 from bt2 import stream_class as bt2_stream_class
  30 from bt2 import trace as bt2_trace
  31 from bt2 import trace_class as bt2_trace_class
  32
  33
  34 class AllMessagesTestCase(unittest.TestCase):
  35     def setUp(self):
  36         class MyIter(bt2._UserMessageIterator):
  37             def __init__(self, config, self_port_output):
  38                 self._at = 0
  39                 self._with_stream_msgs_clock_snapshots = self_port_output.user_data.get(
  40                     'with_stream_msgs_clock_snapshots', False
  41                 )
  42
  43             def __next__(self):
  44                 if test_obj._clock_class:
  45                     if self._at == 0:
  46                         if self._with_stream_msgs_clock_snapshots:
  47                             msg = self._create_stream_beginning_message(
  48                                 test_obj._stream, default_clock_snapshot=self._at
  49                             )
  50                         else:
  51                             msg = self._create_stream_beginning_message(
  52                                 test_obj._stream
  53                             )
  54                         test_obj.assertIs(type(msg), bt2._StreamBeginningMessage)
  55                     elif self._at == 1:
  56                         msg = self._create_packet_beginning_message(
  57                             test_obj._packet, self._at
  58                         )
  59                         test_obj.assertIs(type(msg), bt2._PacketBeginningMessage)
  60                     elif self._at == 2:
  61                         msg = self._create_event_message(
  62                             test_obj._event_class, test_obj._packet, self._at
  63                         )
  64                         test_obj.assertIs(type(msg), bt2._EventMessage)
  65                     elif self._at == 3:
  66                         msg = self._create_message_iterator_inactivity_message(
  67                             test_obj._clock_class, self._at
  68                         )
  69                     elif self._at == 4:
  70                         msg = self._create_discarded_events_message(
  71                             test_obj._stream, 890, self._at, self._at
  72                         )
  73                         test_obj.assertIs(type(msg), bt2._DiscardedEventsMessage)
  74                     elif self._at == 5:
  75                         msg = self._create_packet_end_message(
  76                             test_obj._packet, self._at
  77                         )
  78                         test_obj.assertIs(type(msg), bt2._PacketEndMessage)
  79                     elif self._at == 6:
  80                         msg = self._create_discarded_packets_message(
  81                             test_obj._stream, 678, self._at, self._at
  82                         )
  83                         test_obj.assertIs(type(msg), bt2._DiscardedPacketsMessage)
  84                     elif self._at == 7:
  85                         if self._with_stream_msgs_clock_snapshots:
  86                             msg = self._create_stream_end_message(
  87                                 test_obj._stream, default_clock_snapshot=self._at
  88                             )
  89                         else:
  90                             msg = self._create_stream_end_message(test_obj._stream)
  91                         test_obj.assertIs(type(msg), bt2._StreamEndMessage)
  92                     elif self._at >= 8:
  93                         raise bt2.Stop
  94                 else:
  95                     if self._at == 0:
  96                         msg = self._create_stream_beginning_message(test_obj._stream)
  97                     elif self._at == 1:
  98                         msg = self._create_packet_beginning_message(test_obj._packet)
  99                     elif self._at == 2:
 100                         msg = self._create_event_message(
 101                             test_obj._event_class, test_obj._packet
 102                         )
 103                     elif self._at == 3:
 104                         msg = self._create_discarded_events_message(
 105                             test_obj._stream, 890
 106                         )
 107                     elif self._at == 4:
 108                         msg = self._create_packet_end_message(test_obj._packet)
 109                     elif self._at == 5:
 110                         msg = self._create_discarded_packets_message(
 111                             test_obj._stream, 678
 112                         )
 113                     elif self._at == 6:
 114                         msg = self._create_stream_end_message(test_obj._stream)
 115                     elif self._at >= 7:
 116                         raise bt2.Stop
 117
 118                 self._at += 1
 119                 return msg
 120
 121         class MySrc(bt2._UserSourceComponent, message_iterator_class=MyIter):
 122             def __init__(self, config, params, obj):
 123                 self._add_output_port('out', params)
 124
 125                 with_cc = bool(params['with_cc'])
 126                 tc = self._create_trace_class()
 127                 if with_cc:
 128                     cc = self._create_clock_class()
 129                 else:
 130                     cc = None
 131
 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                 )
 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()
 146                 payload_fc += [('my_int', my_int_fc)]
 147
 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                 )
 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)
 177         self._msg_iter = TestOutputPortMessageIterator(
 178             self._graph, self._src_comp.output_ports['out']
 179         )
 180
 181         for i, msg in enumerate(self._msg_iter):
 182             if i == 0:
 183                 self.assertIs(type(msg), bt2._StreamBeginningMessageConst)
 184                 self.assertIs(type(msg.stream), bt2_stream._StreamConst)
 185                 self.assertEqual(msg.stream.addr, self._stream.addr)
 186                 self.assertIsInstance(
 187                     msg.default_clock_snapshot, bt2._UnknownClockSnapshot
 188                 )
 189             elif i == 1:
 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                 )
 196                 self.assertEqual(msg.packet.addr, self._packet.addr)
 197                 self.assertEqual(msg.default_clock_snapshot.value, i)
 198             elif i == 2:
 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
 213                 self.assertEqual(msg.event.cls.addr, self._event_class.addr)
 214                 self.assertEqual(msg.default_clock_snapshot.value, i)
 215             elif i == 3:
 216                 self.assertIs(type(msg), bt2._MessageIteratorInactivityMessageConst)
 217                 self.assertIs(
 218                     type(msg.clock_snapshot), bt2_clock_snapshot._ClockSnapshotConst
 219                 )
 220                 self.assertEqual(msg.clock_snapshot.value, i)
 221             elif i == 4:
 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
 234                 self.assertEqual(msg.stream.addr, self._stream.addr)
 235                 self.assertEqual(msg.count, 890)
 236                 self.assertEqual(
 237                     msg.stream.cls.default_clock_class.addr, self._clock_class.addr
 238                 )
 239                 self.assertEqual(msg.beginning_default_clock_snapshot.value, i)
 240                 self.assertEqual(msg.end_default_clock_snapshot.value, i)
 241             elif i == 5:
 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                 )
 248                 self.assertEqual(msg.packet.addr, self._packet.addr)
 249                 self.assertEqual(msg.default_clock_snapshot.value, i)
 250             elif i == 6:
 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                 )
 265                 self.assertEqual(msg.stream.addr, self._stream.addr)
 266                 self.assertEqual(msg.count, 678)
 267                 self.assertEqual(
 268                     msg.stream.cls.default_clock_class.addr, self._clock_class.addr
 269                 )
 270                 self.assertEqual(msg.beginning_default_clock_snapshot.value, i)
 271                 self.assertEqual(msg.end_default_clock_snapshot.value, i)
 272             elif i == 7:
 273                 self.assertIs(type(msg), bt2._StreamEndMessageConst)
 274                 self.assertIs(type(msg.stream), bt2_stream._StreamConst)
 275                 self.assertEqual(msg.stream.addr, self._stream.addr)
 276                 self.assertIs(
 277                     type(msg.default_clock_snapshot), bt2._UnknownClockSnapshot
 278                 )
 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)
 285         self._msg_iter = TestOutputPortMessageIterator(
 286             self._graph, self._src_comp.output_ports['out']
 287         )
 288
 289         for i, msg in enumerate(self._msg_iter):
 290             if i == 0:
 291                 self.assertIsInstance(msg, bt2._StreamBeginningMessageConst)
 292                 self.assertIs(type(msg.stream), bt2_stream._StreamConst)
 293                 self.assertEqual(msg.stream.addr, self._stream.addr)
 294                 with self.assertRaisesRegex(
 295                     ValueError, 'stream class has no default clock class'
 296                 ):
 297                     msg.default_clock_snapshot
 298             elif i == 1:
 299                 self.assertIsInstance(msg, bt2._PacketBeginningMessageConst)
 300                 self.assertIs(type(msg.packet), bt2_packet._PacketConst)
 301                 self.assertEqual(msg.packet.addr, self._packet.addr)
 302             elif i == 2:
 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)
 306                 self.assertEqual(msg.event.cls.addr, self._event_class.addr)
 307                 with self.assertRaisesRegex(
 308                     ValueError, 'stream class has no default clock class'
 309                 ):
 310                     msg.default_clock_snapshot
 311             elif i == 3:
 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)
 315                 self.assertEqual(msg.stream.addr, self._stream.addr)
 316                 self.assertEqual(msg.count, 890)
 317                 self.assertIsNone(msg.stream.cls.default_clock_class)
 318                 with self.assertRaisesRegex(
 319                     ValueError,
 320                     'such a message has no clock snapshots for this stream class',
 321                 ):
 322                     msg.beginning_default_clock_snapshot
 323                 with self.assertRaisesRegex(
 324                     ValueError,
 325                     'such a message has no clock snapshots for this stream class',
 326                 ):
 327                     msg.end_default_clock_snapshot
 328             elif i == 4:
 329                 self.assertIsInstance(msg, bt2._PacketEndMessageConst)
 330                 self.assertEqual(msg.packet.addr, self._packet.addr)
 331                 self.assertIs(type(msg.packet), bt2_packet._PacketConst)
 332             elif i == 5:
 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                 )
 339                 self.assertEqual(msg.stream.addr, self._stream.addr)
 340                 self.assertEqual(msg.count, 678)
 341                 self.assertIsNone(msg.stream.cls.default_clock_class)
 342                 with self.assertRaisesRegex(
 343                     ValueError,
 344                     'such a message has no clock snapshots for this stream class',
 345                 ):
 346                     msg.beginning_default_clock_snapshot
 347                 with self.assertRaisesRegex(
 348                     ValueError,
 349                     'such a message has no clock snapshots for this stream class',
 350                 ):
 351                     msg.end_default_clock_snapshot
 352             elif i == 6:
 353                 self.assertIsInstance(msg, bt2._StreamEndMessageConst)
 354                 self.assertIs(type(msg.stream), bt2_stream._StreamConst)
 355                 self.assertEqual(msg.stream.addr, self._stream.addr)
 356                 with self.assertRaisesRegex(
 357                     ValueError, 'stream class has no default clock class'
 358                 ):
 359                     msg.default_clock_snapshot
 360             else:
 361                 raise Exception
 362
 363     def test_msg_stream_with_clock_snapshots(self):
 364         params = {'with_cc': True, 'with_stream_msgs_clock_snapshots': True}
 365
 366         self._src_comp = self._graph.add_component(self._src, 'my_source', params)
 367         self._msg_iter = TestOutputPortMessageIterator(
 368             self._graph, self._src_comp.output_ports['out']
 369         )
 370         msgs = list(self._msg_iter)
 371
 372         msg_stream_beg = msgs[0]
 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         )
 378         self.assertEqual(msg_stream_beg.default_clock_snapshot.value, 0)
 379
 380         msg_stream_end = msgs[7]
 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         )
 386         self.assertEqual(msg_stream_end.default_clock_snapshot.value, 7)
 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)
 407
 408
 409 class 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)
 420         # Broken at the moment.
 421         # self.assertIs(msg.count, None)
 422
 423     # With event count.
 424     def test_create_with_count(self):
 425         def create_stream_class(tc, cc):
 426             return tc.create_stream_class(supports_discarded_events=True)
 427
 428         def msg_iter_next(msg_iter, stream):
 429             return msg_iter._create_discarded_events_message(stream, count=242)
 430
 431         msg = utils.run_in_message_iterator_next(create_stream_class, msg_iter_next)
 432         self.assertIs(type(msg), bt2._DiscardedEventsMessage)
 433         # Broken at the moment.
 434         # self.assertEqual(msg.count, 242)
 435
 436     # With clock snapshots.
 437     def test_create_with_clock_snapshots(self):
 438         def create_stream_class(tc, cc):
 439             return tc.create_stream_class(
 440                 default_clock_class=cc,
 441                 supports_discarded_events=True,
 442                 discarded_events_have_default_clock_snapshots=True,
 443             )
 444
 445         def msg_iter_next(msg_iter, stream):
 446             return msg_iter._create_discarded_events_message(
 447                 stream, beg_clock_snapshot=10, end_clock_snapshot=20
 448             )
 449
 450         msg = utils.run_in_message_iterator_next(create_stream_class, msg_iter_next)
 451         self.assertIs(type(msg), bt2._DiscardedEventsMessage)
 452         # Broken at the moment.
 453         # self.assertEqual(msg.beginning_default_clock_snapshot, 10);
 454         # self.assertEqual(msg.end_default_clock_snapshot, 20);
 455
 456     # Trying to create when the stream does not support discarded events.
 457     def test_create_unsupported_raises(self):
 458         def create_stream_class(tc, cc):
 459             return tc.create_stream_class()
 460
 461         def msg_iter_next(msg_iter, stream):
 462             with self.assertRaisesRegex(
 463                 ValueError, 'stream class does not support discarded events'
 464             ):
 465                 msg_iter._create_discarded_events_message(stream)
 466
 467             return 123
 468
 469         res = utils.run_in_message_iterator_next(create_stream_class, msg_iter_next)
 470         self.assertEqual(res, 123)
 471
 472     # Trying to create with clock snapshots when the stream does not support
 473     # them.
 474     def test_create_unsupported_clock_snapshots_raises(self):
 475         def create_stream_class(tc, cc):
 476             return tc.create_stream_class(supports_discarded_events=True)
 477
 478         def msg_iter_next(msg_iter, stream):
 479             with self.assertRaisesRegex(
 480                 ValueError,
 481                 'discarded events have no default clock snapshots for this stream class',
 482             ):
 483                 msg_iter._create_discarded_events_message(
 484                     stream, beg_clock_snapshot=10, end_clock_snapshot=20
 485                 )
 486
 487             return 123
 488
 489         res = utils.run_in_message_iterator_next(create_stream_class, msg_iter_next)
 490         self.assertEqual(res, 123)
 491
 492     # Trying to create without clock snapshots when the stream requires them.
 493     def test_create_missing_clock_snapshots_raises(self):
 494         def create_stream_class(tc, cc):
 495             return tc.create_stream_class(
 496                 default_clock_class=cc,
 497                 supports_discarded_events=True,
 498                 discarded_events_have_default_clock_snapshots=True,
 499             )
 500
 501         def msg_iter_next(msg_iter, stream):
 502             with self.assertRaisesRegex(
 503                 ValueError,
 504                 'discarded events have default clock snapshots for this stream class',
 505             ):
 506                 msg_iter._create_discarded_events_message(stream)
 507
 508             return 123
 509
 510         res = utils.run_in_message_iterator_next(create_stream_class, msg_iter_next)
 511         self.assertEqual(res, 123)
 512
 513
 514 class CreateDiscardedPacketMessageTestCase(unittest.TestCase):
 515     # Most basic case.
 516     def test_create(self):
 517         def create_stream_class(tc, cc):
 518             return tc.create_stream_class(
 519                 supports_packets=True, supports_discarded_packets=True
 520             )
 521
 522         def msg_iter_next(msg_iter, stream):
 523             return msg_iter._create_discarded_packets_message(stream)
 524
 525         msg = utils.run_in_message_iterator_next(create_stream_class, msg_iter_next)
 526         self.assertIs(type(msg), bt2._DiscardedPacketsMessage)
 527         self.assertIs(msg.count, None)
 528
 529     # With packet count.
 530     def test_create_with_count(self):
 531         def create_stream_class(tc, cc):
 532             return tc.create_stream_class(
 533                 supports_packets=True, supports_discarded_packets=True
 534             )
 535
 536         def msg_iter_next(msg_iter, stream):
 537             return msg_iter._create_discarded_packets_message(stream, count=242)
 538
 539         msg = utils.run_in_message_iterator_next(create_stream_class, msg_iter_next)
 540         self.assertIs(type(msg), bt2._DiscardedPacketsMessage)
 541         self.assertEqual(msg.count, 242)
 542
 543     # With clock snapshots.
 544     def test_create_with_clock_snapshots(self):
 545         def create_stream_class(tc, cc):
 546             return tc.create_stream_class(
 547                 default_clock_class=cc,
 548                 supports_packets=True,
 549                 supports_discarded_packets=True,
 550                 discarded_packets_have_default_clock_snapshots=True,
 551             )
 552
 553         def msg_iter_next(msg_iter, stream):
 554             return msg_iter._create_discarded_packets_message(
 555                 stream, beg_clock_snapshot=10, end_clock_snapshot=20
 556             )
 557
 558         msg = utils.run_in_message_iterator_next(create_stream_class, msg_iter_next)
 559         self.assertIs(type(msg), bt2._DiscardedPacketsMessage)
 560         self.assertEqual(msg.beginning_default_clock_snapshot, 10)
 561         self.assertEqual(msg.end_default_clock_snapshot, 20)
 562
 563     # Trying to create when the stream does not support discarded packets.
 564     def test_create_unsupported_raises(self):
 565         def create_stream_class(tc, cc):
 566             return tc.create_stream_class(supports_packets=True,)
 567
 568         def msg_iter_next(msg_iter, stream):
 569             with self.assertRaisesRegex(
 570                 ValueError, 'stream class does not support discarded packets'
 571             ):
 572                 msg_iter._create_discarded_packets_message(stream)
 573
 574             return 123
 575
 576         res = utils.run_in_message_iterator_next(create_stream_class, msg_iter_next)
 577         self.assertEqual(res, 123)
 578
 579     # Trying to create with clock snapshots when the stream does not support
 580     # them.
 581     def test_create_unsupported_clock_snapshots_raises(self):
 582         def create_stream_class(tc, cc):
 583             return tc.create_stream_class(
 584                 supports_packets=True, supports_discarded_packets=True
 585             )
 586
 587         def msg_iter_next(msg_iter, stream):
 588             with self.assertRaisesRegex(
 589                 ValueError,
 590                 'discarded packets have no default clock snapshots for this stream class',
 591             ):
 592                 msg_iter._create_discarded_packets_message(
 593                     stream, beg_clock_snapshot=10, end_clock_snapshot=20
 594                 )
 595
 596             return 123
 597
 598         res = utils.run_in_message_iterator_next(create_stream_class, msg_iter_next)
 599         self.assertEqual(res, 123)
 600
 601     # Trying to create without clock snapshots when the stream requires them.
 602     def test_create_missing_clock_snapshots_raises(self):
 603         def create_stream_class(tc, cc):
 604             return tc.create_stream_class(
 605                 default_clock_class=cc,
 606                 supports_packets=True,
 607                 supports_discarded_packets=True,
 608                 discarded_packets_have_default_clock_snapshots=True,
 609             )
 610
 611         def msg_iter_next(msg_iter, stream):
 612             with self.assertRaisesRegex(
 613                 ValueError,
 614                 'discarded packets have default clock snapshots for this stream class',
 615             ):
 616                 msg_iter._create_discarded_packets_message(stream)
 617
 618             return 123
 619
 620         res = utils.run_in_message_iterator_next(create_stream_class, msg_iter_next)
 621         self.assertEqual(res, 123)
 622
 623
 624 if __name__ == '__main__':
 625     unittest.main()