lib: make can_seek_ns_from_origin logic use `can_seek_forward` property of iterator
[babeltrace.git] / tests / bindings / python / bt2 / test_trace_class.py
1 #
2 # Copyright (C) 2019 EfficiOS Inc.
3 #
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
7 # of the License.
8 #
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.
13 #
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.
17 #
18
19 import unittest
20 from utils import (
21 run_in_component_init,
22 get_default_trace_class,
23 get_const_stream_beginning_message,
24 )
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
28
29
30 class TraceClassTestCase(unittest.TestCase):
31 def assertRaisesInComponentInit(self, expected_exc_type, user_code):
32 def f(comp_self):
33 try:
34 user_code(comp_self)
35 except Exception as exc:
36 return type(exc)
37
38 exc_type = run_in_component_init(f)
39 self.assertIsNotNone(exc_type)
40 self.assertEqual(exc_type, expected_exc_type)
41
42 def test_create_default(self):
43 def f(comp_self):
44 return comp_self._create_trace_class()
45
46 tc = run_in_component_init(f)
47
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)
52
53 def test_create_user_attributes(self):
54 def f(comp_self):
55 return comp_self._create_trace_class(user_attributes={'salut': 23})
56
57 tc = run_in_component_init(f)
58 self.assertEqual(tc.user_attributes, {'salut': 23})
59
60 def test_create_invalid_user_attributes(self):
61 def f(comp_self):
62 return comp_self._create_trace_class(user_attributes=object())
63
64 self.assertRaisesInComponentInit(TypeError, f)
65
66 def test_create_invalid_user_attributes_value_type(self):
67 def f(comp_self):
68 return comp_self._create_trace_class(user_attributes=23)
69
70 self.assertRaisesInComponentInit(TypeError, f)
71
72 def test_automatic_stream_class_id(self):
73 def f(comp_self):
74 return comp_self._create_trace_class(assigns_automatic_stream_class_id=True)
75
76 tc = run_in_component_init(f)
77 self.assertTrue(tc.assigns_automatic_stream_class_id)
78
79 # This should not throw.
80 sc1 = tc.create_stream_class()
81 sc2 = tc.create_stream_class()
82
83 self.assertIs(type(sc1), bt2_stream_class._StreamClass)
84 self.assertIs(type(sc2), bt2_stream_class._StreamClass)
85 self.assertNotEqual(sc1.id, sc2.id)
86
87 def test_automatic_stream_class_id_raises(self):
88 def f(comp_self):
89 return comp_self._create_trace_class(assigns_automatic_stream_class_id=True)
90
91 tc = run_in_component_init(f)
92 self.assertTrue(tc.assigns_automatic_stream_class_id)
93
94 with self.assertRaises(ValueError):
95 tc.create_stream_class(23)
96
97 def test_no_assigns_automatic_stream_class_id(self):
98 def f(comp_self):
99 return comp_self._create_trace_class(
100 assigns_automatic_stream_class_id=False
101 )
102
103 tc = run_in_component_init(f)
104 self.assertFalse(tc.assigns_automatic_stream_class_id)
105
106 sc = tc.create_stream_class(id=28)
107 self.assertEqual(sc.id, 28)
108
109 def test_no_assigns_automatic_stream_class_id_raises(self):
110 def f(comp_self):
111 return comp_self._create_trace_class(
112 assigns_automatic_stream_class_id=False
113 )
114
115 tc = run_in_component_init(f)
116 self.assertFalse(tc.assigns_automatic_stream_class_id)
117
118 # In this mode, it is required to pass an explicit id.
119 with self.assertRaises(ValueError):
120 tc.create_stream_class()
121
122 @staticmethod
123 def _create_trace_class_with_some_stream_classes():
124 def f(comp_self):
125 return comp_self._create_trace_class(
126 assigns_automatic_stream_class_id=False
127 )
128
129 tc = run_in_component_init(f)
130 sc1 = tc.create_stream_class(id=12)
131 sc2 = tc.create_stream_class(id=54)
132 sc3 = tc.create_stream_class(id=2018)
133 return tc, sc1, sc2, sc3
134
135 def test_getitem(self):
136 tc, _, _, sc3 = self._create_trace_class_with_some_stream_classes()
137 self.assertIs(type(tc[2018]), bt2_stream_class._StreamClass)
138 self.assertEqual(tc[2018].addr, sc3.addr)
139
140 def test_const_getitem(self):
141 const_tc = get_const_stream_beginning_message().stream.trace.cls
142 self.assertIs(type(const_tc[0]), bt2_stream_class._StreamClassConst)
143
144 def test_getitem_wrong_key_type(self):
145 tc, _, _, _ = self._create_trace_class_with_some_stream_classes()
146 with self.assertRaises(TypeError):
147 tc['hello']
148
149 def test_getitem_wrong_key(self):
150 tc, _, _, _ = self._create_trace_class_with_some_stream_classes()
151 with self.assertRaises(KeyError):
152 tc[4]
153
154 def test_len(self):
155 tc = get_default_trace_class()
156 self.assertEqual(len(tc), 0)
157 tc.create_stream_class()
158 self.assertEqual(len(tc), 1)
159
160 def test_iter(self):
161 tc, sc1, sc2, sc3 = self._create_trace_class_with_some_stream_classes()
162
163 for sc_id, stream_class in tc.items():
164 if sc_id == 12:
165 self.assertIs(type(stream_class), bt2_stream_class._StreamClass)
166 self.assertEqual(stream_class.addr, sc1.addr)
167 elif sc_id == 54:
168 self.assertEqual(stream_class.addr, sc2.addr)
169 elif sc_id == 2018:
170 self.assertEqual(stream_class.addr, sc3.addr)
171
172 def test_const_iter(self):
173 const_tc = get_const_stream_beginning_message().stream.trace.cls
174 const_sc = list(const_tc.values())[0]
175 self.assertIs(type(const_sc), bt2_stream_class._StreamClassConst)
176
177 def test_destruction_listener(self):
178 def on_trace_class_destruction(trace_class):
179 nonlocal num_destruct_calls
180 num_destruct_calls += 1
181
182 num_destruct_calls = 0
183
184 trace_class = get_default_trace_class()
185
186 handle1 = trace_class.add_destruction_listener(on_trace_class_destruction)
187 self.assertIs(type(handle1), bt2_utils._ListenerHandle)
188
189 handle2 = trace_class.add_destruction_listener(on_trace_class_destruction)
190
191 trace_class.remove_destruction_listener(handle2)
192
193 del handle1
194 del handle2
195
196 self.assertEqual(num_destruct_calls, 0)
197
198 del trace_class
199
200 self.assertEqual(num_destruct_calls, 1)
201
202 def test_remove_destruction_listener_wrong_type(self):
203 trace_class = get_default_trace_class()
204
205 with self.assertRaisesRegex(
206 TypeError, r"'int' is not a '<class 'bt2.utils._ListenerHandle'>' object"
207 ):
208 trace_class.remove_destruction_listener(123)
209
210 def test_remove_destruction_listener_wrong_object(self):
211 def on_trace_class_destruction(trace_class):
212 pass
213
214 trace_class_1 = get_default_trace_class()
215 trace_class_2 = get_default_trace_class()
216
217 handle1 = trace_class_1.add_destruction_listener(on_trace_class_destruction)
218
219 with self.assertRaisesRegex(
220 ValueError,
221 r'This trace class destruction listener does not match the trace object\.',
222 ):
223 trace_class_2.remove_destruction_listener(handle1)
224
225 def test_remove_destruction_listener_twice(self):
226 def on_trace_class_destruction(trace_class):
227 pass
228
229 trace_class = get_default_trace_class()
230 handle = trace_class.add_destruction_listener(on_trace_class_destruction)
231
232 trace_class.remove_destruction_listener(handle)
233
234 with self.assertRaisesRegex(
235 ValueError, r'This trace class destruction listener was already removed\.'
236 ):
237 trace_class.remove_destruction_listener(handle)
238
239
240 if __name__ == '__main__':
241 unittest.main()
This page took 0.035374 seconds and 4 git commands to generate.