tests/bindings/python/bt2/test_event.py

   1 from collections import OrderedDict
   2 import unittest
   3 import bt2
   4
   5
   6 class EventTestCase(unittest.TestCase):
   7     def _create_test_event_message(self, packet_fields_config=None,
   8                                    event_fields_config=None,
   9                                    with_clockclass=False,
  10                                    with_cc=False, with_sc=False,
  11                                    with_ep=False):
  12
  13         class MyIter(bt2._UserMessageIterator):
  14             def __init__(self, self_output_port):
  15                 self._at = 0
  16
  17             def __next__(self):
  18                 if self._at == 0:
  19                     msg = self._create_stream_beginning_message(test_obj.stream)
  20                 elif self._at == 1:
  21                     assert test_obj.packet
  22                     msg = self._create_packet_beginning_message(test_obj.packet)
  23                 elif self._at == 2:
  24                     default_clock_snapshot = 789 if with_clockclass else None
  25                     assert test_obj.packet
  26                     msg = self._create_event_message(test_obj.event_class, test_obj.packet, default_clock_snapshot)
  27                     if event_fields_config is not None:
  28                         event_fields_config(msg.event)
  29                 elif self._at == 3:
  30                     msg = self._create_packet_end_message(test_obj.packet)
  31                 elif self._at == 4:
  32                     msg = self._create_stream_end_message(test_obj.stream)
  33                 elif self._at >= 5:
  34                     raise bt2.Stop
  35
  36                 self._at += 1
  37                 return msg
  38
  39         class MySrc(bt2._UserSourceComponent, message_iterator_class=MyIter):
  40             def __init__(self, params):
  41                 self._add_output_port('out')
  42                 tc = self._create_trace_class()
  43
  44                 clock_class = None
  45                 if with_clockclass:
  46                     clock_class = self._create_clock_class(frequency=1000)
  47
  48                 # event common context (stream-class-defined)
  49                 cc = None
  50                 if with_cc:
  51                     cc = tc.create_structure_field_class()
  52                     cc += OrderedDict((
  53                         ('cpu_id', tc.create_signed_integer_field_class(8)),
  54                         ('stuff', tc.create_real_field_class()),
  55                     ))
  56
  57                 # packet context (stream-class-defined)
  58                 pc = tc.create_structure_field_class()
  59                 pc += OrderedDict((
  60                     ('something', tc.create_unsigned_integer_field_class(8)),
  61                     ('something_else', tc.create_real_field_class()),
  62                 ))
  63
  64                 stream_class = tc.create_stream_class(default_clock_class=clock_class,
  65                                                       event_common_context_field_class=cc,
  66                                                       packet_context_field_class=pc)
  67
  68                 # specific context (event-class-defined)
  69                 sc = None
  70                 if with_sc:
  71                     sc = tc.create_structure_field_class()
  72                     sc += OrderedDict((
  73                         ('ant', tc.create_signed_integer_field_class(16)),
  74                         ('msg', tc.create_string_field_class()),
  75                     ))
  76
  77                 # event payload
  78                 ep = None
  79                 if with_ep:
  80                     ep = tc.create_structure_field_class()
  81                     ep += OrderedDict((
  82                         ('giraffe', tc.create_signed_integer_field_class(32)),
  83                         ('gnu', tc.create_signed_integer_field_class(8)),
  84                         ('mosquito', tc.create_signed_integer_field_class(8)),
  85                     ))
  86
  87                 event_class = stream_class.create_event_class(name='garou',
  88                                                               specific_context_field_class=sc,
  89                                                               payload_field_class=ep)
  90
  91                 trace = tc()
  92                 stream = trace.create_stream(stream_class)
  93                 packet = stream.create_packet()
  94
  95                 if packet_fields_config is not None:
  96                     packet_fields_config(packet)
  97
  98                 test_obj.packet = packet
  99                 test_obj.stream = stream
 100                 test_obj.event_class = event_class
 101
 102         test_obj = self
 103         self._graph = bt2.Graph()
 104         self._src_comp = self._graph.add_component(MySrc, 'my_source')
 105         self._msg_iter = self._graph.create_output_port_message_iterator(self._src_comp.output_ports['out'])
 106
 107         for i, msg in enumerate(self._msg_iter):
 108             if i == 2:
 109                 return msg
 110
 111     def test_attr_event_class(self):
 112         msg = self._create_test_event_message()
 113         self.assertEqual(msg.event.event_class.addr, self.event_class.addr)
 114
 115     def test_attr_name(self):
 116         msg = self._create_test_event_message()
 117         self.assertEqual(msg.event.name, self.event_class.name)
 118
 119     def test_attr_id(self):
 120         msg = self._create_test_event_message()
 121         self.assertEqual(msg.event.id, self.event_class.id)
 122
 123     def test_get_common_context_field(self):
 124         def event_fields_config(event):
 125             event.common_context_field['cpu_id'] = 1
 126             event.common_context_field['stuff'] = 13.194
 127
 128         msg = self._create_test_event_message(event_fields_config=event_fields_config, with_cc=True)
 129
 130         self.assertEqual(msg.event.common_context_field['cpu_id'], 1)
 131         self.assertEqual(msg.event.common_context_field['stuff'], 13.194)
 132
 133     def test_no_common_context_field(self):
 134         msg = self._create_test_event_message(with_cc=False)
 135         self.assertIsNone(msg.event.common_context_field)
 136
 137     def test_get_specific_context_field(self):
 138         def event_fields_config(event):
 139             event.specific_context_field['ant'] = -1
 140             event.specific_context_field['msg'] = 'hellooo'
 141
 142         msg = self._create_test_event_message(event_fields_config=event_fields_config, with_sc=True)
 143
 144         self.assertEqual(msg.event.specific_context_field['ant'], -1)
 145         self.assertEqual(msg.event.specific_context_field['msg'], 'hellooo')
 146
 147     def test_no_specific_context_field(self):
 148         msg = self._create_test_event_message(with_sc=False)
 149         self.assertIsNone(msg.event.specific_context_field)
 150
 151     def test_get_event_payload_field(self):
 152         def event_fields_config(event):
 153             event.payload_field['giraffe'] = 1
 154             event.payload_field['gnu'] = 23
 155             event.payload_field['mosquito'] = 42
 156
 157         msg = self._create_test_event_message(event_fields_config=event_fields_config, with_ep=True)
 158
 159         self.assertEqual(msg.event.payload_field['giraffe'], 1)
 160         self.assertEqual(msg.event.payload_field['gnu'], 23)
 161         self.assertEqual(msg.event.payload_field['mosquito'], 42)
 162
 163     def test_no_payload_field(self):
 164         msg = self._create_test_event_message(with_ep=False)
 165         self.assertIsNone(msg.event.payload_field)
 166
 167     def test_clock_value(self):
 168         msg = self._create_test_event_message(with_clockclass=True)
 169         self.assertEqual(msg.default_clock_snapshot.value, 789)
 170
 171     def test_no_clock_value(self):
 172         msg = self._create_test_event_message(with_clockclass=False)
 173         with self.assertRaises(bt2.NonexistentClockSnapshot):
 174             msg.default_clock_snapshot
 175
 176     def test_stream(self):
 177         msg = self._create_test_event_message()
 178         self.assertEqual(msg.event.stream.addr, self.stream.addr)
 179
 180     def test_getitem(self):
 181         def event_fields_config(event):
 182             event.payload_field['giraffe'] = 1
 183             event.payload_field['gnu'] = 23
 184             event.payload_field['mosquito'] = 42
 185             event.specific_context_field['ant'] = -1
 186             event.specific_context_field['msg'] = 'hellooo'
 187             event.common_context_field['cpu_id'] = 1
 188             event.common_context_field['stuff'] = 13.194
 189
 190         def packet_fields_config(packet):
 191             packet.context_field['something'] = 154
 192             packet.context_field['something_else'] = 17.2
 193
 194         msg = self._create_test_event_message(packet_fields_config=packet_fields_config,
 195                                               event_fields_config=event_fields_config,
 196                                               with_cc=True, with_sc=True, with_ep=True)
 197         ev = msg.event
 198
 199         # Test event fields
 200         self.assertEqual(ev['giraffe'], 1)
 201         self.assertEqual(ev['gnu'], 23)
 202         self.assertEqual(ev['mosquito'], 42)
 203         self.assertEqual(ev['ant'], -1)
 204         self.assertEqual(ev['msg'], 'hellooo')
 205         self.assertEqual(ev['cpu_id'], 1)
 206         self.assertEqual(ev['stuff'], 13.194)
 207
 208         # Test packet fields
 209         self.assertEqual(ev['something'], 154)
 210         self.assertEqual(ev['something_else'], 17.2)
 211
 212         with self.assertRaises(KeyError):
 213             ev['yes']
 214
 215
 216 if __name__ == "__main__":
 217     unittest.main()