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