tests/bindings/python/bt2/test_message.py

   1 # SPDX-License-Identifier: GPL-2.0-only
   2 #
   3 # Copyright (C) 2019 EfficiOS Inc.
   4 #
   5
   6 import unittest
   7 import bt2
   8 import utils
   9 from utils import TestOutputPortMessageIterator
  10 from bt2 import clock_snapshot as bt2_clock_snapshot
  11 from bt2 import event as bt2_event
  12 from bt2 import event_class as bt2_event_class
  13 from bt2 import field as bt2_field
  14 from bt2 import packet as bt2_packet
  15 from bt2 import stream as bt2_stream
  16 from bt2 import stream_class as bt2_stream_class
  17 from bt2 import trace as bt2_trace
  18 from bt2 import trace_class as bt2_trace_class
  19
  20
  21 class AllMessagesTestCase(unittest.TestCase):
  22     def setUp(self):
  23         class MyIter(bt2._UserMessageIterator):
  24             def __init__(self, config, self_port_output):
  25                 self._at = 0
  26                 self._with_stream_msgs_clock_snapshots = self_port_output.user_data.get(
  27                     'with_stream_msgs_clock_snapshots', False
  28                 )
  29
  30             def __next__(self):
  31                 if test_obj._clock_class:
  32                     if self._at == 0:
  33                         if self._with_stream_msgs_clock_snapshots:
  34                             msg = self._create_stream_beginning_message(
  35                                 test_obj._stream, default_clock_snapshot=self._at
  36                             )
  37                         else:
  38                             msg = self._create_stream_beginning_message(
  39                                 test_obj._stream
  40                             )
  41                         test_obj.assertIs(type(msg), bt2._StreamBeginningMessage)
  42                     elif self._at == 1:
  43                         msg = self._create_packet_beginning_message(
  44                             test_obj._packet, self._at
  45                         )
  46                         test_obj.assertIs(type(msg), bt2._PacketBeginningMessage)
  47                     elif self._at == 2:
  48                         msg = self._create_event_message(
  49                             test_obj._event_class, test_obj._packet, self._at
  50                         )
  51                         test_obj.assertIs(type(msg), bt2._EventMessage)
  52                     elif self._at == 3:
  53                         msg = self._create_message_iterator_inactivity_message(
  54                             test_obj._clock_class, self._at
  55                         )
  56                     elif self._at == 4:
  57                         msg = self._create_discarded_events_message(
  58                             test_obj._stream, 890, self._at, self._at
  59                         )
  60                         test_obj.assertIs(type(msg), bt2._DiscardedEventsMessage)
  61                     elif self._at == 5:
  62                         msg = self._create_packet_end_message(
  63                             test_obj._packet, self._at
  64                         )
  65                         test_obj.assertIs(type(msg), bt2._PacketEndMessage)
  66                     elif self._at == 6:
  67                         msg = self._create_discarded_packets_message(
  68                             test_obj._stream, 678, self._at, self._at
  69                         )
  70                         test_obj.assertIs(type(msg), bt2._DiscardedPacketsMessage)
  71                     elif self._at == 7:
  72                         if self._with_stream_msgs_clock_snapshots:
  73                             msg = self._create_stream_end_message(
  74                                 test_obj._stream, default_clock_snapshot=self._at
  75                             )
  76                         else:
  77                             msg = self._create_stream_end_message(test_obj._stream)
  78                         test_obj.assertIs(type(msg), bt2._StreamEndMessage)
  79                     elif self._at >= 8:
  80                         raise bt2.Stop
  81                 else:
  82                     if self._at == 0:
  83                         msg = self._create_stream_beginning_message(test_obj._stream)
  84                     elif self._at == 1:
  85                         msg = self._create_packet_beginning_message(test_obj._packet)
  86                     elif self._at == 2:
  87                         msg = self._create_event_message(
  88                             test_obj._event_class, test_obj._packet
  89                         )
  90                     elif self._at == 3:
  91                         msg = self._create_discarded_events_message(
  92                             test_obj._stream, 890
  93                         )
  94                     elif self._at == 4:
  95                         msg = self._create_packet_end_message(test_obj._packet)
  96                     elif self._at == 5:
  97                         msg = self._create_discarded_packets_message(
  98                             test_obj._stream, 678
  99                         )
 100                     elif self._at == 6:
 101                         msg = self._create_stream_end_message(test_obj._stream)
 102                     elif self._at >= 7:
 103                         raise bt2.Stop
 104
 105                 self._at += 1
 106                 return msg
 107
 108         class MySrc(bt2._UserSourceComponent, message_iterator_class=MyIter):
 109             def __init__(self, config, params, obj):
 110                 self._add_output_port('out', params)
 111
 112                 with_cc = bool(params['with_cc'])
 113                 tc = self._create_trace_class()
 114                 if with_cc:
 115                     cc = self._create_clock_class()
 116                 else:
 117                     cc = None
 118
 119                 sc = tc.create_stream_class(
 120                     default_clock_class=cc,
 121                     supports_packets=True,
 122                     packets_have_beginning_default_clock_snapshot=with_cc,
 123                     packets_have_end_default_clock_snapshot=with_cc,
 124                     supports_discarded_events=True,
 125                     discarded_events_have_default_clock_snapshots=with_cc,
 126                     supports_discarded_packets=True,
 127                     discarded_packets_have_default_clock_snapshots=with_cc,
 128                 )
 129
 130                 # Create payload field class
 131                 my_int_fc = tc.create_signed_integer_field_class(32)
 132                 payload_fc = tc.create_structure_field_class()
 133                 payload_fc += [('my_int', my_int_fc)]
 134
 135                 # Create specific context field class
 136                 my_int_fc = tc.create_signed_integer_field_class(32)
 137                 specific_fc = tc.create_structure_field_class()
 138                 specific_fc += [('my_int', my_int_fc)]
 139
 140                 ec = sc.create_event_class(
 141                     name='salut',
 142                     payload_field_class=payload_fc,
 143                     specific_context_field_class=specific_fc,
 144                 )
 145
 146                 trace = tc()
 147                 stream = trace.create_stream(sc)
 148                 packet = stream.create_packet()
 149
 150                 test_obj._trace = trace
 151                 test_obj._stream = stream
 152                 test_obj._packet = packet
 153                 test_obj._event_class = ec
 154                 test_obj._clock_class = cc
 155
 156         test_obj = self
 157         self._graph = bt2.Graph()
 158         self._src = MySrc
 159         self._iter = MyIter
 160
 161     def test_all_msg_with_cc(self):
 162         params = {'with_cc': True}
 163         self._src_comp = self._graph.add_component(self._src, 'my_source', params)
 164         self._msg_iter = TestOutputPortMessageIterator(
 165             self._graph, self._src_comp.output_ports['out']
 166         )
 167
 168         for i, msg in enumerate(self._msg_iter):
 169             if i == 0:
 170                 self.assertIs(type(msg), bt2._StreamBeginningMessageConst)
 171                 self.assertIs(type(msg.stream), bt2_stream._StreamConst)
 172                 self.assertEqual(msg.stream.addr, self._stream.addr)
 173                 self.assertIsInstance(
 174                     msg.default_clock_snapshot, bt2._UnknownClockSnapshot
 175                 )
 176             elif i == 1:
 177                 self.assertIs(type(msg), bt2._PacketBeginningMessageConst)
 178                 self.assertIs(type(msg.packet), bt2_packet._PacketConst)
 179                 self.assertIs(
 180                     type(msg.default_clock_snapshot),
 181                     bt2_clock_snapshot._ClockSnapshotConst,
 182                 )
 183                 self.assertEqual(msg.packet.addr, self._packet.addr)
 184                 self.assertEqual(msg.default_clock_snapshot.value, i)
 185             elif i == 2:
 186                 self.assertIs(type(msg), bt2._EventMessageConst)
 187                 self.assertIs(type(msg.event), bt2_event._EventConst)
 188                 self.assertIs(
 189                     type(msg.default_clock_snapshot),
 190                     bt2_clock_snapshot._ClockSnapshotConst,
 191                 )
 192                 self.assertIs(
 193                     type(msg.event.payload_field), bt2_field._StructureFieldConst
 194                 )
 195                 self.assertIs(
 196                     type(msg.event.payload_field['my_int']),
 197                     bt2_field._SignedIntegerFieldConst,
 198                 )
 199
 200                 self.assertEqual(msg.event.cls.addr, self._event_class.addr)
 201                 self.assertEqual(msg.default_clock_snapshot.value, i)
 202             elif i == 3:
 203                 self.assertIs(type(msg), bt2._MessageIteratorInactivityMessageConst)
 204                 self.assertIs(
 205                     type(msg.clock_snapshot), bt2_clock_snapshot._ClockSnapshotConst
 206                 )
 207                 self.assertEqual(msg.clock_snapshot.value, i)
 208             elif i == 4:
 209                 self.assertIs(type(msg), bt2._DiscardedEventsMessageConst)
 210                 self.assertIs(type(msg.stream), bt2_stream._StreamConst)
 211                 self.assertIs(type(msg.stream.cls), bt2_stream_class._StreamClassConst)
 212                 self.assertIs(
 213                     type(msg.beginning_default_clock_snapshot),
 214                     bt2_clock_snapshot._ClockSnapshotConst,
 215                 )
 216                 self.assertIs(
 217                     type(msg.end_default_clock_snapshot),
 218                     bt2_clock_snapshot._ClockSnapshotConst,
 219                 )
 220
 221                 self.assertEqual(msg.stream.addr, self._stream.addr)
 222                 self.assertEqual(msg.count, 890)
 223                 self.assertEqual(
 224                     msg.stream.cls.default_clock_class.addr, self._clock_class.addr
 225                 )
 226                 self.assertEqual(msg.beginning_default_clock_snapshot.value, i)
 227                 self.assertEqual(msg.end_default_clock_snapshot.value, i)
 228             elif i == 5:
 229                 self.assertIs(type(msg), bt2._PacketEndMessageConst)
 230                 self.assertIs(type(msg.packet), bt2_packet._PacketConst)
 231                 self.assertIs(
 232                     type(msg.default_clock_snapshot),
 233                     bt2_clock_snapshot._ClockSnapshotConst,
 234                 )
 235                 self.assertEqual(msg.packet.addr, self._packet.addr)
 236                 self.assertEqual(msg.default_clock_snapshot.value, i)
 237             elif i == 6:
 238                 self.assertIs(type(msg), bt2._DiscardedPacketsMessageConst)
 239                 self.assertIs(type(msg.stream), bt2_stream._StreamConst)
 240                 self.assertIs(type(msg.stream.trace), bt2_trace._TraceConst)
 241                 self.assertIs(
 242                     type(msg.stream.trace.cls), bt2_trace_class._TraceClassConst
 243                 )
 244                 self.assertIs(
 245                     type(msg.beginning_default_clock_snapshot),
 246                     bt2_clock_snapshot._ClockSnapshotConst,
 247                 )
 248                 self.assertIs(
 249                     type(msg.end_default_clock_snapshot),
 250                     bt2_clock_snapshot._ClockSnapshotConst,
 251                 )
 252                 self.assertEqual(msg.stream.addr, self._stream.addr)
 253                 self.assertEqual(msg.count, 678)
 254                 self.assertEqual(
 255                     msg.stream.cls.default_clock_class.addr, self._clock_class.addr
 256                 )
 257                 self.assertEqual(msg.beginning_default_clock_snapshot.value, i)
 258                 self.assertEqual(msg.end_default_clock_snapshot.value, i)
 259             elif i == 7:
 260                 self.assertIs(type(msg), bt2._StreamEndMessageConst)
 261                 self.assertIs(type(msg.stream), bt2_stream._StreamConst)
 262                 self.assertEqual(msg.stream.addr, self._stream.addr)
 263                 self.assertIs(
 264                     type(msg.default_clock_snapshot), bt2._UnknownClockSnapshot
 265                 )
 266             else:
 267                 raise Exception
 268
 269     def test_all_msg_without_cc(self):
 270         params = {'with_cc': False}
 271         self._src_comp = self._graph.add_component(self._src, 'my_source', params)
 272         self._msg_iter = TestOutputPortMessageIterator(
 273             self._graph, self._src_comp.output_ports['out']
 274         )
 275
 276         for i, msg in enumerate(self._msg_iter):
 277             if i == 0:
 278                 self.assertIsInstance(msg, bt2._StreamBeginningMessageConst)
 279                 self.assertIs(type(msg.stream), bt2_stream._StreamConst)
 280                 self.assertEqual(msg.stream.addr, self._stream.addr)
 281                 with self.assertRaisesRegex(
 282                     ValueError, 'stream class has no default clock class'
 283                 ):
 284                     msg.default_clock_snapshot
 285             elif i == 1:
 286                 self.assertIsInstance(msg, bt2._PacketBeginningMessageConst)
 287                 self.assertIs(type(msg.packet), bt2_packet._PacketConst)
 288                 self.assertEqual(msg.packet.addr, self._packet.addr)
 289             elif i == 2:
 290                 self.assertIsInstance(msg, bt2._EventMessageConst)
 291                 self.assertIs(type(msg.event), bt2_event._EventConst)
 292                 self.assertIs(type(msg.event.cls), bt2_event_class._EventClassConst)
 293                 self.assertEqual(msg.event.cls.addr, self._event_class.addr)
 294                 with self.assertRaisesRegex(
 295                     ValueError, 'stream class has no default clock class'
 296                 ):
 297                     msg.default_clock_snapshot
 298             elif i == 3:
 299                 self.assertIsInstance(msg, bt2._DiscardedEventsMessageConst)
 300                 self.assertIs(type(msg.stream), bt2_stream._StreamConst)
 301                 self.assertIs(type(msg.stream.cls), bt2_stream_class._StreamClassConst)
 302                 self.assertEqual(msg.stream.addr, self._stream.addr)
 303                 self.assertEqual(msg.count, 890)
 304                 self.assertIsNone(msg.stream.cls.default_clock_class)
 305                 with self.assertRaisesRegex(
 306                     ValueError,
 307                     'such a message has no clock snapshots for this stream class',
 308                 ):
 309                     msg.beginning_default_clock_snapshot
 310                 with self.assertRaisesRegex(
 311                     ValueError,
 312                     'such a message has no clock snapshots for this stream class',
 313                 ):
 314                     msg.end_default_clock_snapshot
 315             elif i == 4:
 316                 self.assertIsInstance(msg, bt2._PacketEndMessageConst)
 317                 self.assertEqual(msg.packet.addr, self._packet.addr)
 318                 self.assertIs(type(msg.packet), bt2_packet._PacketConst)
 319             elif i == 5:
 320                 self.assertIsInstance(msg, bt2._DiscardedPacketsMessageConst)
 321                 self.assertIs(type(msg.stream), bt2_stream._StreamConst)
 322                 self.assertIs(type(msg.stream.cls), bt2_stream_class._StreamClassConst)
 323                 self.assertIs(
 324                     type(msg.stream.cls.trace_class), bt2_trace_class._TraceClassConst
 325                 )
 326                 self.assertEqual(msg.stream.addr, self._stream.addr)
 327                 self.assertEqual(msg.count, 678)
 328                 self.assertIsNone(msg.stream.cls.default_clock_class)
 329                 with self.assertRaisesRegex(
 330                     ValueError,
 331                     'such a message has no clock snapshots for this stream class',
 332                 ):
 333                     msg.beginning_default_clock_snapshot
 334                 with self.assertRaisesRegex(
 335                     ValueError,
 336                     'such a message has no clock snapshots for this stream class',
 337                 ):
 338                     msg.end_default_clock_snapshot
 339             elif i == 6:
 340                 self.assertIsInstance(msg, bt2._StreamEndMessageConst)
 341                 self.assertIs(type(msg.stream), bt2_stream._StreamConst)
 342                 self.assertEqual(msg.stream.addr, self._stream.addr)
 343                 with self.assertRaisesRegex(
 344                     ValueError, 'stream class has no default clock class'
 345                 ):
 346                     msg.default_clock_snapshot
 347             else:
 348                 raise Exception
 349
 350     def test_msg_stream_with_clock_snapshots(self):
 351         params = {'with_cc': True, 'with_stream_msgs_clock_snapshots': True}
 352
 353         self._src_comp = self._graph.add_component(self._src, 'my_source', params)
 354         self._msg_iter = TestOutputPortMessageIterator(
 355             self._graph, self._src_comp.output_ports['out']
 356         )
 357         msgs = list(self._msg_iter)
 358
 359         msg_stream_beg = msgs[0]
 360         self.assertIsInstance(msg_stream_beg, bt2._StreamBeginningMessageConst)
 361         self.assertIs(
 362             type(msg_stream_beg.default_clock_snapshot),
 363             bt2_clock_snapshot._ClockSnapshotConst,
 364         )
 365         self.assertEqual(msg_stream_beg.default_clock_snapshot.value, 0)
 366
 367         msg_stream_end = msgs[7]
 368         self.assertIsInstance(msg_stream_end, bt2._StreamEndMessageConst)
 369         self.assertIs(
 370             type(msg_stream_end.default_clock_snapshot),
 371             bt2_clock_snapshot._ClockSnapshotConst,
 372         )
 373         self.assertEqual(msg_stream_end.default_clock_snapshot.value, 7)
 374
 375     def test_stream_beg_msg(self):
 376         msg = utils.get_stream_beginning_message()
 377         self.assertIs(type(msg.stream), bt2_stream._Stream)
 378
 379     def test_stream_end_msg(self):
 380         msg = utils.get_stream_end_message()
 381         self.assertIs(type(msg.stream), bt2_stream._Stream)
 382
 383     def test_packet_beg_msg(self):
 384         msg = utils.get_packet_beginning_message()
 385         self.assertIs(type(msg.packet), bt2_packet._Packet)
 386
 387     def test_packet_end_msg(self):
 388         msg = utils.get_packet_end_message()
 389         self.assertIs(type(msg.packet), bt2_packet._Packet)
 390
 391     def test_event_msg(self):
 392         msg = utils.get_event_message()
 393         self.assertIs(type(msg.event), bt2_event._Event)
 394
 395
 396 if __name__ == '__main__':
 397     unittest.main()