1 # SPDX-License-Identifier: GPL-2.0-only
3 # Copyright (C) 2019 EfficiOS Inc.
11 class QueryExecutorTestCase(unittest
.TestCase
):
12 def test_default_interrupter(self
):
13 class MySink(bt2
._UserSinkComponent
):
14 def _user_consume(self
):
17 query_exec
= bt2
.QueryExecutor(MySink
, 'obj')
18 interrupter
= query_exec
.default_interrupter
19 self
.assertIs(type(interrupter
), bt2
.Interrupter
)
22 class MySink(bt2
._UserSinkComponent
):
23 def _user_consume(self
):
27 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
30 return {'null': None, 'bt2': 'BT2'}
34 'array': ['coucou', 23, None],
35 'other_map': {'yes': 'yeah', '19': 19, 'minus 1.5': -1.5},
39 res
= bt2
.QueryExecutor(MySink
, 'obj', params
).query()
40 self
.assertIs(type(res
), bt2
._MapValueConst
)
41 self
.assertIs(type(res
['bt2']), bt2
._StringValueConst
)
42 self
.assertEqual(query_params
, params
)
43 self
.assertEqual(res
, {'null': None, 'bt2': 'BT2'})
46 def test_query_params_none(self
):
47 class MySink(bt2
._UserSinkComponent
):
48 def _user_consume(self
):
52 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
57 bt2
.QueryExecutor(MySink
, 'obj', None).query()
58 self
.assertIs(query_params
, None)
61 def test_query_no_params(self
):
62 class MySink(bt2
._UserSinkComponent
):
63 def _user_consume(self
):
67 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
72 bt2
.QueryExecutor(MySink
, 'obj').query()
73 self
.assertIs(query_params
, None)
76 def test_query_with_method_obj(self
):
77 class MySink(bt2
._UserSinkComponent
):
78 def _user_consume(self
):
82 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
83 nonlocal query_method_obj
84 query_method_obj
= method_obj
86 query_method_obj
= None
88 bt2
.QueryExecutor(MySink
, 'obj', method_obj
=method_obj
).query()
89 self
.assertIs(query_method_obj
, method_obj
)
92 def test_query_with_method_obj_del_ref(self
):
93 class MySink(bt2
._UserSinkComponent
):
94 def _user_consume(self
):
98 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
99 nonlocal query_method_obj
100 query_method_obj
= method_obj
105 query_method_obj
= None
106 method_obj
= Custom()
107 method_obj
.hola
= 'hello'
108 query_exec
= bt2
.QueryExecutor(MySink
, 'obj', method_obj
=method_obj
)
111 self
.assertIsInstance(query_method_obj
, Custom
)
112 self
.assertEqual(query_method_obj
.hola
, 'hello')
115 def test_query_with_none_method_obj(self
):
116 class MySink(bt2
._UserSinkComponent
):
117 def _user_consume(self
):
121 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
122 nonlocal query_method_obj
123 query_method_obj
= method_obj
125 query_method_obj
= object()
126 bt2
.QueryExecutor(MySink
, 'obj').query()
127 self
.assertIsNone(query_method_obj
)
130 def test_query_with_method_obj_non_python_comp_cls(self
):
131 plugin
= bt2
.find_plugin('text', find_in_user_dir
=False, find_in_sys_dir
=False)
132 assert plugin
is not None
133 cc
= plugin
.source_component_classes
['dmesg']
134 assert cc
is not None
136 with self
.assertRaisesRegex(
138 re
.escape(r
'cannot pass a Python object to a non-Python component class'),
140 bt2
.QueryExecutor(cc
, 'obj', method_obj
=object()).query()
142 def test_query_logging_level(self
):
143 class MySink(bt2
._UserSinkComponent
):
144 def _user_consume(self
):
148 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
149 nonlocal query_log_level
150 query_log_level
= priv_query_exec
.logging_level
152 query_log_level
= None
153 query_exec
= bt2
.QueryExecutor(MySink
, 'obj', None)
154 query_exec
.logging_level
= bt2
.LoggingLevel
.INFO
156 self
.assertEqual(query_log_level
, bt2
.LoggingLevel
.INFO
)
159 def test_query_gen_error(self
):
160 class MySink(bt2
._UserSinkComponent
):
161 def _user_consume(self
):
165 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
168 with self
.assertRaises(bt2
._Error
) as ctx
:
169 bt2
.QueryExecutor(MySink
, 'obj', [17, 23]).query()
172 self
.assertEqual(len(exc
), 3)
174 self
.assertIsInstance(cause
, bt2
._ComponentClassErrorCause
)
175 self
.assertIn('raise ValueError', cause
.message
)
176 self
.assertEqual(cause
.component_class_type
, bt2
.ComponentClassType
.SINK
)
177 self
.assertEqual(cause
.component_class_name
, 'MySink')
179 def test_query_unknown_object(self
):
180 class MySink(bt2
._UserSinkComponent
):
181 def _user_consume(self
):
185 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
186 raise bt2
.UnknownObject
188 with self
.assertRaises(bt2
.UnknownObject
):
189 bt2
.QueryExecutor(MySink
, 'obj', [17, 23]).query()
191 def test_query_logging_level_invalid_type(self
):
192 class MySink(bt2
._UserSinkComponent
):
193 def _user_consume(self
):
197 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
200 query_exec
= bt2
.QueryExecutor(MySink
, 'obj', [17, 23])
202 with self
.assertRaises(TypeError):
203 query_exec
.logging_level
= 'yeah'
205 def test_query_logging_level_invalid_value(self
):
206 class MySink(bt2
._UserSinkComponent
):
207 def _user_consume(self
):
211 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
214 query_exec
= bt2
.QueryExecutor(MySink
, 'obj', [17, 23])
216 with self
.assertRaises(ValueError):
217 query_exec
.logging_level
= 12345
219 def test_query_try_again(self
):
220 class MySink(bt2
._UserSinkComponent
):
221 def _user_consume(self
):
225 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
228 with self
.assertRaises(bt2
.TryAgain
):
229 bt2
.QueryExecutor(MySink
, 'obj', [17, 23]).query()
231 def test_query_add_interrupter(self
):
232 class MySink(bt2
._UserSinkComponent
):
233 def _user_consume(self
):
237 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
238 nonlocal interrupter2
239 test_self
.assertFalse(query_exec
.is_interrupted
)
241 test_self
.assertTrue(query_exec
.is_interrupted
)
243 test_self
.assertFalse(query_exec
.is_interrupted
)
245 interrupter1
= bt2
.Interrupter()
246 interrupter2
= bt2
.Interrupter()
248 query_exec
= bt2
.QueryExecutor(MySink
, 'obj', [17, 23])
249 query_exec
.add_interrupter(interrupter1
)
250 query_exec
.add_interrupter(interrupter2
)
253 def test_query_interrupt(self
):
254 class MySink(bt2
._UserSinkComponent
):
255 def _user_consume(self
):
259 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
260 test_self
.assertFalse(query_exec
.is_interrupted
)
261 query_exec
.default_interrupter
.set()
262 test_self
.assertTrue(query_exec
.is_interrupted
)
265 query_exec
= bt2
.QueryExecutor(MySink
, 'obj', [17, 23])
268 def test_query_priv_executor_invalid_after(self
):
269 class MySink(bt2
._UserSinkComponent
):
270 def _user_consume(self
):
274 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
275 nonlocal test_priv_query_exec
276 test_priv_query_exec
= priv_query_exec
278 test_priv_query_exec
= None
279 query_exec
= bt2
.QueryExecutor(MySink
, 'obj', [17, 23])
281 assert test_priv_query_exec
is not None
283 with self
.assertRaises(RuntimeError):
284 test_priv_query_exec
.logging_level
286 del test_priv_query_exec
289 if __name__
== '__main__':