1 # SPDX-License-Identifier: GPL-2.0-only
3 # Copyright (C) 2019 EfficiOS Inc.
12 class QueryExecutorTestCase(unittest
.TestCase
):
13 def test_default_interrupter(self
):
14 class MySink(bt2
._UserSinkComponent
):
15 def _user_consume(self
):
18 query_exec
= bt2
.QueryExecutor(MySink
, "obj")
19 interrupter
= query_exec
.default_interrupter
20 self
.assertIs(type(interrupter
), bt2
.Interrupter
)
23 class MySink(bt2
._UserSinkComponent
):
24 def _user_consume(self
):
28 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
31 return {"null": None, "bt2": "BT2"}
35 "array": ["coucou", 23, None],
36 "other_map": {"yes": "yeah", "19": 19, "minus 1.5": -1.5},
40 res
= bt2
.QueryExecutor(MySink
, "obj", params
).query()
41 self
.assertIs(type(res
), bt2
._MapValueConst
)
42 self
.assertIs(type(res
["bt2"]), bt2
._StringValueConst
)
43 self
.assertEqual(query_params
, params
)
44 self
.assertEqual(res
, {"null": None, "bt2": "BT2"})
47 def test_query_params_none(self
):
48 class MySink(bt2
._UserSinkComponent
):
49 def _user_consume(self
):
53 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
58 bt2
.QueryExecutor(MySink
, "obj", None).query()
59 self
.assertIs(query_params
, None)
62 def test_query_no_params(self
):
63 class MySink(bt2
._UserSinkComponent
):
64 def _user_consume(self
):
68 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
73 bt2
.QueryExecutor(MySink
, "obj").query()
74 self
.assertIs(query_params
, None)
77 def test_query_with_method_obj(self
):
78 class MySink(bt2
._UserSinkComponent
):
79 def _user_consume(self
):
83 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
84 nonlocal query_method_obj
85 query_method_obj
= method_obj
87 query_method_obj
= None
89 bt2
.QueryExecutor(MySink
, "obj", method_obj
=method_obj
).query()
90 self
.assertIs(query_method_obj
, method_obj
)
93 def test_query_with_method_obj_del_ref(self
):
94 class MySink(bt2
._UserSinkComponent
):
95 def _user_consume(self
):
99 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
100 nonlocal query_method_obj
101 query_method_obj
= method_obj
106 query_method_obj
= None
107 method_obj
= Custom()
108 method_obj
.hola
= "hello"
109 query_exec
= bt2
.QueryExecutor(MySink
, "obj", method_obj
=method_obj
)
112 self
.assertIsInstance(query_method_obj
, Custom
)
113 self
.assertEqual(query_method_obj
.hola
, "hello")
116 def test_query_with_none_method_obj(self
):
117 class MySink(bt2
._UserSinkComponent
):
118 def _user_consume(self
):
122 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
123 nonlocal query_method_obj
124 query_method_obj
= method_obj
126 query_method_obj
= object()
127 bt2
.QueryExecutor(MySink
, "obj").query()
128 self
.assertIsNone(query_method_obj
)
131 def test_query_with_method_obj_non_python_comp_cls(self
):
132 plugin
= bt2
.find_plugin("text", find_in_user_dir
=False, find_in_sys_dir
=False)
133 assert plugin
is not None
134 cc
= plugin
.source_component_classes
["dmesg"]
135 assert cc
is not None
137 with self
.assertRaisesRegex(
139 re
.escape(r
"cannot pass a Python object to a non-Python component class"),
141 bt2
.QueryExecutor(cc
, "obj", method_obj
=object()).query()
143 def test_query_logging_level(self
):
144 class MySink(bt2
._UserSinkComponent
):
145 def _user_consume(self
):
149 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
150 nonlocal query_log_level
151 query_log_level
= priv_query_exec
.logging_level
153 query_log_level
= None
154 query_exec
= bt2
.QueryExecutor(MySink
, "obj", None)
155 query_exec
.logging_level
= bt2
.LoggingLevel
.INFO
157 self
.assertEqual(query_log_level
, bt2
.LoggingLevel
.INFO
)
160 def test_query_gen_error(self
):
161 class MySink(bt2
._UserSinkComponent
):
162 def _user_consume(self
):
166 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
169 with self
.assertRaises(bt2
._Error
) as ctx
:
170 bt2
.QueryExecutor(MySink
, "obj", [17, 23]).query()
173 self
.assertEqual(len(exc
), 3)
175 self
.assertIsInstance(cause
, bt2
._ComponentClassErrorCause
)
176 self
.assertIn("raise ValueError", cause
.message
)
177 self
.assertEqual(cause
.component_class_type
, bt2
.ComponentClassType
.SINK
)
178 self
.assertEqual(cause
.component_class_name
, "MySink")
180 def test_query_unknown_object(self
):
181 class MySink(bt2
._UserSinkComponent
):
182 def _user_consume(self
):
186 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
187 raise bt2
.UnknownObject
189 with self
.assertRaises(bt2
.UnknownObject
):
190 bt2
.QueryExecutor(MySink
, "obj", [17, 23]).query()
192 def test_query_logging_level_invalid_type(self
):
193 class MySink(bt2
._UserSinkComponent
):
194 def _user_consume(self
):
198 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
201 query_exec
= bt2
.QueryExecutor(MySink
, "obj", [17, 23])
203 with self
.assertRaises(TypeError):
204 query_exec
.logging_level
= "yeah"
206 def test_query_logging_level_invalid_value(self
):
207 class MySink(bt2
._UserSinkComponent
):
208 def _user_consume(self
):
212 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
215 query_exec
= bt2
.QueryExecutor(MySink
, "obj", [17, 23])
217 with self
.assertRaises(ValueError):
218 query_exec
.logging_level
= 12345
220 def test_query_try_again(self
):
221 class MySink(bt2
._UserSinkComponent
):
222 def _user_consume(self
):
226 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
229 with self
.assertRaises(bt2
.TryAgain
):
230 bt2
.QueryExecutor(MySink
, "obj", [17, 23]).query()
232 def test_query_add_interrupter(self
):
233 class MySink(bt2
._UserSinkComponent
):
234 def _user_consume(self
):
238 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
239 nonlocal interrupter2
240 test_self
.assertFalse(query_exec
.is_interrupted
)
242 test_self
.assertTrue(query_exec
.is_interrupted
)
244 test_self
.assertFalse(query_exec
.is_interrupted
)
246 interrupter1
= bt2
.Interrupter()
247 interrupter2
= bt2
.Interrupter()
249 query_exec
= bt2
.QueryExecutor(MySink
, "obj", [17, 23])
250 query_exec
.add_interrupter(interrupter1
)
251 query_exec
.add_interrupter(interrupter2
)
254 def test_query_interrupt(self
):
255 class MySink(bt2
._UserSinkComponent
):
256 def _user_consume(self
):
260 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
261 test_self
.assertFalse(query_exec
.is_interrupted
)
262 query_exec
.default_interrupter
.set()
263 test_self
.assertTrue(query_exec
.is_interrupted
)
266 query_exec
= bt2
.QueryExecutor(MySink
, "obj", [17, 23])
269 def test_query_priv_executor_invalid_after(self
):
270 class MySink(bt2
._UserSinkComponent
):
271 def _user_consume(self
):
275 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
276 nonlocal test_priv_query_exec
277 test_priv_query_exec
= priv_query_exec
279 test_priv_query_exec
= None
280 query_exec
= bt2
.QueryExecutor(MySink
, "obj", [17, 23])
282 assert test_priv_query_exec
is not None
284 with self
.assertRaises(RuntimeError):
285 test_priv_query_exec
.logging_level
287 del test_priv_query_exec
290 if __name__
== "__main__":