tests/bindings/python/bt2/test_event.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
  24 from bt2 import field as bt2_field
  25 from bt2 import stream as bt2_stream
  26 from bt2 import event_class as bt2_event_class
  27 from bt2 import clock_snapshot as bt2_clock_snapshot
  28
  29
  30 class EventTestCase(unittest.TestCase):
  31     def _create_test_const_event_message(
  32         self,
  33         packet_fields_config=None,
  34         event_fields_config=None,
  35         with_clockclass=False,
  36         with_cc=False,
  37         with_sc=False,
  38         with_ep=False,
  39         with_packet=False,
  40     ):
  41         class MyIter(bt2._UserMessageIterator):
  42             def __init__(self, config, self_output_port):
  43                 self._at = 0
  44                 self._msgs = [self._create_stream_beginning_message(test_obj.stream)]
  45
  46                 if with_packet:
  47                     assert test_obj.packet
  48                     self._msgs.append(
  49                         self._create_packet_beginning_message(test_obj.packet)
  50                     )
  51
  52                 default_clock_snapshot = 789 if with_clockclass else None
  53
  54                 if with_packet:
  55                     assert test_obj.packet
  56                     ev_parent = test_obj.packet
  57                 else:
  58                     assert test_obj.stream
  59                     ev_parent = test_obj.stream
  60
  61                 msg = self._create_event_message(
  62                     test_obj.event_class, ev_parent, default_clock_snapshot
  63                 )
  64
  65                 if event_fields_config is not None:
  66                     event_fields_config(msg.event)
  67
  68                 self._msgs.append(msg)
  69
  70                 if with_packet:
  71                     self._msgs.append(self._create_packet_end_message(test_obj.packet))
  72
  73                 self._msgs.append(self._create_stream_end_message(test_obj.stream))
  74
  75             def __next__(self):
  76                 if self._at == len(self._msgs):
  77                     raise bt2.Stop
  78
  79                 msg = self._msgs[self._at]
  80                 self._at += 1
  81                 return msg
  82
  83         class MySrc(bt2._UserSourceComponent, message_iterator_class=MyIter):
  84             def __init__(self, config, params, obj):
  85                 self._add_output_port('out')
  86                 tc = self._create_trace_class()
  87
  88                 clock_class = None
  89                 if with_clockclass:
  90                     clock_class = self._create_clock_class(frequency=1000)
  91
  92                 # event common context (stream-class-defined)
  93                 cc = None
  94                 if with_cc:
  95                     cc = tc.create_structure_field_class()
  96                     cc += [
  97                         ('cpu_id', tc.create_signed_integer_field_class(8)),
  98                         ('stuff', tc.create_double_precision_real_field_class()),
  99                     ]
 100
 101                 # packet context (stream-class-defined)
 102                 pc = None
 103
 104                 if with_packet:
 105                     pc = tc.create_structure_field_class()
 106                     pc += [
 107                         ('something', tc.create_unsigned_integer_field_class(8)),
 108                         (
 109                             'something_else',
 110                             tc.create_double_precision_real_field_class(),
 111                         ),
 112                     ]
 113
 114                 stream_class = tc.create_stream_class(
 115                     default_clock_class=clock_class,
 116                     event_common_context_field_class=cc,
 117                     packet_context_field_class=pc,
 118                     supports_packets=with_packet,
 119                 )
 120
 121                 # specific context (event-class-defined)
 122                 sc = None
 123                 if with_sc:
 124                     sc = tc.create_structure_field_class()
 125                     sc += [
 126                         ('ant', tc.create_signed_integer_field_class(16)),
 127                         ('msg', tc.create_string_field_class()),
 128                     ]
 129
 130                 # event payload
 131                 ep = None
 132                 if with_ep:
 133                     ep = tc.create_structure_field_class()
 134                     ep += [
 135                         ('giraffe', tc.create_signed_integer_field_class(32)),
 136                         ('gnu', tc.create_signed_integer_field_class(8)),
 137                         ('mosquito', tc.create_signed_integer_field_class(8)),
 138                     ]
 139
 140                 event_class = stream_class.create_event_class(
 141                     name='garou',
 142                     specific_context_field_class=sc,
 143                     payload_field_class=ep,
 144                 )
 145
 146                 trace = tc()
 147                 stream = trace.create_stream(stream_class)
 148
 149                 if with_packet:
 150                     packet = stream.create_packet()
 151
 152                 if packet_fields_config is not None:
 153                     assert packet
 154                     packet_fields_config(packet)
 155
 156                 if with_packet:
 157                     test_obj.packet = packet
 158
 159                 test_obj.stream = stream
 160                 test_obj.event_class = event_class
 161
 162         test_obj = self
 163         self._graph = bt2.Graph()
 164         self._src_comp = self._graph.add_component(MySrc, 'my_source')
 165         self._msg_iter = TestOutputPortMessageIterator(
 166             self._graph, self._src_comp.output_ports['out']
 167         )
 168
 169         for msg in self._msg_iter:
 170             if type(msg) is bt2._EventMessageConst:
 171                 self._event_msg = msg
 172                 return msg
 173
 174     def test_const_attr_event_class(self):
 175         msg = self._create_test_const_event_message()
 176         self.assertEqual(msg.event.cls.addr, self.event_class.addr)
 177         self.assertIs(type(msg.event.cls), bt2_event_class._EventClassConst)
 178
 179     def test_attr_event_class(self):
 180         msg = utils.get_event_message()
 181         self.assertIs(type(msg.event.cls), bt2_event_class._EventClass)
 182
 183     def test_const_attr_name(self):
 184         msg = self._create_test_const_event_message()
 185         self.assertEqual(msg.event.name, self.event_class.name)
 186
 187     def test_const_attr_id(self):
 188         msg = self._create_test_const_event_message()
 189         self.assertEqual(msg.event.id, self.event_class.id)
 190
 191     def test_const_get_common_context_field(self):
 192         def event_fields_config(event):
 193             event.common_context_field['cpu_id'] = 1
 194             event.common_context_field['stuff'] = 13.194
 195
 196         msg = self._create_test_const_event_message(
 197             event_fields_config=event_fields_config, with_cc=True
 198         )
 199
 200         self.assertEqual(msg.event.common_context_field['cpu_id'], 1)
 201         self.assertEqual(msg.event.common_context_field['stuff'], 13.194)
 202         self.assertIs(
 203             type(msg.event.common_context_field), bt2_field._StructureFieldConst
 204         )
 205
 206     def test_attr_common_context_field(self):
 207         msg = utils.get_event_message()
 208         self.assertIs(type(msg.event.common_context_field), bt2_field._StructureField)
 209
 210     def test_const_no_common_context_field(self):
 211         msg = self._create_test_const_event_message(with_cc=False)
 212         self.assertIsNone(msg.event.common_context_field)
 213
 214     def test_const_get_specific_context_field(self):
 215         def event_fields_config(event):
 216             event.specific_context_field['ant'] = -1
 217             event.specific_context_field['msg'] = 'hellooo'
 218
 219         msg = self._create_test_const_event_message(
 220             event_fields_config=event_fields_config, with_sc=True
 221         )
 222
 223         self.assertEqual(msg.event.specific_context_field['ant'], -1)
 224         self.assertEqual(msg.event.specific_context_field['msg'], 'hellooo')
 225         self.assertIs(
 226             type(msg.event.specific_context_field), bt2_field._StructureFieldConst
 227         )
 228
 229     def test_attr_specific_context_field(self):
 230         msg = utils.get_event_message()
 231         self.assertIs(type(msg.event.specific_context_field), bt2_field._StructureField)
 232
 233     def test_const_no_specific_context_field(self):
 234         msg = self._create_test_const_event_message(with_sc=False)
 235         self.assertIsNone(msg.event.specific_context_field)
 236
 237     def test_const_get_event_payload_field(self):
 238         def event_fields_config(event):
 239             event.payload_field['giraffe'] = 1
 240             event.payload_field['gnu'] = 23
 241             event.payload_field['mosquito'] = 42
 242
 243         msg = self._create_test_const_event_message(
 244             event_fields_config=event_fields_config, with_ep=True
 245         )
 246
 247         self.assertEqual(msg.event.payload_field['giraffe'], 1)
 248         self.assertEqual(msg.event.payload_field['gnu'], 23)
 249         self.assertEqual(msg.event.payload_field['mosquito'], 42)
 250         self.assertIs(type(msg.event.payload_field), bt2_field._StructureFieldConst)
 251
 252     def test_attr_payload_field(self):
 253         msg = utils.get_event_message()
 254         self.assertIs(type(msg.event.payload_field), bt2_field._StructureField)
 255
 256     def test_const_no_payload_field(self):
 257         msg = self._create_test_const_event_message(with_ep=False)
 258         self.assertIsNone(msg.event.payload_field)
 259
 260     def test_const_clock_value(self):
 261         msg = self._create_test_const_event_message(with_clockclass=True)
 262         self.assertEqual(msg.default_clock_snapshot.value, 789)
 263         self.assertIs(
 264             type(msg.default_clock_snapshot), bt2_clock_snapshot._ClockSnapshotConst
 265         )
 266
 267     def test_clock_value(self):
 268         msg = utils.get_event_message()
 269         self.assertEqual(msg.default_clock_snapshot.value, 789)
 270         self.assertIs(
 271             type(msg.default_clock_snapshot), bt2_clock_snapshot._ClockSnapshotConst
 272         )
 273
 274     def test_const_no_clock_value(self):
 275         msg = self._create_test_const_event_message(with_clockclass=False)
 276         with self.assertRaisesRegex(
 277             ValueError, 'stream class has no default clock class'
 278         ):
 279             msg.default_clock_snapshot
 280
 281     def test_const_stream(self):
 282         msg = self._create_test_const_event_message()
 283         self.assertEqual(msg.event.stream.addr, self.stream.addr)
 284         self.assertIs(type(msg.event.stream), bt2_stream._StreamConst)
 285
 286     def test_stream(self):
 287         msg = utils.get_event_message()
 288         self.assertIs(type(msg.event.stream), bt2_stream._Stream)
 289
 290     def test_const_getitem(self):
 291         def event_fields_config(event):
 292             event.payload_field['giraffe'] = 1
 293             event.payload_field['gnu'] = 23
 294             event.payload_field['mosquito'] = 42
 295             event.specific_context_field['ant'] = -1
 296             event.specific_context_field['msg'] = 'hellooo'
 297             event.common_context_field['cpu_id'] = 1
 298             event.common_context_field['stuff'] = 13.194
 299
 300         def packet_fields_config(packet):
 301             packet.context_field['something'] = 154
 302             packet.context_field['something_else'] = 17.2
 303
 304         msg = self._create_test_const_event_message(
 305             packet_fields_config=packet_fields_config,
 306             event_fields_config=event_fields_config,
 307             with_cc=True,
 308             with_sc=True,
 309             with_ep=True,
 310             with_packet=True,
 311         )
 312         ev = msg.event
 313
 314         # Test event fields
 315         self.assertEqual(ev['giraffe'], 1)
 316         self.assertIs(type(ev['giraffe']), bt2_field._SignedIntegerFieldConst)
 317         self.assertEqual(ev['gnu'], 23)
 318         self.assertEqual(ev['mosquito'], 42)
 319         self.assertEqual(ev['ant'], -1)
 320         self.assertIs(type(ev['ant']), bt2_field._SignedIntegerFieldConst)
 321         self.assertEqual(ev['msg'], 'hellooo')
 322         self.assertEqual(ev['cpu_id'], 1)
 323         self.assertIs(type(ev['cpu_id']), bt2_field._SignedIntegerFieldConst)
 324         self.assertEqual(ev['stuff'], 13.194)
 325
 326         # Test packet fields
 327         self.assertEqual(ev['something'], 154)
 328         self.assertIs(type(ev['something']), bt2_field._UnsignedIntegerFieldConst)
 329         self.assertEqual(ev['something_else'], 17.2)
 330
 331         with self.assertRaises(KeyError):
 332             ev['yes']
 333
 334     def test_const_getitem_no_packet(self):
 335         def event_fields_config(event):
 336             event.payload_field['giraffe'] = 1
 337             event.payload_field['gnu'] = 23
 338             event.payload_field['mosquito'] = 42
 339
 340         msg = self._create_test_const_event_message(
 341             event_fields_config=event_fields_config, with_ep=True,
 342         )
 343         ev = msg.event
 344
 345         with self.assertRaises(KeyError):
 346             ev['yes']
 347
 348     def test_getitem(self):
 349         msg = utils.get_event_message()
 350         ev = msg.event
 351         self.assertEqual(ev['giraffe'], 1)
 352         self.assertIs(type(ev['giraffe']), bt2_field._SignedIntegerField)
 353         self.assertEqual(ev['ant'], -1)
 354         self.assertIs(type(ev['ant']), bt2_field._SignedIntegerField)
 355         self.assertEqual(ev['cpu_id'], 1)
 356         self.assertIs(type(ev['cpu_id']), bt2_field._SignedIntegerField)
 357         self.assertEqual(ev['something'], 154)
 358         self.assertIs(type(ev['something']), bt2_field._UnsignedIntegerField)
 359
 360
 361 if __name__ == "__main__":
 362     unittest.main()