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