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 collections
  20 import unittest
  21 import bt2
  22
  23
  24 class AllMessagesTestCase(unittest.TestCase):
  25     def setUp(self):
  26         class MyIter(bt2._UserMessageIterator):
  27             def __init__(self, self_port_output):
  28                 self._at = 0
  29                 self._with_stream_msgs_clock_snapshots = self_port_output.user_data.get(
  30                     'with_stream_msgs_clock_snapshots', False
  31                 )
  32
  33             def __next__(self):
  34                 if test_obj._clock_class:
  35                     if self._at == 0:
  36                         if self._with_stream_msgs_clock_snapshots:
  37                             msg = self._create_stream_beginning_message(
  38                                 test_obj._stream, default_clock_snapshot=self._at
  39                             )
  40                         else:
  41                             msg = self._create_stream_beginning_message(
  42                                 test_obj._stream
  43                             )
  44                     elif self._at == 1:
  45                         msg = self._create_packet_beginning_message(
  46                             test_obj._packet, self._at
  47                         )
  48                     elif self._at == 2:
  49                         msg = self._create_event_message(
  50                             test_obj._event_class, test_obj._packet, self._at
  51                         )
  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                     elif self._at == 5:
  61                         msg = self._create_packet_end_message(
  62                             test_obj._packet, self._at
  63                         )
  64                     elif self._at == 6:
  65                         msg = self._create_discarded_packets_message(
  66                             test_obj._stream, 678, self._at, self._at
  67                         )
  68                     elif self._at == 7:
  69                         if self._with_stream_msgs_clock_snapshots:
  70                             msg = self._create_stream_end_message(
  71                                 test_obj._stream, default_clock_snapshot=self._at
  72                             )
  73                         else:
  74                             msg = self._create_stream_end_message(test_obj._stream)
  75                     elif self._at >= 8:
  76                         raise bt2.Stop
  77                 else:
  78                     if self._at == 0:
  79                         msg = self._create_stream_beginning_message(test_obj._stream)
  80                     elif self._at == 1:
  81                         msg = self._create_packet_beginning_message(test_obj._packet)
  82                     elif self._at == 2:
  83                         msg = self._create_event_message(
  84                             test_obj._event_class, test_obj._packet
  85                         )
  86                     elif self._at == 3:
  87                         msg = self._create_discarded_events_message(
  88                             test_obj._stream, 890
  89                         )
  90                     elif self._at == 4:
  91                         msg = self._create_packet_end_message(test_obj._packet)
  92                     elif self._at == 5:
  93                         msg = self._create_discarded_packets_message(
  94                             test_obj._stream, 678
  95                         )
  96                     elif self._at == 6:
  97                         msg = self._create_stream_end_message(test_obj._stream)
  98                     elif self._at >= 7:
  99                         raise bt2.Stop
 100
 101                 self._at += 1
 102                 return msg
 103
 104         class MySrc(bt2._UserSourceComponent, message_iterator_class=MyIter):
 105             def __init__(self, params):
 106                 self._add_output_port('out', params)
 107
 108                 with_cc = bool(params['with_cc'])
 109                 tc = self._create_trace_class()
 110                 if with_cc:
 111                     cc = self._create_clock_class()
 112                 else:
 113                     cc = None
 114
 115                 sc = tc.create_stream_class(
 116                     default_clock_class=cc,
 117                     supports_packets=True,
 118                     packets_have_beginning_default_clock_snapshot=with_cc,
 119                     packets_have_end_default_clock_snapshot=with_cc,
 120                     supports_discarded_events=True,
 121                     discarded_events_have_default_clock_snapshots=with_cc,
 122                     supports_discarded_packets=True,
 123                     discarded_packets_have_default_clock_snapshots=with_cc,
 124                 )
 125
 126                 # Create payload field class
 127                 my_int_fc = tc.create_signed_integer_field_class(32)
 128                 payload_fc = tc.create_structure_field_class()
 129                 payload_fc += [('my_int', my_int_fc)]
 130
 131                 ec = sc.create_event_class(name='salut', payload_field_class=payload_fc)
 132
 133                 trace = tc()
 134                 stream = trace.create_stream(sc)
 135                 packet = stream.create_packet()
 136
 137                 test_obj._trace = trace
 138                 test_obj._stream = stream
 139                 test_obj._packet = packet
 140                 test_obj._event_class = ec
 141                 test_obj._clock_class = cc
 142
 143         test_obj = self
 144         self._graph = bt2.Graph()
 145         self._src = MySrc
 146         self._iter = MyIter
 147
 148     def test_all_msg_with_cc(self):
 149         params = {'with_cc': True}
 150         self._src_comp = self._graph.add_component(self._src, 'my_source', params)
 151         self._msg_iter = self._graph.create_output_port_message_iterator(
 152             self._src_comp.output_ports['out']
 153         )
 154
 155         for i, msg in enumerate(self._msg_iter):
 156             if i == 0:
 157                 self.assertIsInstance(msg, bt2._StreamBeginningMessage)
 158                 self.assertEqual(msg.stream.addr, self._stream.addr)
 159                 self.assertIsInstance(
 160                     msg.default_clock_snapshot, bt2._UnknownClockSnapshot
 161                 )
 162             elif i == 1:
 163                 self.assertIsInstance(msg, bt2._PacketBeginningMessage)
 164                 self.assertEqual(msg.packet.addr, self._packet.addr)
 165                 self.assertEqual(msg.default_clock_snapshot.value, i)
 166             elif i == 2:
 167                 self.assertIsInstance(msg, bt2._EventMessage)
 168                 self.assertEqual(msg.event.cls.addr, self._event_class.addr)
 169                 self.assertEqual(msg.default_clock_snapshot.value, i)
 170             elif i == 3:
 171                 self.assertIsInstance(msg, bt2._MessageIteratorInactivityMessage)
 172                 self.assertEqual(msg.default_clock_snapshot.value, i)
 173             elif i == 4:
 174                 self.assertIsInstance(msg, bt2._DiscardedEventsMessage)
 175                 self.assertEqual(msg.stream.addr, self._stream.addr)
 176                 self.assertEqual(msg.count, 890)
 177                 self.assertEqual(
 178                     msg.stream.cls.default_clock_class.addr, self._clock_class.addr
 179                 )
 180                 self.assertEqual(msg.beginning_default_clock_snapshot.value, i)
 181                 self.assertEqual(msg.end_default_clock_snapshot.value, i)
 182             elif i == 5:
 183                 self.assertIsInstance(msg, bt2._PacketEndMessage)
 184                 self.assertEqual(msg.packet.addr, self._packet.addr)
 185                 self.assertEqual(msg.default_clock_snapshot.value, i)
 186             elif i == 6:
 187                 self.assertIsInstance(msg, bt2._DiscardedPacketsMessage)
 188                 self.assertEqual(msg.stream.addr, self._stream.addr)
 189                 self.assertEqual(msg.count, 678)
 190                 self.assertEqual(
 191                     msg.stream.cls.default_clock_class.addr, self._clock_class.addr
 192                 )
 193                 self.assertEqual(msg.beginning_default_clock_snapshot.value, i)
 194                 self.assertEqual(msg.end_default_clock_snapshot.value, i)
 195             elif i == 7:
 196                 self.assertIsInstance(msg, bt2._StreamEndMessage)
 197                 self.assertEqual(msg.stream.addr, self._stream.addr)
 198                 self.assertIsInstance(
 199                     msg.default_clock_snapshot, bt2._UnknownClockSnapshot
 200                 )
 201             else:
 202                 raise Exception
 203
 204     def test_all_msg_without_cc(self):
 205         params = {'with_cc': False}
 206         self._src_comp = self._graph.add_component(self._src, 'my_source', params)
 207         self._msg_iter = self._graph.create_output_port_message_iterator(
 208             self._src_comp.output_ports['out']
 209         )
 210
 211         for i, msg in enumerate(self._msg_iter):
 212             if i == 0:
 213                 self.assertIsInstance(msg, bt2._StreamBeginningMessage)
 214                 self.assertEqual(msg.stream.addr, self._stream.addr)
 215                 with self.assertRaisesRegex(
 216                     ValueError, 'stream class has no default clock class'
 217                 ):
 218                     msg.default_clock_snapshot
 219             elif i == 1:
 220                 self.assertIsInstance(msg, bt2._PacketBeginningMessage)
 221                 self.assertEqual(msg.packet.addr, self._packet.addr)
 222             elif i == 2:
 223                 self.assertIsInstance(msg, bt2._EventMessage)
 224                 self.assertEqual(msg.event.cls.addr, self._event_class.addr)
 225                 with self.assertRaisesRegex(
 226                     ValueError, 'stream class has no default clock class'
 227                 ):
 228                     msg.default_clock_snapshot
 229             elif i == 3:
 230                 self.assertIsInstance(msg, bt2._DiscardedEventsMessage)
 231                 self.assertEqual(msg.stream.addr, self._stream.addr)
 232                 self.assertEqual(msg.count, 890)
 233                 self.assertIsNone(msg.stream.cls.default_clock_class)
 234                 with self.assertRaisesRegex(
 235                     ValueError,
 236                     'such a message has no clock snapshots for this stream class',
 237                 ):
 238                     msg.beginning_default_clock_snapshot
 239                 with self.assertRaisesRegex(
 240                     ValueError,
 241                     'such a message has no clock snapshots for this stream class',
 242                 ):
 243                     msg.end_default_clock_snapshot
 244             elif i == 4:
 245                 self.assertIsInstance(msg, bt2._PacketEndMessage)
 246                 self.assertEqual(msg.packet.addr, self._packet.addr)
 247             elif i == 5:
 248                 self.assertIsInstance(msg, bt2._DiscardedPacketsMessage)
 249                 self.assertEqual(msg.stream.addr, self._stream.addr)
 250                 self.assertEqual(msg.count, 678)
 251                 self.assertIsNone(msg.stream.cls.default_clock_class)
 252                 with self.assertRaisesRegex(
 253                     ValueError,
 254                     'such a message has no clock snapshots for this stream class',
 255                 ):
 256                     msg.beginning_default_clock_snapshot
 257                 with self.assertRaisesRegex(
 258                     ValueError,
 259                     'such a message has no clock snapshots for this stream class',
 260                 ):
 261                     msg.end_default_clock_snapshot
 262             elif i == 6:
 263                 self.assertIsInstance(msg, bt2._StreamEndMessage)
 264                 self.assertEqual(msg.stream.addr, self._stream.addr)
 265                 with self.assertRaisesRegex(
 266                     ValueError, 'stream class has no default clock class'
 267                 ):
 268                     msg.default_clock_snapshot
 269             else:
 270                 raise Exception
 271
 272     def test_msg_stream_with_clock_snapshots(self):
 273         params = {'with_cc': True, 'with_stream_msgs_clock_snapshots': True}
 274
 275         self._src_comp = self._graph.add_component(self._src, 'my_source', params)
 276         self._msg_iter = self._graph.create_output_port_message_iterator(
 277             self._src_comp.output_ports['out']
 278         )
 279         msgs = list(self._msg_iter)
 280
 281         msg_stream_beg = msgs[0]
 282         self.assertIsInstance(msg_stream_beg, bt2._StreamBeginningMessage)
 283         self.assertEqual(msg_stream_beg.default_clock_snapshot.value, 0)
 284
 285         msg_stream_end = msgs[7]
 286         self.assertIsInstance(msg_stream_end, bt2._StreamEndMessage)
 287         self.assertEqual(msg_stream_end.default_clock_snapshot.value, 7)