1 # SPDX-License-Identifier: GPL-2.0-only
3 # Copyright (C) 2019 EfficiOS Inc.
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
17 class TraceTestCase(unittest
.TestCase
):
19 self
._tc
= get_default_trace_class()
21 def test_create_default(self
):
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)
28 def test_create_invalid_name(self
):
29 with self
.assertRaises(TypeError):
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
)
37 def test_create_invalid_user_attributes(self
):
38 with self
.assertRaises(TypeError):
39 self
._tc
(user_attributes
=object())
41 def test_create_invalid_user_attributes_value_type(self
):
42 with self
.assertRaises(TypeError):
43 self
._tc
(user_attributes
=23)
45 def test_attr_trace_class(self
):
47 self
.assertEqual(trace
.cls
.addr
, self
._tc
.addr
)
48 self
.assertIs(type(trace
.cls
), bt2_trace_class
._TraceClass
)
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
)
54 def test_attr_name(self
):
55 trace
= self
._tc
(name
='mein trace')
56 self
.assertEqual(trace
.name
, 'mein trace')
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'))
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)
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'})
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
)
78 type(trace
.environment
['patate']), bt2_value
._SignedIntegerValueConst
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'})
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
91 def test_env_get_non_existent(self
):
92 trace
= self
._tc
(environment
={'hello': 'you', 'foo': -5})
94 with self
.assertRaises(KeyError):
95 trace
.environment
['lel']
99 sc
= self
._tc
.create_stream_class()
100 self
.assertEqual(len(trace
), 0)
102 trace
.create_stream(sc
)
103 self
.assertEqual(len(trace
), 1)
105 def _create_trace_with_some_streams(self
):
106 sc
= self
._tc
.create_stream_class(assigns_automatic_stream_id
=False)
108 trace
.create_stream(sc
, id=12)
109 trace
.create_stream(sc
, id=15)
110 trace
.create_stream(sc
, id=17)
115 trace
= self
._create
_trace
_with
_some
_streams
()
116 stream_ids
= set(trace
)
117 self
.assertEqual(stream_ids
, {12, 15, 17})
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
)
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
)
128 def test_getitem_invalid_key(self
):
129 trace
= self
._create
_trace
_with
_some
_streams
()
130 with self
.assertRaises(KeyError):
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
138 def on_trace_destruction(trace
):
139 nonlocal num_trace_destroyed_calls
140 num_trace_destroyed_calls
+= 1
142 num_trace_class_destroyed_calls
= 0
143 num_trace_destroyed_calls
= 0
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
)
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
)
154 self
.assertIs(type(td_handle1
), bt2_utils
._ListenerHandle
)
156 trace
.remove_destruction_listener(td_handle2
)
158 self
.assertEqual(num_trace_class_destroyed_calls
, 0)
159 self
.assertEqual(num_trace_destroyed_calls
, 0)
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
, 1)
173 self
.assertEqual(num_trace_class_destroyed_calls
, 0)
174 self
.assertEqual(num_trace_destroyed_calls
, 1)
178 self
.assertEqual(num_trace_class_destroyed_calls
, 1)
179 self
.assertEqual(num_trace_destroyed_calls
, 1)
181 def test_remove_destruction_listener_wrong_type(self
):
182 trace_class
= get_default_trace_class()
183 trace
= trace_class()
185 with self
.assertRaisesRegex(
186 TypeError, r
"'int' is not a '<class 'bt2.utils._ListenerHandle'>' object"
188 trace
.remove_destruction_listener(123)
190 def test_remove_destruction_listener_wrong_object(self
):
191 def on_trace_destruction(trace
):
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()
199 handle1
= trace1
.add_destruction_listener(on_trace_destruction
)
201 with self
.assertRaisesRegex(
203 r
'This trace destruction listener does not match the trace object\.',
205 trace2
.remove_destruction_listener(handle1
)
207 def test_remove_destruction_listener_twice(self
):
208 def on_trace_destruction(trace_class
):
211 trace_class
= get_default_trace_class()
212 trace
= trace_class()
213 handle
= trace
.add_destruction_listener(on_trace_destruction
)
215 trace
.remove_destruction_listener(handle
)
217 with self
.assertRaisesRegex(
218 ValueError, r
'This trace destruction listener was already removed\.'
220 trace
.remove_destruction_listener(handle
)
222 def test_raise_in_destruction_listener(self
):
223 def on_trace_destruction(trace
):
224 raise ValueError('it hurts')
226 trace_class
= get_default_trace_class()
227 trace
= trace_class()
228 trace
.add_destruction_listener(on_trace_destruction
)
233 if __name__
== '__main__':