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