Move to kernel style SPDX license identifiers
[babeltrace.git] / tests / bindings / python / bt2 / test_component_class.py
1 # SPDX-License-Identifier: GPL-2.0-only
2 #
3 # Copyright (C) 2019 EfficiOS Inc.
4 #
5
6 import unittest
7 import bt2
8
9
10 class UserComponentClassTestCase(unittest.TestCase):
11 def _test_no_init(self, cls):
12 with self.assertRaises(RuntimeError):
13 cls()
14
15 def test_no_init_source(self):
16 class MySource(
17 bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
18 ):
19 pass
20
21 self._test_no_init(MySource)
22
23 def test_no_init_filter(self):
24 class MyFilter(
25 bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
26 ):
27 pass
28
29 self._test_no_init(MyFilter)
30
31 def test_no_init_sink(self):
32 class MySink(bt2._UserSinkComponent):
33 def _user_consume(self):
34 pass
35
36 self._test_no_init(MySink)
37
38 def test_incomplete_source_no_msg_iter_cls(self):
39 with self.assertRaises(bt2._IncompleteUserClass):
40
41 class MySource(bt2._UserSourceComponent):
42 pass
43
44 def test_incomplete_source_wrong_msg_iter_cls_type(self):
45 with self.assertRaises(bt2._IncompleteUserClass):
46
47 class MySource(bt2._UserSourceComponent, message_iterator_class=int):
48 pass
49
50 def test_incomplete_filter_no_msg_iter_cls(self):
51 with self.assertRaises(bt2._IncompleteUserClass):
52
53 class MyFilter(bt2._UserFilterComponent):
54 pass
55
56 def test_incomplete_sink_no_consume_method(self):
57 with self.assertRaises(bt2._IncompleteUserClass):
58
59 class MySink(bt2._UserSinkComponent):
60 pass
61
62 def test_minimal_source(self):
63 class MySource(
64 bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
65 ):
66 pass
67
68 def test_minimal_filter(self):
69 class MyFilter(
70 bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
71 ):
72 pass
73
74 def test_minimal_sink(self):
75 class MySink(bt2._UserSinkComponent):
76 def _user_consume(self):
77 pass
78
79 def test_default_name(self):
80 class MySink(bt2._UserSinkComponent):
81 def _user_consume(self):
82 pass
83
84 self.assertEqual(MySink.name, 'MySink')
85
86 def test_custom_name(self):
87 class MySink(bt2._UserSinkComponent, name='salut'):
88 def _user_consume(self):
89 pass
90
91 self.assertEqual(MySink.name, 'salut')
92
93 def test_invalid_custom_name(self):
94 with self.assertRaises(TypeError):
95
96 class MySink(bt2._UserSinkComponent, name=23):
97 def _user_consume(self):
98 pass
99
100 def test_description(self):
101 class MySink(bt2._UserSinkComponent):
102 """
103 The description.
104
105 Bacon ipsum dolor amet ribeye t-bone corned beef, beef jerky
106 porchetta burgdoggen prosciutto chicken frankfurter boudin
107 hamburger doner bacon turducken. Sirloin shank sausage,
108 boudin meatloaf alcatra meatball t-bone tongue pastrami
109 cupim flank tenderloin.
110 """
111
112 def _user_consume(self):
113 pass
114
115 self.assertEqual(MySink.description, 'The description.')
116
117 def test_empty_description(self):
118 class MySink(bt2._UserSinkComponent):
119 """
120 """
121
122 def _user_consume(self):
123 pass
124
125 self.assertIsNone(MySink.description)
126
127 def test_help(self):
128 class MySink(bt2._UserSinkComponent):
129 """
130 The description.
131
132 The help
133 text is
134 here.
135 """
136
137 def _user_consume(self):
138 pass
139
140 self.assertEqual(MySink.help, 'The help\ntext is\nhere.')
141
142 def test_addr(self):
143 class MySink(bt2._UserSinkComponent):
144 def _user_consume(self):
145 pass
146
147 self.assertIsInstance(MySink.addr, int)
148 self.assertNotEqual(MySink.addr, 0)
149
150 def test_query_not_implemented(self):
151 class MySink(bt2._UserSinkComponent):
152 def _user_consume(self):
153 pass
154
155 with self.assertRaises(bt2.UnknownObject):
156 bt2.QueryExecutor(MySink, 'obj', 23).query()
157
158 def test_query_raises(self):
159 class MySink(bt2._UserSinkComponent):
160 def _user_consume(self):
161 pass
162
163 @classmethod
164 def _user_query(cls, priv_query_exec, obj, params):
165 raise ValueError
166
167 with self.assertRaises(bt2._Error):
168 bt2.QueryExecutor(MySink, 'obj', 23).query()
169
170 def test_query_wrong_return_type(self):
171 class MySink(bt2._UserSinkComponent):
172 def _user_consume(self):
173 pass
174
175 @classmethod
176 def _user_query(cls, priv_query_exec, obj, params, method_obj):
177 return ...
178
179 with self.assertRaises(bt2._Error):
180 bt2.QueryExecutor(MySink, 'obj', 23).query()
181
182 def test_query_params_none(self):
183 class MySink(bt2._UserSinkComponent):
184 def _user_consume(self):
185 pass
186
187 @classmethod
188 def _user_query(cls, priv_query_exec, obj, params, method_obj):
189 nonlocal query_params
190 query_params = params
191 return None
192
193 query_params = None
194 params = None
195 res = bt2.QueryExecutor(MySink, 'obj', params).query()
196 self.assertEqual(query_params, params)
197 self.assertIsNone(res)
198 del query_params
199
200 def test_query_logging_level(self):
201 class MySink(bt2._UserSinkComponent):
202 def _user_consume(self):
203 pass
204
205 @classmethod
206 def _user_query(cls, priv_query_exec, obj, params, method_obj):
207 nonlocal query_log_level
208 query_log_level = priv_query_exec.logging_level
209
210 query_log_level = None
211 query_exec = bt2.QueryExecutor(MySink, 'obj', None)
212 query_exec.logging_level = bt2.LoggingLevel.WARNING
213 query_exec.query()
214 self.assertEqual(query_log_level, bt2.LoggingLevel.WARNING)
215 del query_log_level
216
217 def test_query_returns_none(self):
218 class MySink(bt2._UserSinkComponent):
219 def _user_consume(self):
220 pass
221
222 @staticmethod
223 def _user_query(priv_query_exec, obj, params, method_obj):
224 return
225
226 res = bt2.QueryExecutor(MySink, 'obj', None).query()
227 self.assertIsNone(res)
228
229 def test_query_simple(self):
230 class MySink(bt2._UserSinkComponent):
231 def _user_consume(self):
232 pass
233
234 @classmethod
235 def _user_query(cls, priv_query_exec, obj, params, method_obj):
236 nonlocal query_params
237 query_params = params
238 return 17.5
239
240 query_params = None
241 params = ['coucou', 23, None]
242 res = bt2.QueryExecutor(MySink, 'obj', params).query()
243 self.assertEqual(query_params, params)
244 self.assertEqual(res, 17.5)
245 del query_params
246
247 def test_query_complex(self):
248 class MySink(bt2._UserSinkComponent):
249 def _user_consume(self):
250 pass
251
252 @classmethod
253 def _user_query(cls, priv_query_exec, obj, params, method_obj):
254 nonlocal query_params
255 query_params = params
256 return {'null': None, 'bt2': 'BT2'}
257
258 query_params = None
259 params = {
260 'array': ['coucou', 23, None],
261 'other_map': {'yes': 'yeah', '19': 19, 'minus 1.5': -1.5},
262 'null': None,
263 }
264
265 res = bt2.QueryExecutor(MySink, 'obj', params).query()
266 self.assertEqual(query_params, params)
267 self.assertEqual(res, {'null': None, 'bt2': 'BT2'})
268 del query_params
269
270 def test_eq(self):
271 class MySink(bt2._UserSinkComponent):
272 def _user_consume(self):
273 pass
274
275 self.assertEqual(MySink, MySink)
276
277
278 class ComponentClassTestCase(unittest.TestCase):
279 def setUp(self):
280 class MySink(bt2._UserSinkComponent):
281 '''
282 The description.
283
284 The help.
285 '''
286
287 def _user_consume(self):
288 pass
289
290 @classmethod
291 def _user_query(cls, priv_query_exec, obj, params, method_obj):
292 return [obj, params, 23]
293
294 self._py_comp_cls = MySink
295 graph = bt2.Graph()
296 comp = graph.add_component(MySink, 'salut')
297 self._comp_cls = comp.cls
298 self.assertIs(type(self._comp_cls), bt2._SinkComponentClassConst)
299
300 def tearDown(self):
301 del self._py_comp_cls
302 del self._comp_cls
303
304 def test_description(self):
305 self.assertEqual(self._comp_cls.description, 'The description.')
306
307 def test_help(self):
308 self.assertEqual(self._comp_cls.help, 'The help.')
309
310 def test_name(self):
311 self.assertEqual(self._comp_cls.name, 'MySink')
312
313 def test_addr(self):
314 self.assertIsInstance(self._comp_cls.addr, int)
315 self.assertNotEqual(self._comp_cls.addr, 0)
316
317 def test_eq_invalid(self):
318 self.assertFalse(self._comp_cls == 23)
319
320 def test_eq(self):
321 self.assertEqual(self._comp_cls, self._comp_cls)
322 self.assertEqual(self._py_comp_cls, self._comp_cls)
323
324 def test_query(self):
325 res = bt2.QueryExecutor(
326 self._comp_cls, 'an object', {'yes': 'no', 'book': -17}
327 ).query()
328 expected = ['an object', {'yes': 'no', 'book': -17}, 23]
329 self.assertEqual(res, expected)
330
331
332 if __name__ == '__main__':
333 unittest.main()
This page took 0.036685 seconds and 4 git commands to generate.