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