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 from collections import OrderedDict
  20 import unittest
  21 import bt2
  22 from utils import TestOutputPortMessageIterator
  23
  24
  25 class EventTestCase(unittest.TestCase):
  26     def _create_test_event_message(
  27         self,
  28         packet_fields_config=None,
  29         event_fields_config=None,
  30         with_clockclass=False,
  31         with_cc=False,
  32         with_sc=False,
  33         with_ep=False,
  34         with_packet=False,
  35     ):
  36         class MyIter(bt2._UserMessageIterator):
  37             def __init__(self, self_output_port):
  38                 self._at = 0
  39                 self._msgs = [self._create_stream_beginning_message(test_obj.stream)]
  40
  41                 if with_packet:
  42                     assert test_obj.packet
  43                     self._msgs.append(
  44                         self._create_packet_beginning_message(test_obj.packet)
  45                     )
  46
  47                 default_clock_snapshot = 789 if with_clockclass else None
  48
  49                 if with_packet:
  50                     assert test_obj.packet
  51                     ev_parent = test_obj.packet
  52                 else:
  53                     assert test_obj.stream
  54                     ev_parent = test_obj.stream
  55
  56                 msg = self._create_event_message(
  57                     test_obj.event_class, ev_parent, default_clock_snapshot
  58                 )
  59
  60                 if event_fields_config is not None:
  61                     event_fields_config(msg.event)
  62
  63                 self._msgs.append(msg)
  64
  65                 if with_packet:
  66                     self._msgs.append(self._create_packet_end_message(test_obj.packet))
  67
  68                 self._msgs.append(self._create_stream_end_message(test_obj.stream))
  69
  70             def __next__(self):
  71                 if self._at == len(self._msgs):
  72                     raise bt2.Stop
  73
  74                 msg = self._msgs[self._at]
  75                 self._at += 1
  76                 return msg
  77
  78         class MySrc(bt2._UserSourceComponent, message_iterator_class=MyIter):
  79             def __init__(self, params, obj):
  80                 self._add_output_port('out')
  81                 tc = self._create_trace_class()
  82
  83                 clock_class = None
  84                 if with_clockclass:
  85                     clock_class = self._create_clock_class(frequency=1000)
  86
  87                 # event common context (stream-class-defined)
  88                 cc = None
  89                 if with_cc:
  90                     cc = tc.create_structure_field_class()
  91                     cc += [
  92                         ('cpu_id', tc.create_signed_integer_field_class(8)),
  93                         ('stuff', tc.create_real_field_class()),
  94                     ]
  95
  96                 # packet context (stream-class-defined)
  97                 pc = None
  98
  99                 if with_packet:
 100                     pc = tc.create_structure_field_class()
 101                     pc += [
 102                         ('something', tc.create_unsigned_integer_field_class(8)),
 103                         ('something_else', tc.create_real_field_class()),
 104                     ]
 105
 106                 stream_class = tc.create_stream_class(
 107                     default_clock_class=clock_class,
 108                     event_common_context_field_class=cc,
 109                     packet_context_field_class=pc,
 110                     supports_packets=with_packet,
 111                 )
 112
 113                 # specific context (event-class-defined)
 114                 sc = None
 115                 if with_sc:
 116                     sc = tc.create_structure_field_class()
 117                     sc += [
 118                         ('ant', tc.create_signed_integer_field_class(16)),
 119                         ('msg', tc.create_string_field_class()),
 120                     ]
 121
 122                 # event payload
 123                 ep = None
 124                 if with_ep:
 125                     ep = tc.create_structure_field_class()
 126                     ep += [
 127                         ('giraffe', tc.create_signed_integer_field_class(32)),
 128                         ('gnu', tc.create_signed_integer_field_class(8)),
 129                         ('mosquito', tc.create_signed_integer_field_class(8)),
 130                     ]
 131
 132                 event_class = stream_class.create_event_class(
 133                     name='garou',
 134                     specific_context_field_class=sc,
 135                     payload_field_class=ep,
 136                 )
 137
 138                 trace = tc()
 139                 stream = trace.create_stream(stream_class)
 140
 141                 if with_packet:
 142                     packet = stream.create_packet()
 143
 144                 if packet_fields_config is not None:
 145                     assert packet
 146                     packet_fields_config(packet)
 147
 148                 if with_packet:
 149                     test_obj.packet = packet
 150
 151                 test_obj.stream = stream
 152                 test_obj.event_class = event_class
 153
 154         test_obj = self
 155         self._graph = bt2.Graph()
 156         self._src_comp = self._graph.add_component(MySrc, 'my_source')
 157         self._msg_iter = TestOutputPortMessageIterator(
 158             self._graph, self._src_comp.output_ports['out']
 159         )
 160
 161         for msg in self._msg_iter:
 162             if type(msg) is bt2._EventMessage:
 163                 return msg
 164
 165     def test_attr_event_class(self):
 166         msg = self._create_test_event_message()
 167         self.assertEqual(msg.event.cls.addr, self.event_class.addr)
 168
 169     def test_attr_name(self):
 170         msg = self._create_test_event_message()
 171         self.assertEqual(msg.event.name, self.event_class.name)
 172
 173     def test_attr_id(self):
 174         msg = self._create_test_event_message()
 175         self.assertEqual(msg.event.id, self.event_class.id)
 176
 177     def test_get_common_context_field(self):
 178         def event_fields_config(event):
 179             event.common_context_field['cpu_id'] = 1
 180             event.common_context_field['stuff'] = 13.194
 181
 182         msg = self._create_test_event_message(
 183             event_fields_config=event_fields_config, with_cc=True
 184         )
 185
 186         self.assertEqual(msg.event.common_context_field['cpu_id'], 1)
 187         self.assertEqual(msg.event.common_context_field['stuff'], 13.194)
 188
 189     def test_no_common_context_field(self):
 190         msg = self._create_test_event_message(with_cc=False)
 191         self.assertIsNone(msg.event.common_context_field)
 192
 193     def test_get_specific_context_field(self):
 194         def event_fields_config(event):
 195             event.specific_context_field['ant'] = -1
 196             event.specific_context_field['msg'] = 'hellooo'
 197
 198         msg = self._create_test_event_message(
 199             event_fields_config=event_fields_config, with_sc=True
 200         )
 201
 202         self.assertEqual(msg.event.specific_context_field['ant'], -1)
 203         self.assertEqual(msg.event.specific_context_field['msg'], 'hellooo')
 204
 205     def test_no_specific_context_field(self):
 206         msg = self._create_test_event_message(with_sc=False)
 207         self.assertIsNone(msg.event.specific_context_field)
 208
 209     def test_get_event_payload_field(self):
 210         def event_fields_config(event):
 211             event.payload_field['giraffe'] = 1
 212             event.payload_field['gnu'] = 23
 213             event.payload_field['mosquito'] = 42
 214
 215         msg = self._create_test_event_message(
 216             event_fields_config=event_fields_config, with_ep=True
 217         )
 218
 219         self.assertEqual(msg.event.payload_field['giraffe'], 1)
 220         self.assertEqual(msg.event.payload_field['gnu'], 23)
 221         self.assertEqual(msg.event.payload_field['mosquito'], 42)
 222
 223     def test_no_payload_field(self):
 224         msg = self._create_test_event_message(with_ep=False)
 225         self.assertIsNone(msg.event.payload_field)
 226
 227     def test_clock_value(self):
 228         msg = self._create_test_event_message(with_clockclass=True)
 229         self.assertEqual(msg.default_clock_snapshot.value, 789)
 230
 231     def test_no_clock_value(self):
 232         msg = self._create_test_event_message(with_clockclass=False)
 233         with self.assertRaisesRegex(
 234             ValueError, 'stream class has no default clock class'
 235         ):
 236             msg.default_clock_snapshot
 237
 238     def test_stream(self):
 239         msg = self._create_test_event_message()
 240         self.assertEqual(msg.event.stream.addr, self.stream.addr)
 241
 242     def test_getitem(self):
 243         def event_fields_config(event):
 244             event.payload_field['giraffe'] = 1
 245             event.payload_field['gnu'] = 23
 246             event.payload_field['mosquito'] = 42
 247             event.specific_context_field['ant'] = -1
 248             event.specific_context_field['msg'] = 'hellooo'
 249             event.common_context_field['cpu_id'] = 1
 250             event.common_context_field['stuff'] = 13.194
 251
 252         def packet_fields_config(packet):
 253             packet.context_field['something'] = 154
 254             packet.context_field['something_else'] = 17.2
 255
 256         msg = self._create_test_event_message(
 257             packet_fields_config=packet_fields_config,
 258             event_fields_config=event_fields_config,
 259             with_cc=True,
 260             with_sc=True,
 261             with_ep=True,
 262             with_packet=True,
 263         )
 264         ev = msg.event
 265
 266         # Test event fields
 267         self.assertEqual(ev['giraffe'], 1)
 268         self.assertEqual(ev['gnu'], 23)
 269         self.assertEqual(ev['mosquito'], 42)
 270         self.assertEqual(ev['ant'], -1)
 271         self.assertEqual(ev['msg'], 'hellooo')
 272         self.assertEqual(ev['cpu_id'], 1)
 273         self.assertEqual(ev['stuff'], 13.194)
 274
 275         # Test packet fields
 276         self.assertEqual(ev['something'], 154)
 277         self.assertEqual(ev['something_else'], 17.2)
 278
 279         with self.assertRaises(KeyError):
 280             ev['yes']
 281
 282
 283 if __name__ == "__main__":
 284     unittest.main()