tests/bindings/python/bt2/test_event_class.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 from utils import get_default_trace_class
   9 from bt2 import stream_class as bt2_stream_class
  10 from bt2 import event_class as bt2_event_class
  11 from bt2 import field_class as bt2_field_class
  12 from bt2 import value as bt2_value
  13 from utils import TestOutputPortMessageIterator
  14
  15
  16 def _create_const_event_class(tc, stream_class):
  17     fc1 = tc.create_structure_field_class()
  18     fc2 = tc.create_structure_field_class()
  19     event_class = stream_class.create_event_class(
  20         payload_field_class=fc1, specific_context_field_class=fc2
  21     )
  22
  23     class MyIter(bt2._UserMessageIterator):
  24         def __init__(self, config, self_port_output):
  25
  26             trace = tc()
  27             stream = trace.create_stream(stream_class)
  28             self._msgs = [
  29                 self._create_stream_beginning_message(stream),
  30                 self._create_event_message(event_class, stream),
  31             ]
  32
  33         def __next__(self):
  34             if len(self._msgs) == 0:
  35                 raise StopIteration
  36
  37             return self._msgs.pop(0)
  38
  39     class MySrc(bt2._UserSourceComponent, message_iterator_class=MyIter):
  40         def __init__(self, config, params, obj):
  41             self._add_output_port('out', params)
  42
  43     graph = bt2.Graph()
  44     src_comp = graph.add_component(MySrc, 'my_source', None)
  45     msg_iter = TestOutputPortMessageIterator(graph, src_comp.output_ports['out'])
  46
  47     # Ignore first message, stream beginning
  48     _ = next(msg_iter)
  49
  50     event_msg = next(msg_iter)
  51
  52     return event_msg.event.cls
  53
  54
  55 class EventClassTestCase(unittest.TestCase):
  56     def setUp(self):
  57         self._tc = get_default_trace_class()
  58
  59         self._context_fc = self._tc.create_structure_field_class()
  60         self._context_fc.append_member('allo', self._tc.create_string_field_class())
  61         self._context_fc.append_member(
  62             'zola', self._tc.create_signed_integer_field_class(18)
  63         )
  64
  65         self._payload_fc = self._tc.create_structure_field_class()
  66         self._payload_fc.append_member('zoom', self._tc.create_string_field_class())
  67
  68         self._stream_class = self._tc.create_stream_class(
  69             assigns_automatic_event_class_id=True
  70         )
  71
  72     def test_create_default(self):
  73         ec = self._stream_class.create_event_class()
  74
  75         self.assertIs(type(ec), bt2_event_class._EventClass)
  76         self.assertIsNone(ec.name, 'my_event')
  77         self.assertTrue(type(ec.id), int)
  78         self.assertIsNone(ec.specific_context_field_class)
  79         self.assertIsNone(ec.payload_field_class)
  80         self.assertIsNone(ec.emf_uri)
  81         self.assertIsNone(ec.log_level)
  82         self.assertEqual(len(ec.user_attributes), 0)
  83
  84     def test_create_invalid_id(self):
  85         sc = self._tc.create_stream_class(assigns_automatic_event_class_id=False)
  86         with self.assertRaises(TypeError):
  87             sc.create_event_class(id='lel')
  88
  89         self.assertEqual(len(sc), 0)
  90
  91     def test_create_specific_context_field_class(self):
  92         fc = self._tc.create_structure_field_class()
  93         ec = self._stream_class.create_event_class(specific_context_field_class=fc)
  94         self.assertEqual(ec.specific_context_field_class.addr, fc.addr)
  95         self.assertIs(
  96             type(ec.specific_context_field_class), bt2_field_class._StructureFieldClass
  97         )
  98
  99     def test_const_create_specific_context_field_class(self):
 100         ec_const = _create_const_event_class(self._tc, self._stream_class)
 101         self.assertIs(
 102             type(ec_const.specific_context_field_class),
 103             bt2_field_class._StructureFieldClassConst,
 104         )
 105
 106     def test_create_invalid_specific_context_field_class(self):
 107         with self.assertRaises(TypeError):
 108             self._stream_class.create_event_class(specific_context_field_class='lel')
 109
 110         self.assertEqual(len(self._stream_class), 0)
 111
 112     def test_create_payload_field_class(self):
 113         fc = self._tc.create_structure_field_class()
 114         ec = self._stream_class.create_event_class(payload_field_class=fc)
 115         self.assertEqual(ec.payload_field_class.addr, fc.addr)
 116         self.assertIs(
 117             type(ec.payload_field_class), bt2_field_class._StructureFieldClass
 118         )
 119
 120     def test_const_create_payload_field_class(self):
 121         ec_const = _create_const_event_class(self._tc, self._stream_class)
 122         self.assertIs(
 123             type(ec_const.payload_field_class),
 124             bt2_field_class._StructureFieldClassConst,
 125         )
 126
 127     def test_create_invalid_payload_field_class(self):
 128         with self.assertRaises(TypeError):
 129             self._stream_class.create_event_class(payload_field_class='lel')
 130
 131         self.assertEqual(len(self._stream_class), 0)
 132
 133     def test_create_name(self):
 134         ec = self._stream_class.create_event_class(name='viande à chien')
 135         self.assertEqual(ec.name, 'viande à chien')
 136
 137     def test_create_invalid_name(self):
 138         with self.assertRaises(TypeError):
 139             self._stream_class.create_event_class(name=2)
 140
 141         self.assertEqual(len(self._stream_class), 0)
 142
 143     def test_emf_uri(self):
 144         ec = self._stream_class.create_event_class(emf_uri='salut')
 145         self.assertEqual(ec.emf_uri, 'salut')
 146
 147     def test_create_invalid_emf_uri(self):
 148         with self.assertRaises(TypeError):
 149             self._stream_class.create_event_class(emf_uri=23)
 150
 151         self.assertEqual(len(self._stream_class), 0)
 152
 153     def test_create_log_level(self):
 154         ec = self._stream_class.create_event_class(
 155             log_level=bt2.EventClassLogLevel.EMERGENCY
 156         )
 157         self.assertEqual(ec.log_level, bt2.EventClassLogLevel.EMERGENCY)
 158
 159     def test_create_invalid_log_level(self):
 160         with self.assertRaises(ValueError):
 161             self._stream_class.create_event_class(log_level='zoom')
 162
 163         self.assertEqual(len(self._stream_class), 0)
 164
 165     def test_create_user_attributes(self):
 166         ec = self._stream_class.create_event_class(user_attributes={'salut': 23})
 167         self.assertEqual(ec.user_attributes, {'salut': 23})
 168         self.assertIs(type(ec.user_attributes), bt2_value.MapValue)
 169
 170     def test_const_create_user_attributes(self):
 171         ec_const = _create_const_event_class(self._tc, self._stream_class)
 172         self.assertIs(type(ec_const.user_attributes), bt2_value._MapValueConst)
 173
 174     def test_create_invalid_user_attributes(self):
 175         with self.assertRaises(TypeError):
 176             self._stream_class.create_event_class(user_attributes=object())
 177
 178         self.assertEqual(len(self._stream_class), 0)
 179
 180     def test_create_invalid_user_attributes_value_type(self):
 181         with self.assertRaises(TypeError):
 182             self._stream_class.create_event_class(user_attributes=23)
 183
 184         self.assertEqual(len(self._stream_class), 0)
 185
 186     def test_stream_class(self):
 187         ec = self._stream_class.create_event_class()
 188         self.assertEqual(ec.stream_class.addr, self._stream_class.addr)
 189         self.assertIs(type(ec.stream_class), bt2_stream_class._StreamClass)
 190
 191     def test_const_stream_class(self):
 192         ec_const = _create_const_event_class(self._tc, self._stream_class)
 193         self.assertIs(type(ec_const.stream_class), bt2_stream_class._StreamClassConst)
 194
 195
 196 if __name__ == '__main__':
 197     unittest.main()