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.message._StreamBeginningMessage)
 158                 self.assertEqual(msg.stream.addr, self._stream.addr)
 159                 self.assertIsInstance(
 160                     msg.default_clock_snapshot, bt2.clock_snapshot._UnknownClockSnapshot
 161                 )
 162             elif i == 1:
 163                 self.assertIsInstance(msg, bt2.message._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.message._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(
 172                     msg, bt2.message._MessageIteratorInactivityMessage
 173                 )
 174                 self.assertEqual(msg.default_clock_snapshot.value, i)
 175             elif i == 4:
 176                 self.assertIsInstance(msg, bt2.message._DiscardedEventsMessage)
 177                 self.assertEqual(msg.stream.addr, self._stream.addr)
 178                 self.assertEqual(msg.count, 890)
 179                 self.assertEqual(
 180                     msg.stream.cls.default_clock_class.addr, self._clock_class.addr
 181                 )
 182                 self.assertEqual(msg.beginning_default_clock_snapshot.value, i)
 183                 self.assertEqual(msg.end_default_clock_snapshot.value, i)
 184             elif i == 5:
 185                 self.assertIsInstance(msg, bt2.message._PacketEndMessage)
 186                 self.assertEqual(msg.packet.addr, self._packet.addr)
 187                 self.assertEqual(msg.default_clock_snapshot.value, i)
 188             elif i == 6:
 189                 self.assertIsInstance(msg, bt2.message._DiscardedPacketsMessage)
 190                 self.assertEqual(msg.stream.addr, self._stream.addr)
 191                 self.assertEqual(msg.count, 678)
 192                 self.assertEqual(
 193                     msg.stream.cls.default_clock_class.addr, self._clock_class.addr
 194                 )
 195                 self.assertEqual(msg.beginning_default_clock_snapshot.value, i)
 196                 self.assertEqual(msg.end_default_clock_snapshot.value, i)
 197             elif i == 7:
 198                 self.assertIsInstance(msg, bt2.message._StreamEndMessage)
 199                 self.assertEqual(msg.stream.addr, self._stream.addr)
 200                 self.assertIsInstance(
 201                     msg.default_clock_snapshot, bt2.clock_snapshot._UnknownClockSnapshot
 202                 )
 203             else:
 204                 raise Exception
 205
 206     def test_all_msg_without_cc(self):
 207         params = {'with_cc': False}
 208         self._src_comp = self._graph.add_component(self._src, 'my_source', params)
 209         self._msg_iter = self._graph.create_output_port_message_iterator(
 210             self._src_comp.output_ports['out']
 211         )
 212
 213         for i, msg in enumerate(self._msg_iter):
 214             if i == 0:
 215                 self.assertIsInstance(msg, bt2.message._StreamBeginningMessage)
 216                 self.assertEqual(msg.stream.addr, self._stream.addr)
 217                 with self.assertRaisesRegex(
 218                     ValueError, 'stream class has no default clock class'
 219                 ):
 220                     msg.default_clock_snapshot
 221             elif i == 1:
 222                 self.assertIsInstance(msg, bt2.message._PacketBeginningMessage)
 223                 self.assertEqual(msg.packet.addr, self._packet.addr)
 224             elif i == 2:
 225                 self.assertIsInstance(msg, bt2.message._EventMessage)
 226                 self.assertEqual(msg.event.cls.addr, self._event_class.addr)
 227                 with self.assertRaisesRegex(
 228                     ValueError, 'stream class has no default clock class'
 229                 ):
 230                     msg.default_clock_snapshot
 231             elif i == 3:
 232                 self.assertIsInstance(msg, bt2.message._DiscardedEventsMessage)
 233                 self.assertEqual(msg.stream.addr, self._stream.addr)
 234                 self.assertEqual(msg.count, 890)
 235                 self.assertIsNone(msg.stream.cls.default_clock_class)
 236                 with self.assertRaisesRegex(
 237                     ValueError,
 238                     'such a message has no clock snapshots for this stream class',
 239                 ):
 240                     msg.beginning_default_clock_snapshot
 241                 with self.assertRaisesRegex(
 242                     ValueError,
 243                     'such a message has no clock snapshots for this stream class',
 244                 ):
 245                     msg.end_default_clock_snapshot
 246             elif i == 4:
 247                 self.assertIsInstance(msg, bt2.message._PacketEndMessage)
 248                 self.assertEqual(msg.packet.addr, self._packet.addr)
 249             elif i == 5:
 250                 self.assertIsInstance(msg, bt2.message._DiscardedPacketsMessage)
 251                 self.assertEqual(msg.stream.addr, self._stream.addr)
 252                 self.assertEqual(msg.count, 678)
 253                 self.assertIsNone(msg.stream.cls.default_clock_class)
 254                 with self.assertRaisesRegex(
 255                     ValueError,
 256                     'such a message has no clock snapshots for this stream class',
 257                 ):
 258                     msg.beginning_default_clock_snapshot
 259                 with self.assertRaisesRegex(
 260                     ValueError,
 261                     'such a message has no clock snapshots for this stream class',
 262                 ):
 263                     msg.end_default_clock_snapshot
 264             elif i == 6:
 265                 self.assertIsInstance(msg, bt2.message._StreamEndMessage)
 266                 self.assertEqual(msg.stream.addr, self._stream.addr)
 267                 with self.assertRaisesRegex(
 268                     ValueError, 'stream class has no default clock class'
 269                 ):
 270                     msg.default_clock_snapshot
 271             else:
 272                 raise Exception
 273
 274     def test_msg_stream_with_clock_snapshots(self):
 275         params = {'with_cc': True, 'with_stream_msgs_clock_snapshots': True}
 276
 277         self._src_comp = self._graph.add_component(self._src, 'my_source', params)
 278         self._msg_iter = self._graph.create_output_port_message_iterator(
 279             self._src_comp.output_ports['out']
 280         )
 281         msgs = list(self._msg_iter)
 282
 283         msg_stream_beg = msgs[0]
 284         self.assertIsInstance(msg_stream_beg, bt2.message._StreamBeginningMessage)
 285         self.assertEqual(msg_stream_beg.default_clock_snapshot.value, 0)
 286
 287         msg_stream_end = msgs[7]
 288         self.assertIsInstance(msg_stream_end, bt2.message._StreamEndMessage)
 289         self.assertEqual(msg_stream_end.default_clock_snapshot.value, 7)