1 # SPDX-License-Identifier: GPL-2.0-only
3 # Copyright (C) 2019 EfficiOS Inc.
9 get_default_trace_class
,
10 get_const_stream_beginning_message
,
12 from bt2
import stream_class
as bt2_stream_class
13 from bt2
import trace_class
as bt2_trace_class
14 from bt2
import utils
as bt2_utils
17 class TraceClassTestCase(unittest
.TestCase
):
18 def assertRaisesInComponentInit(self
, expected_exc_type
, user_code
):
22 except Exception as exc
:
25 exc_type
= run_in_component_init(f
)
26 self
.assertIsNotNone(exc_type
)
27 self
.assertEqual(exc_type
, expected_exc_type
)
29 def test_create_default(self
):
31 return comp_self
._create
_trace
_class
()
33 tc
= run_in_component_init(f
)
35 self
.assertEqual(len(tc
), 0)
36 self
.assertIs(type(tc
), bt2_trace_class
._TraceClass
)
37 self
.assertTrue(tc
.assigns_automatic_stream_class_id
)
38 self
.assertEqual(len(tc
.user_attributes
), 0)
40 def test_create_user_attributes(self
):
42 return comp_self
._create
_trace
_class
(user_attributes
={"salut": 23})
44 tc
= run_in_component_init(f
)
45 self
.assertEqual(tc
.user_attributes
, {"salut": 23})
47 def test_create_invalid_user_attributes(self
):
49 return comp_self
._create
_trace
_class
(user_attributes
=object())
51 self
.assertRaisesInComponentInit(TypeError, f
)
53 def test_create_invalid_user_attributes_value_type(self
):
55 return comp_self
._create
_trace
_class
(user_attributes
=23)
57 self
.assertRaisesInComponentInit(TypeError, f
)
59 def test_create_invalid_automatic_stream_class_id_type(self
):
61 return comp_self
._create
_trace
_class
(
62 assigns_automatic_stream_class_id
="perchaude"
65 self
.assertRaisesInComponentInit(TypeError, f
)
67 def test_automatic_stream_class_id(self
):
69 return comp_self
._create
_trace
_class
(assigns_automatic_stream_class_id
=True)
71 tc
= run_in_component_init(f
)
72 self
.assertTrue(tc
.assigns_automatic_stream_class_id
)
74 # This should not throw.
75 sc1
= tc
.create_stream_class()
76 sc2
= tc
.create_stream_class()
78 self
.assertIs(type(sc1
), bt2_stream_class
._StreamClass
)
79 self
.assertIs(type(sc2
), bt2_stream_class
._StreamClass
)
80 self
.assertNotEqual(sc1
.id, sc2
.id)
82 def test_automatic_stream_class_id_raises(self
):
84 return comp_self
._create
_trace
_class
(assigns_automatic_stream_class_id
=True)
86 tc
= run_in_component_init(f
)
87 self
.assertTrue(tc
.assigns_automatic_stream_class_id
)
89 with self
.assertRaises(ValueError):
90 tc
.create_stream_class(23)
92 def test_no_assigns_automatic_stream_class_id(self
):
94 return comp_self
._create
_trace
_class
(
95 assigns_automatic_stream_class_id
=False
98 tc
= run_in_component_init(f
)
99 self
.assertFalse(tc
.assigns_automatic_stream_class_id
)
101 sc
= tc
.create_stream_class(id=28)
102 self
.assertEqual(sc
.id, 28)
104 def test_no_assigns_automatic_stream_class_id_raises(self
):
106 return comp_self
._create
_trace
_class
(
107 assigns_automatic_stream_class_id
=False
110 tc
= run_in_component_init(f
)
111 self
.assertFalse(tc
.assigns_automatic_stream_class_id
)
113 # In this mode, it is required to pass an explicit id.
114 with self
.assertRaises(ValueError):
115 tc
.create_stream_class()
118 def _create_trace_class_with_some_stream_classes():
120 return comp_self
._create
_trace
_class
(
121 assigns_automatic_stream_class_id
=False
124 tc
= run_in_component_init(f
)
125 sc1
= tc
.create_stream_class(id=12)
126 sc2
= tc
.create_stream_class(id=54)
127 sc3
= tc
.create_stream_class(id=2018)
128 return tc
, sc1
, sc2
, sc3
130 def test_getitem(self
):
131 tc
, _
, _
, sc3
= self
._create
_trace
_class
_with
_some
_stream
_classes
()
132 self
.assertIs(type(tc
[2018]), bt2_stream_class
._StreamClass
)
133 self
.assertEqual(tc
[2018].addr
, sc3
.addr
)
135 def test_const_getitem(self
):
136 const_tc
= get_const_stream_beginning_message().stream
.trace
.cls
137 self
.assertIs(type(const_tc
[0]), bt2_stream_class
._StreamClassConst
)
139 def test_getitem_wrong_key_type(self
):
140 tc
, _
, _
, _
= self
._create
_trace
_class
_with
_some
_stream
_classes
()
141 with self
.assertRaises(TypeError):
144 def test_getitem_wrong_key(self
):
145 tc
, _
, _
, _
= self
._create
_trace
_class
_with
_some
_stream
_classes
()
146 with self
.assertRaises(KeyError):
150 tc
= get_default_trace_class()
151 self
.assertEqual(len(tc
), 0)
152 tc
.create_stream_class()
153 self
.assertEqual(len(tc
), 1)
156 tc
, sc1
, sc2
, sc3
= self
._create
_trace
_class
_with
_some
_stream
_classes
()
158 for sc_id
, stream_class
in tc
.items():
160 self
.assertIs(type(stream_class
), bt2_stream_class
._StreamClass
)
161 self
.assertEqual(stream_class
.addr
, sc1
.addr
)
163 self
.assertEqual(stream_class
.addr
, sc2
.addr
)
165 self
.assertEqual(stream_class
.addr
, sc3
.addr
)
167 def test_const_iter(self
):
168 const_tc
= get_const_stream_beginning_message().stream
.trace
.cls
169 const_sc
= list(const_tc
.values())[0]
170 self
.assertIs(type(const_sc
), bt2_stream_class
._StreamClassConst
)
172 def test_destruction_listener(self
):
173 def on_trace_class_destruction(trace_class
):
174 nonlocal num_destruct_calls
175 num_destruct_calls
+= 1
177 num_destruct_calls
= 0
179 trace_class
= get_default_trace_class()
181 handle1
= trace_class
.add_destruction_listener(on_trace_class_destruction
)
182 self
.assertIs(type(handle1
), bt2_utils
._ListenerHandle
)
184 handle2
= trace_class
.add_destruction_listener(on_trace_class_destruction
)
186 trace_class
.remove_destruction_listener(handle2
)
188 self
.assertEqual(num_destruct_calls
, 0)
192 self
.assertEqual(num_destruct_calls
, 1)
194 def test_remove_destruction_listener_wrong_type(self
):
195 trace_class
= get_default_trace_class()
197 with self
.assertRaisesRegex(
198 TypeError, r
"'int' is not a '<class 'bt2.utils._ListenerHandle'>' object"
200 trace_class
.remove_destruction_listener(123)
202 def test_remove_destruction_listener_wrong_object(self
):
203 def on_trace_class_destruction(trace_class
):
206 trace_class_1
= get_default_trace_class()
207 trace_class_2
= get_default_trace_class()
209 handle1
= trace_class_1
.add_destruction_listener(on_trace_class_destruction
)
211 with self
.assertRaisesRegex(
213 r
"This trace class destruction listener does not match the trace class object\.",
215 trace_class_2
.remove_destruction_listener(handle1
)
217 def test_remove_destruction_listener_twice(self
):
218 def on_trace_class_destruction(trace_class
):
221 trace_class
= get_default_trace_class()
222 handle
= trace_class
.add_destruction_listener(on_trace_class_destruction
)
224 trace_class
.remove_destruction_listener(handle
)
226 with self
.assertRaisesRegex(
227 ValueError, r
"This trace class destruction listener was already removed\."
229 trace_class
.remove_destruction_listener(handle
)
231 def test_raise_in_destruction_listener(self
):
232 def on_trace_class_destruction(trace_class
):
233 raise ValueError("it hurts")
235 trace_class
= get_default_trace_class()
236 trace_class
.add_destruction_listener(on_trace_class_destruction
)
241 if __name__
== "__main__":