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