1 # SPDX-License-Identifier: GPL-2.0-only
3 # Copyright (C) 2019 EfficiOS Inc.
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
18 class TraceTestCase(unittest
.TestCase
):
20 self
._tc
= get_default_trace_class()
22 def test_create_default(self
):
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)
29 def test_create_invalid_name(self
):
30 with self
.assertRaises(TypeError):
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
)
38 def test_create_invalid_user_attributes(self
):
39 with self
.assertRaises(TypeError):
40 self
._tc
(user_attributes
=object())
42 def test_create_invalid_user_attributes_value_type(self
):
43 with self
.assertRaises(TypeError):
44 self
._tc
(user_attributes
=23)
46 def test_attr_trace_class(self
):
48 self
.assertEqual(trace
.cls
.addr
, self
._tc
.addr
)
49 self
.assertIs(type(trace
.cls
), bt2_trace_class
._TraceClass
)
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
)
55 def test_attr_name(self
):
56 trace
= self
._tc
(name
="mein trace")
57 self
.assertEqual(trace
.name
, "mein trace")
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"))
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)
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"})
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
)
79 type(trace
.environment
["patate"]), bt2_value
._SignedIntegerValueConst
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"})
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
92 def test_env_get_non_existent(self
):
93 trace
= self
._tc
(environment
={"hello": "you", "foo": -5})
95 with self
.assertRaises(KeyError):
96 trace
.environment
["lel"]
100 sc
= self
._tc
.create_stream_class()
101 self
.assertEqual(len(trace
), 0)
103 trace
.create_stream(sc
)
104 self
.assertEqual(len(trace
), 1)
106 def _create_trace_with_some_streams(self
):
107 sc
= self
._tc
.create_stream_class(assigns_automatic_stream_id
=False)
109 trace
.create_stream(sc
, id=12)
110 trace
.create_stream(sc
, id=15)
111 trace
.create_stream(sc
, id=17)
116 trace
= self
._create
_trace
_with
_some
_streams
()
117 stream_ids
= set(trace
)
118 self
.assertEqual(stream_ids
, {12, 15, 17})
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
)
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
)
129 def test_getitem_invalid_key(self
):
130 trace
= self
._create
_trace
_with
_some
_streams
()
131 with self
.assertRaises(KeyError):
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
139 def on_trace_destruction(trace
):
140 nonlocal num_trace_destroyed_calls
141 num_trace_destroyed_calls
+= 1
143 num_trace_class_destroyed_calls
= 0
144 num_trace_destroyed_calls
= 0
146 trace_class
= get_default_trace_class()
147 stream_class
= trace_class
.create_stream_class()
148 trace
= trace_class()
149 stream
= trace
.create_stream(stream_class
)
151 trace_class
.add_destruction_listener(on_trace_class_destruction
)
152 td_handle1
= trace
.add_destruction_listener(on_trace_destruction
)
153 td_handle2
= trace
.add_destruction_listener(on_trace_destruction
)
155 self
.assertIs(type(td_handle1
), bt2_utils
._ListenerHandle
)
157 trace
.remove_destruction_listener(td_handle2
)
159 self
.assertEqual(num_trace_class_destroyed_calls
, 0)
160 self
.assertEqual(num_trace_destroyed_calls
, 0)
164 self
.assertEqual(num_trace_class_destroyed_calls
, 0)
165 self
.assertEqual(num_trace_destroyed_calls
, 0)
169 self
.assertEqual(num_trace_class_destroyed_calls
, 0)
170 self
.assertEqual(num_trace_destroyed_calls
, 1)
174 self
.assertEqual(num_trace_class_destroyed_calls
, 0)
175 self
.assertEqual(num_trace_destroyed_calls
, 1)
179 self
.assertEqual(num_trace_class_destroyed_calls
, 1)
180 self
.assertEqual(num_trace_destroyed_calls
, 1)
182 def test_remove_destruction_listener_wrong_type(self
):
183 trace_class
= get_default_trace_class()
184 trace
= trace_class()
186 with self
.assertRaisesRegex(
187 TypeError, r
"'int' is not a '<class 'bt2.utils._ListenerHandle'>' object"
189 trace
.remove_destruction_listener(123)
191 def test_remove_destruction_listener_wrong_object(self
):
192 def on_trace_destruction(trace
):
195 trace_class_1
= get_default_trace_class()
196 trace1
= trace_class_1()
197 trace_class_2
= get_default_trace_class()
198 trace2
= trace_class_2()
200 handle1
= trace1
.add_destruction_listener(on_trace_destruction
)
202 with self
.assertRaisesRegex(
204 r
"This trace destruction listener does not match the trace object\.",
206 trace2
.remove_destruction_listener(handle1
)
208 def test_remove_destruction_listener_twice(self
):
209 def on_trace_destruction(trace_class
):
212 trace_class
= get_default_trace_class()
213 trace
= trace_class()
214 handle
= trace
.add_destruction_listener(on_trace_destruction
)
216 trace
.remove_destruction_listener(handle
)
218 with self
.assertRaisesRegex(
219 ValueError, r
"This trace destruction listener was already removed\."
221 trace
.remove_destruction_listener(handle
)
223 def test_raise_in_destruction_listener(self
):
224 def on_trace_destruction(trace
):
225 raise ValueError("it hurts")
227 trace_class
= get_default_trace_class()
228 trace
= trace_class()
229 trace
.add_destruction_listener(on_trace_destruction
)
234 if __name__
== "__main__":