1 # SPDX-License-Identifier: GPL-2.0-only
3 # Copyright (C) 2019 EfficiOS Inc.
8 from bt2
import utils
as bt2_utils
9 from bt2
import trace_class
as bt2_trace_class
10 from bt2
import stream_class
as bt2_stream_class
12 run_in_component_init
,
13 get_default_trace_class
,
14 get_const_stream_beginning_message
,
18 class TraceClassTestCase(unittest
.TestCase
):
19 def assertRaisesInComponentInit(self
, expected_exc_type
, user_code
):
23 except Exception as exc
:
26 exc_type
= run_in_component_init(f
)
27 self
.assertIsNotNone(exc_type
)
28 self
.assertEqual(exc_type
, expected_exc_type
)
30 def test_create_default(self
):
32 return comp_self
._create
_trace
_class
()
34 tc
= run_in_component_init(f
)
36 self
.assertEqual(len(tc
), 0)
37 self
.assertIs(type(tc
), bt2_trace_class
._TraceClass
)
38 self
.assertTrue(tc
.assigns_automatic_stream_class_id
)
39 self
.assertEqual(len(tc
.user_attributes
), 0)
41 def test_create_user_attributes(self
):
43 return comp_self
._create
_trace
_class
(user_attributes
={"salut": 23})
45 tc
= run_in_component_init(f
)
46 self
.assertEqual(tc
.user_attributes
, {"salut": 23})
48 def test_create_invalid_user_attributes(self
):
50 return comp_self
._create
_trace
_class
(user_attributes
=object())
52 self
.assertRaisesInComponentInit(TypeError, f
)
54 def test_create_invalid_user_attributes_value_type(self
):
56 return comp_self
._create
_trace
_class
(user_attributes
=23)
58 self
.assertRaisesInComponentInit(TypeError, f
)
60 def test_create_invalid_automatic_stream_class_id_type(self
):
62 return comp_self
._create
_trace
_class
(
63 assigns_automatic_stream_class_id
="perchaude"
66 self
.assertRaisesInComponentInit(TypeError, f
)
68 def test_automatic_stream_class_id(self
):
70 return comp_self
._create
_trace
_class
(assigns_automatic_stream_class_id
=True)
72 tc
= run_in_component_init(f
)
73 self
.assertTrue(tc
.assigns_automatic_stream_class_id
)
75 # This should not throw.
76 sc1
= tc
.create_stream_class()
77 sc2
= tc
.create_stream_class()
79 self
.assertIs(type(sc1
), bt2_stream_class
._StreamClass
)
80 self
.assertIs(type(sc2
), bt2_stream_class
._StreamClass
)
81 self
.assertNotEqual(sc1
.id, sc2
.id)
83 def test_automatic_stream_class_id_raises(self
):
85 return comp_self
._create
_trace
_class
(assigns_automatic_stream_class_id
=True)
87 tc
= run_in_component_init(f
)
88 self
.assertTrue(tc
.assigns_automatic_stream_class_id
)
90 with self
.assertRaises(ValueError):
91 tc
.create_stream_class(23)
93 def test_no_assigns_automatic_stream_class_id(self
):
95 return comp_self
._create
_trace
_class
(
96 assigns_automatic_stream_class_id
=False
99 tc
= run_in_component_init(f
)
100 self
.assertFalse(tc
.assigns_automatic_stream_class_id
)
102 sc
= tc
.create_stream_class(id=28)
103 self
.assertEqual(sc
.id, 28)
105 def test_no_assigns_automatic_stream_class_id_raises(self
):
107 return comp_self
._create
_trace
_class
(
108 assigns_automatic_stream_class_id
=False
111 tc
= run_in_component_init(f
)
112 self
.assertFalse(tc
.assigns_automatic_stream_class_id
)
114 # In this mode, it is required to pass an explicit id.
115 with self
.assertRaises(ValueError):
116 tc
.create_stream_class()
119 def _create_trace_class_with_some_stream_classes():
121 return comp_self
._create
_trace
_class
(
122 assigns_automatic_stream_class_id
=False
125 tc
= run_in_component_init(f
)
126 sc1
= tc
.create_stream_class(id=12)
127 sc2
= tc
.create_stream_class(id=54)
128 sc3
= tc
.create_stream_class(id=2018)
129 return tc
, sc1
, sc2
, sc3
131 def test_getitem(self
):
132 tc
, _
, _
, sc3
= self
._create
_trace
_class
_with
_some
_stream
_classes
()
133 self
.assertIs(type(tc
[2018]), bt2_stream_class
._StreamClass
)
134 self
.assertEqual(tc
[2018].addr
, sc3
.addr
)
136 def test_const_getitem(self
):
137 const_tc
= get_const_stream_beginning_message().stream
.trace
.cls
138 self
.assertIs(type(const_tc
[0]), bt2_stream_class
._StreamClassConst
)
140 def test_getitem_wrong_key_type(self
):
141 tc
, _
, _
, _
= self
._create
_trace
_class
_with
_some
_stream
_classes
()
142 with self
.assertRaises(TypeError):
145 def test_getitem_wrong_key(self
):
146 tc
, _
, _
, _
= self
._create
_trace
_class
_with
_some
_stream
_classes
()
147 with self
.assertRaises(KeyError):
151 tc
= get_default_trace_class()
152 self
.assertEqual(len(tc
), 0)
153 tc
.create_stream_class()
154 self
.assertEqual(len(tc
), 1)
157 tc
, sc1
, sc2
, sc3
= self
._create
_trace
_class
_with
_some
_stream
_classes
()
159 for sc_id
, stream_class
in tc
.items():
161 self
.assertIs(type(stream_class
), bt2_stream_class
._StreamClass
)
162 self
.assertEqual(stream_class
.addr
, sc1
.addr
)
164 self
.assertEqual(stream_class
.addr
, sc2
.addr
)
166 self
.assertEqual(stream_class
.addr
, sc3
.addr
)
168 def test_const_iter(self
):
169 const_tc
= get_const_stream_beginning_message().stream
.trace
.cls
170 const_sc
= list(const_tc
.values())[0]
171 self
.assertIs(type(const_sc
), bt2_stream_class
._StreamClassConst
)
173 def test_destruction_listener(self
):
174 def on_trace_class_destruction(trace_class
):
175 nonlocal type_of_passed_trace_class
176 type_of_passed_trace_class
= type(trace_class
)
178 nonlocal num_destruct_calls
179 num_destruct_calls
+= 1
181 type_of_passed_trace_class
= None
182 num_destruct_calls
= 0
184 trace_class
= get_default_trace_class()
186 handle1
= trace_class
.add_destruction_listener(on_trace_class_destruction
)
187 self
.assertIs(type(handle1
), bt2_utils
._ListenerHandle
)
189 handle2
= trace_class
.add_destruction_listener(on_trace_class_destruction
)
191 trace_class
.remove_destruction_listener(handle2
)
193 self
.assertEqual(num_destruct_calls
, 0)
197 self
.assertEqual(num_destruct_calls
, 1)
198 self
.assertIs(type_of_passed_trace_class
, bt2_trace_class
._TraceClassConst
)
200 def test_remove_destruction_listener_wrong_type(self
):
201 trace_class
= get_default_trace_class()
203 with self
.assertRaisesRegex(
204 TypeError, r
"'int' is not a '<class 'bt2.utils._ListenerHandle'>' object"
206 trace_class
.remove_destruction_listener(123)
208 def test_remove_destruction_listener_wrong_object(self
):
209 def on_trace_class_destruction(trace_class
):
212 trace_class_1
= get_default_trace_class()
213 trace_class_2
= get_default_trace_class()
215 handle1
= trace_class_1
.add_destruction_listener(on_trace_class_destruction
)
217 with self
.assertRaisesRegex(
219 r
"This trace class destruction listener does not match the trace class object\.",
221 trace_class_2
.remove_destruction_listener(handle1
)
223 def test_remove_destruction_listener_twice(self
):
224 def on_trace_class_destruction(trace_class
):
227 trace_class
= get_default_trace_class()
228 handle
= trace_class
.add_destruction_listener(on_trace_class_destruction
)
230 trace_class
.remove_destruction_listener(handle
)
232 with self
.assertRaisesRegex(
233 ValueError, r
"This trace class destruction listener was already removed\."
235 trace_class
.remove_destruction_listener(handle
)
237 def test_raise_in_destruction_listener(self
):
238 def on_trace_class_destruction(trace_class
):
239 raise ValueError("it hurts")
241 trace_class
= get_default_trace_class()
242 trace_class
.add_destruction_listener(on_trace_class_destruction
)
247 if __name__
== "__main__":