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 type_of_passed_trace
141 type_of_passed_trace
= type(trace
)
143 nonlocal num_trace_destroyed_calls
144 num_trace_destroyed_calls
+= 1
146 num_trace_class_destroyed_calls
= 0
147 num_trace_destroyed_calls
= 0
148 type_of_passed_trace
= None
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
)
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
)
159 self
.assertIs(type(td_handle1
), bt2_utils
._ListenerHandle
)
161 trace
.remove_destruction_listener(td_handle2
)
163 self
.assertEqual(num_trace_class_destroyed_calls
, 0)
164 self
.assertEqual(num_trace_destroyed_calls
, 0)
168 self
.assertEqual(num_trace_class_destroyed_calls
, 0)
169 self
.assertEqual(num_trace_destroyed_calls
, 0)
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
)
179 self
.assertEqual(num_trace_class_destroyed_calls
, 0)
180 self
.assertEqual(num_trace_destroyed_calls
, 1)
184 self
.assertEqual(num_trace_class_destroyed_calls
, 1)
185 self
.assertEqual(num_trace_destroyed_calls
, 1)
187 def test_remove_destruction_listener_wrong_type(self
):
188 trace_class
= get_default_trace_class()
189 trace
= trace_class()
191 with self
.assertRaisesRegex(
192 TypeError, r
"'int' is not a '<class 'bt2.utils._ListenerHandle'>' object"
194 trace
.remove_destruction_listener(123)
196 def test_remove_destruction_listener_wrong_object(self
):
197 def on_trace_destruction(trace
):
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()
205 handle1
= trace1
.add_destruction_listener(on_trace_destruction
)
207 with self
.assertRaisesRegex(
209 r
"This trace destruction listener does not match the trace object\.",
211 trace2
.remove_destruction_listener(handle1
)
213 def test_remove_destruction_listener_twice(self
):
214 def on_trace_destruction(trace_class
):
217 trace_class
= get_default_trace_class()
218 trace
= trace_class()
219 handle
= trace
.add_destruction_listener(on_trace_destruction
)
221 trace
.remove_destruction_listener(handle
)
223 with self
.assertRaisesRegex(
224 ValueError, r
"This trace destruction listener was already removed\."
226 trace
.remove_destruction_listener(handle
)
228 def test_raise_in_destruction_listener(self
):
229 def on_trace_destruction(trace
):
230 raise ValueError("it hurts")
232 trace_class
= get_default_trace_class()
233 trace
= trace_class()
234 trace
.add_destruction_listener(on_trace_destruction
)
239 if __name__
== "__main__":