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 if __name__ == '__main__':
 410     unittest.main()