2 # Copyright (C) 2019 EfficiOS Inc.
4 # This program is free software; you can redistribute it and/or
5 # modify it under the terms of the GNU General Public License
6 # as published by the Free Software Foundation; only version 2
9 # This program is distributed in the hope that it will be useful,
10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 # GNU General Public License for more details.
14 # You should have received a copy of the GNU General Public License
15 # along with this program; if not, write to the Free Software
16 # Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
21 run_in_component_init
,
22 get_default_trace_class
,
23 get_const_stream_beginning_message
,
25 from bt2
import stream_class
as bt2_stream_class
26 from bt2
import trace_class
as bt2_trace_class
27 from bt2
import utils
as bt2_utils
30 class TraceClassTestCase(unittest
.TestCase
):
31 def assertRaisesInComponentInit(self
, expected_exc_type
, user_code
):
35 except Exception as exc
:
38 exc_type
= run_in_component_init(f
)
39 self
.assertIsNotNone(exc_type
)
40 self
.assertEqual(exc_type
, expected_exc_type
)
42 def test_create_default(self
):
44 return comp_self
._create
_trace
_class
()
46 tc
= run_in_component_init(f
)
48 self
.assertEqual(len(tc
), 0)
49 self
.assertIs(type(tc
), bt2_trace_class
._TraceClass
)
50 self
.assertTrue(tc
.assigns_automatic_stream_class_id
)
51 self
.assertEqual(len(tc
.user_attributes
), 0)
53 def test_create_user_attributes(self
):
55 return comp_self
._create
_trace
_class
(user_attributes
={'salut': 23})
57 tc
= run_in_component_init(f
)
58 self
.assertEqual(tc
.user_attributes
, {'salut': 23})
60 def test_create_invalid_user_attributes(self
):
62 return comp_self
._create
_trace
_class
(user_attributes
=object())
64 self
.assertRaisesInComponentInit(TypeError, f
)
66 def test_create_invalid_user_attributes_value_type(self
):
68 return comp_self
._create
_trace
_class
(user_attributes
=23)
70 self
.assertRaisesInComponentInit(TypeError, f
)
72 def test_create_invalid_automatic_stream_class_id_type(self
):
74 return comp_self
._create
_trace
_class
(
75 assigns_automatic_stream_class_id
='perchaude'
78 self
.assertRaisesInComponentInit(TypeError, f
)
80 def test_automatic_stream_class_id(self
):
82 return comp_self
._create
_trace
_class
(assigns_automatic_stream_class_id
=True)
84 tc
= run_in_component_init(f
)
85 self
.assertTrue(tc
.assigns_automatic_stream_class_id
)
87 # This should not throw.
88 sc1
= tc
.create_stream_class()
89 sc2
= tc
.create_stream_class()
91 self
.assertIs(type(sc1
), bt2_stream_class
._StreamClass
)
92 self
.assertIs(type(sc2
), bt2_stream_class
._StreamClass
)
93 self
.assertNotEqual(sc1
.id, sc2
.id)
95 def test_automatic_stream_class_id_raises(self
):
97 return comp_self
._create
_trace
_class
(assigns_automatic_stream_class_id
=True)
99 tc
= run_in_component_init(f
)
100 self
.assertTrue(tc
.assigns_automatic_stream_class_id
)
102 with self
.assertRaises(ValueError):
103 tc
.create_stream_class(23)
105 def test_no_assigns_automatic_stream_class_id(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 sc
= tc
.create_stream_class(id=28)
115 self
.assertEqual(sc
.id, 28)
117 def test_no_assigns_automatic_stream_class_id_raises(self
):
119 return comp_self
._create
_trace
_class
(
120 assigns_automatic_stream_class_id
=False
123 tc
= run_in_component_init(f
)
124 self
.assertFalse(tc
.assigns_automatic_stream_class_id
)
126 # In this mode, it is required to pass an explicit id.
127 with self
.assertRaises(ValueError):
128 tc
.create_stream_class()
131 def _create_trace_class_with_some_stream_classes():
133 return comp_self
._create
_trace
_class
(
134 assigns_automatic_stream_class_id
=False
137 tc
= run_in_component_init(f
)
138 sc1
= tc
.create_stream_class(id=12)
139 sc2
= tc
.create_stream_class(id=54)
140 sc3
= tc
.create_stream_class(id=2018)
141 return tc
, sc1
, sc2
, sc3
143 def test_getitem(self
):
144 tc
, _
, _
, sc3
= self
._create
_trace
_class
_with
_some
_stream
_classes
()
145 self
.assertIs(type(tc
[2018]), bt2_stream_class
._StreamClass
)
146 self
.assertEqual(tc
[2018].addr
, sc3
.addr
)
148 def test_const_getitem(self
):
149 const_tc
= get_const_stream_beginning_message().stream
.trace
.cls
150 self
.assertIs(type(const_tc
[0]), bt2_stream_class
._StreamClassConst
)
152 def test_getitem_wrong_key_type(self
):
153 tc
, _
, _
, _
= self
._create
_trace
_class
_with
_some
_stream
_classes
()
154 with self
.assertRaises(TypeError):
157 def test_getitem_wrong_key(self
):
158 tc
, _
, _
, _
= self
._create
_trace
_class
_with
_some
_stream
_classes
()
159 with self
.assertRaises(KeyError):
163 tc
= get_default_trace_class()
164 self
.assertEqual(len(tc
), 0)
165 tc
.create_stream_class()
166 self
.assertEqual(len(tc
), 1)
169 tc
, sc1
, sc2
, sc3
= self
._create
_trace
_class
_with
_some
_stream
_classes
()
171 for sc_id
, stream_class
in tc
.items():
173 self
.assertIs(type(stream_class
), bt2_stream_class
._StreamClass
)
174 self
.assertEqual(stream_class
.addr
, sc1
.addr
)
176 self
.assertEqual(stream_class
.addr
, sc2
.addr
)
178 self
.assertEqual(stream_class
.addr
, sc3
.addr
)
180 def test_const_iter(self
):
181 const_tc
= get_const_stream_beginning_message().stream
.trace
.cls
182 const_sc
= list(const_tc
.values())[0]
183 self
.assertIs(type(const_sc
), bt2_stream_class
._StreamClassConst
)
185 def test_destruction_listener(self
):
186 def on_trace_class_destruction(trace_class
):
187 nonlocal num_destruct_calls
188 num_destruct_calls
+= 1
190 num_destruct_calls
= 0
192 trace_class
= get_default_trace_class()
194 handle1
= trace_class
.add_destruction_listener(on_trace_class_destruction
)
195 self
.assertIs(type(handle1
), bt2_utils
._ListenerHandle
)
197 handle2
= trace_class
.add_destruction_listener(on_trace_class_destruction
)
199 trace_class
.remove_destruction_listener(handle2
)
204 self
.assertEqual(num_destruct_calls
, 0)
208 self
.assertEqual(num_destruct_calls
, 1)
210 def test_remove_destruction_listener_wrong_type(self
):
211 trace_class
= get_default_trace_class()
213 with self
.assertRaisesRegex(
214 TypeError, r
"'int' is not a '<class 'bt2.utils._ListenerHandle'>' object"
216 trace_class
.remove_destruction_listener(123)
218 def test_remove_destruction_listener_wrong_object(self
):
219 def on_trace_class_destruction(trace_class
):
222 trace_class_1
= get_default_trace_class()
223 trace_class_2
= get_default_trace_class()
225 handle1
= trace_class_1
.add_destruction_listener(on_trace_class_destruction
)
227 with self
.assertRaisesRegex(
229 r
'This trace class destruction listener does not match the trace object\.',
231 trace_class_2
.remove_destruction_listener(handle1
)
233 def test_remove_destruction_listener_twice(self
):
234 def on_trace_class_destruction(trace_class
):
237 trace_class
= get_default_trace_class()
238 handle
= trace_class
.add_destruction_listener(on_trace_class_destruction
)
240 trace_class
.remove_destruction_listener(handle
)
242 with self
.assertRaisesRegex(
243 ValueError, r
'This trace class destruction listener was already removed\.'
245 trace_class
.remove_destruction_listener(handle
)
247 def test_raise_in_destruction_listener(self
):
248 def on_trace_class_destruction(trace_class
):
249 raise ValueError('it hurts')
251 trace_class
= get_default_trace_class()
252 trace_class
.add_destruction_listener(on_trace_class_destruction
)
257 if __name__
== '__main__':