tests/bindings/python/bt2/test_trace.py

   1 # SPDX-License-Identifier: GPL-2.0-only
   2 #
   3 # Copyright (C) 2019 EfficiOS Inc.
   4 #
   5
   6 import uuid
   7 import unittest
   8
   9 import utils
  10 from bt2 import trace as bt2_trace
  11 from bt2 import utils as bt2_utils
  12 from bt2 import value as bt2_value
  13 from bt2 import stream as bt2_stream
  14 from bt2 import trace_class as bt2_trace_class
  15 from utils import get_default_trace_class
  16
  17
  18 class TraceTestCase(unittest.TestCase):
  19     def setUp(self):
  20         self._tc = get_default_trace_class()
  21
  22     def test_create_default(self):
  23         trace = self._tc()
  24         self.assertIsNone(trace.name)
  25         self.assertIsNone(trace.uuid)
  26         self.assertEqual(len(trace.environment), 0)
  27         self.assertEqual(len(trace.user_attributes), 0)
  28
  29     def test_create_invalid_name(self):
  30         with self.assertRaises(TypeError):
  31             self._tc(name=17)
  32
  33     def test_create_user_attributes(self):
  34         trace = self._tc(user_attributes={"salut": 23})
  35         self.assertEqual(trace.user_attributes, {"salut": 23})
  36         self.assertIs(type(trace.user_attributes), bt2_value.MapValue)
  37
  38     def test_create_invalid_user_attributes(self):
  39         with self.assertRaises(TypeError):
  40             self._tc(user_attributes=object())
  41
  42     def test_create_invalid_user_attributes_value_type(self):
  43         with self.assertRaises(TypeError):
  44             self._tc(user_attributes=23)
  45
  46     def test_attr_trace_class(self):
  47         trace = self._tc()
  48         self.assertEqual(trace.cls.addr, self._tc.addr)
  49         self.assertIs(type(trace.cls), bt2_trace_class._TraceClass)
  50
  51     def test_const_attr_trace_class(self):
  52         trace = utils.get_const_stream_beginning_message().stream.trace
  53         self.assertIs(type(trace.cls), bt2_trace_class._TraceClassConst)
  54
  55     def test_attr_name(self):
  56         trace = self._tc(name="mein trace")
  57         self.assertEqual(trace.name, "mein trace")
  58
  59     def test_attr_uuid(self):
  60         trace = self._tc(uuid=uuid.UUID("da7d6b6f-3108-4706-89bd-ab554732611b"))
  61         self.assertEqual(trace.uuid, uuid.UUID("da7d6b6f-3108-4706-89bd-ab554732611b"))
  62
  63     def test_env_get(self):
  64         trace = self._tc(environment={"hello": "you", "foo": -5})
  65         self.assertIs(type(trace.environment), bt2_trace._TraceEnvironment)
  66         self.assertIs(type(trace.environment["foo"]), bt2_value.SignedIntegerValue)
  67         self.assertEqual(trace.environment["hello"], "you")
  68         self.assertEqual(trace.environment["foo"], -5)
  69
  70     def test_env_iter(self):
  71         trace = self._tc(environment={"hello": "you", "foo": -5})
  72         values = set(trace.environment)
  73         self.assertEqual(values, {"hello", "foo"})
  74
  75     def test_const_env_get(self):
  76         trace = utils.get_const_stream_beginning_message().stream.trace
  77         self.assertIs(type(trace.environment), bt2_trace._TraceEnvironmentConst)
  78         self.assertIs(
  79             type(trace.environment["patate"]), bt2_value._SignedIntegerValueConst
  80         )
  81
  82     def test_const_env_iter(self):
  83         trace = utils.get_const_stream_beginning_message().stream.trace
  84         values = set(trace.environment)
  85         self.assertEqual(values, {"patate"})
  86
  87     def test_const_env_set(self):
  88         trace = utils.get_const_stream_beginning_message().stream.trace
  89         with self.assertRaises(TypeError):
  90             trace.environment["patate"] = 33
  91
  92     def test_env_get_non_existent(self):
  93         trace = self._tc(environment={"hello": "you", "foo": -5})
  94
  95         with self.assertRaises(KeyError):
  96             trace.environment["lel"]
  97
  98     def test_len(self):
  99         trace = self._tc()
 100         sc = self._tc.create_stream_class()
 101         self.assertEqual(len(trace), 0)
 102
 103         trace.create_stream(sc)
 104         self.assertEqual(len(trace), 1)
 105
 106     def _create_trace_with_some_streams(self):
 107         sc = self._tc.create_stream_class(assigns_automatic_stream_id=False)
 108         trace = self._tc()
 109         trace.create_stream(sc, id=12)
 110         trace.create_stream(sc, id=15)
 111         trace.create_stream(sc, id=17)
 112
 113         return trace
 114
 115     def test_iter(self):
 116         trace = self._create_trace_with_some_streams()
 117         stream_ids = set(trace)
 118         self.assertEqual(stream_ids, {12, 15, 17})
 119
 120     def test_getitem(self):
 121         trace = self._create_trace_with_some_streams()
 122         self.assertEqual(trace[12].id, 12)
 123         self.assertIs(type(trace[12]), bt2_stream._Stream)
 124
 125     def test_const_getitem(self):
 126         trace = utils.get_const_stream_beginning_message().stream.trace
 127         self.assertIs(type(trace[0]), bt2_stream._StreamConst)
 128
 129     def test_getitem_invalid_key(self):
 130         trace = self._create_trace_with_some_streams()
 131         with self.assertRaises(KeyError):
 132             trace[18]
 133
 134     def test_destruction_listener(self):
 135         def on_trace_class_destruction(trace_class):
 136             nonlocal num_trace_class_destroyed_calls
 137             num_trace_class_destroyed_calls += 1
 138
 139         def on_trace_destruction(trace):
 140             nonlocal type_of_passed_trace
 141             type_of_passed_trace = type(trace)
 142
 143             nonlocal num_trace_destroyed_calls
 144             num_trace_destroyed_calls += 1
 145
 146         num_trace_class_destroyed_calls = 0
 147         num_trace_destroyed_calls = 0
 148         type_of_passed_trace = None
 149
 150         trace_class = get_default_trace_class()
 151         stream_class = trace_class.create_stream_class()
 152         trace = trace_class()
 153         stream = trace.create_stream(stream_class)
 154
 155         trace_class.add_destruction_listener(on_trace_class_destruction)
 156         td_handle1 = trace.add_destruction_listener(on_trace_destruction)
 157         td_handle2 = trace.add_destruction_listener(on_trace_destruction)
 158
 159         self.assertIs(type(td_handle1), bt2_utils._ListenerHandle)
 160
 161         trace.remove_destruction_listener(td_handle2)
 162
 163         self.assertEqual(num_trace_class_destroyed_calls, 0)
 164         self.assertEqual(num_trace_destroyed_calls, 0)
 165
 166         del trace
 167
 168         self.assertEqual(num_trace_class_destroyed_calls, 0)
 169         self.assertEqual(num_trace_destroyed_calls, 0)
 170
 171         del stream
 172
 173         self.assertEqual(num_trace_class_destroyed_calls, 0)
 174         self.assertEqual(num_trace_destroyed_calls, 1)
 175         self.assertIs(type_of_passed_trace, bt2_trace._TraceConst)
 176
 177         del trace_class
 178
 179         self.assertEqual(num_trace_class_destroyed_calls, 0)
 180         self.assertEqual(num_trace_destroyed_calls, 1)
 181
 182         del stream_class
 183
 184         self.assertEqual(num_trace_class_destroyed_calls, 1)
 185         self.assertEqual(num_trace_destroyed_calls, 1)
 186
 187     def test_remove_destruction_listener_wrong_type(self):
 188         trace_class = get_default_trace_class()
 189         trace = trace_class()
 190
 191         with self.assertRaisesRegex(
 192             TypeError, r"'int' is not a '<class 'bt2.utils._ListenerHandle'>' object"
 193         ):
 194             trace.remove_destruction_listener(123)
 195
 196     def test_remove_destruction_listener_wrong_object(self):
 197         def on_trace_destruction(trace):
 198             pass
 199
 200         trace_class_1 = get_default_trace_class()
 201         trace1 = trace_class_1()
 202         trace_class_2 = get_default_trace_class()
 203         trace2 = trace_class_2()
 204
 205         handle1 = trace1.add_destruction_listener(on_trace_destruction)
 206
 207         with self.assertRaisesRegex(
 208             ValueError,
 209             r"This trace destruction listener does not match the trace object\.",
 210         ):
 211             trace2.remove_destruction_listener(handle1)
 212
 213     def test_remove_destruction_listener_twice(self):
 214         def on_trace_destruction(trace_class):
 215             pass
 216
 217         trace_class = get_default_trace_class()
 218         trace = trace_class()
 219         handle = trace.add_destruction_listener(on_trace_destruction)
 220
 221         trace.remove_destruction_listener(handle)
 222
 223         with self.assertRaisesRegex(
 224             ValueError, r"This trace destruction listener was already removed\."
 225         ):
 226             trace.remove_destruction_listener(handle)
 227
 228     def test_raise_in_destruction_listener(self):
 229         def on_trace_destruction(trace):
 230             raise ValueError("it hurts")
 231
 232         trace_class = get_default_trace_class()
 233         trace = trace_class()
 234         trace.add_destruction_listener(on_trace_destruction)
 235
 236         del trace
 237
 238
 239 if __name__ == "__main__":
 240     unittest.main()