Commit | Line | Data |
---|---|---|
0235b0db | 1 | # SPDX-License-Identifier: GPL-2.0-only |
d2d857a8 MJ |
2 | # |
3 | # Copyright (C) 2019 EfficiOS Inc. | |
4 | # | |
d2d857a8 | 5 | |
c7eee084 | 6 | import unittest |
c7eee084 | 7 | import bt2 |
7c14d641 | 8 | import re |
c7eee084 PP |
9 | |
10 | ||
11 | class QueryExecutorTestCase(unittest.TestCase): | |
88d1a0b7 SM |
12 | def test_default_interrupter(self): |
13 | class MySink(bt2._UserSinkComponent): | |
14 | def _user_consume(self): | |
15 | pass | |
16 | ||
f5567ea8 | 17 | query_exec = bt2.QueryExecutor(MySink, "obj") |
88d1a0b7 SM |
18 | interrupter = query_exec.default_interrupter |
19 | self.assertIs(type(interrupter), bt2.Interrupter) | |
20 | ||
c7eee084 PP |
21 | def test_query(self): |
22 | class MySink(bt2._UserSinkComponent): | |
6a91742b | 23 | def _user_consume(self): |
a01b452b SM |
24 | pass |
25 | ||
c7eee084 | 26 | @classmethod |
7c14d641 | 27 | def _user_query(cls, priv_query_exec, obj, params, method_obj): |
c7eee084 PP |
28 | nonlocal query_params |
29 | query_params = params | |
f5567ea8 | 30 | return {"null": None, "bt2": "BT2"} |
c7eee084 PP |
31 | |
32 | query_params = None | |
33 | params = { | |
f5567ea8 FD |
34 | "array": ["coucou", 23, None], |
35 | "other_map": {"yes": "yeah", "19": 19, "minus 1.5": -1.5}, | |
36 | "null": None, | |
c7eee084 PP |
37 | } |
38 | ||
f5567ea8 | 39 | res = bt2.QueryExecutor(MySink, "obj", params).query() |
e42e1587 | 40 | self.assertIs(type(res), bt2._MapValueConst) |
f5567ea8 | 41 | self.assertIs(type(res["bt2"]), bt2._StringValueConst) |
c7eee084 | 42 | self.assertEqual(query_params, params) |
f5567ea8 | 43 | self.assertEqual(res, {"null": None, "bt2": "BT2"}) |
c7eee084 PP |
44 | del query_params |
45 | ||
46 | def test_query_params_none(self): | |
47 | class MySink(bt2._UserSinkComponent): | |
6a91742b | 48 | def _user_consume(self): |
a01b452b SM |
49 | pass |
50 | ||
c7eee084 | 51 | @classmethod |
7c14d641 | 52 | def _user_query(cls, priv_query_exec, obj, params, method_obj): |
c7eee084 PP |
53 | nonlocal query_params |
54 | query_params = params | |
55 | ||
56 | query_params = 23 | |
f5567ea8 | 57 | bt2.QueryExecutor(MySink, "obj", None).query() |
3c729b9a PP |
58 | self.assertIs(query_params, None) |
59 | del query_params | |
60 | ||
61 | def test_query_no_params(self): | |
62 | class MySink(bt2._UserSinkComponent): | |
63 | def _user_consume(self): | |
64 | pass | |
65 | ||
66 | @classmethod | |
7c14d641 | 67 | def _user_query(cls, priv_query_exec, obj, params, method_obj): |
3c729b9a PP |
68 | nonlocal query_params |
69 | query_params = params | |
70 | ||
71 | query_params = 23 | |
f5567ea8 | 72 | bt2.QueryExecutor(MySink, "obj").query() |
3c729b9a | 73 | self.assertIs(query_params, None) |
c7eee084 PP |
74 | del query_params |
75 | ||
7c14d641 PP |
76 | def test_query_with_method_obj(self): |
77 | class MySink(bt2._UserSinkComponent): | |
78 | def _user_consume(self): | |
79 | pass | |
80 | ||
81 | @classmethod | |
82 | def _user_query(cls, priv_query_exec, obj, params, method_obj): | |
83 | nonlocal query_method_obj | |
84 | query_method_obj = method_obj | |
85 | ||
86 | query_method_obj = None | |
87 | method_obj = object() | |
f5567ea8 | 88 | bt2.QueryExecutor(MySink, "obj", method_obj=method_obj).query() |
7c14d641 PP |
89 | self.assertIs(query_method_obj, method_obj) |
90 | del query_method_obj | |
91 | ||
92 | def test_query_with_method_obj_del_ref(self): | |
93 | class MySink(bt2._UserSinkComponent): | |
94 | def _user_consume(self): | |
95 | pass | |
96 | ||
97 | @classmethod | |
98 | def _user_query(cls, priv_query_exec, obj, params, method_obj): | |
99 | nonlocal query_method_obj | |
100 | query_method_obj = method_obj | |
101 | ||
102 | class Custom: | |
103 | pass | |
104 | ||
105 | query_method_obj = None | |
106 | method_obj = Custom() | |
f5567ea8 FD |
107 | method_obj.hola = "hello" |
108 | query_exec = bt2.QueryExecutor(MySink, "obj", method_obj=method_obj) | |
7c14d641 PP |
109 | del method_obj |
110 | query_exec.query() | |
111 | self.assertIsInstance(query_method_obj, Custom) | |
f5567ea8 | 112 | self.assertEqual(query_method_obj.hola, "hello") |
7c14d641 PP |
113 | del query_method_obj |
114 | ||
115 | def test_query_with_none_method_obj(self): | |
116 | class MySink(bt2._UserSinkComponent): | |
117 | def _user_consume(self): | |
118 | pass | |
119 | ||
120 | @classmethod | |
121 | def _user_query(cls, priv_query_exec, obj, params, method_obj): | |
122 | nonlocal query_method_obj | |
123 | query_method_obj = method_obj | |
124 | ||
125 | query_method_obj = object() | |
f5567ea8 | 126 | bt2.QueryExecutor(MySink, "obj").query() |
7c14d641 PP |
127 | self.assertIsNone(query_method_obj) |
128 | del query_method_obj | |
129 | ||
130 | def test_query_with_method_obj_non_python_comp_cls(self): | |
f5567ea8 | 131 | plugin = bt2.find_plugin("text", find_in_user_dir=False, find_in_sys_dir=False) |
7c14d641 | 132 | assert plugin is not None |
f5567ea8 | 133 | cc = plugin.source_component_classes["dmesg"] |
7c14d641 PP |
134 | assert cc is not None |
135 | ||
136 | with self.assertRaisesRegex( | |
137 | ValueError, | |
f5567ea8 | 138 | re.escape(r"cannot pass a Python object to a non-Python component class"), |
7c14d641 | 139 | ): |
f5567ea8 | 140 | bt2.QueryExecutor(cc, "obj", method_obj=object()).query() |
7c14d641 | 141 | |
761e1890 PP |
142 | def test_query_logging_level(self): |
143 | class MySink(bt2._UserSinkComponent): | |
6a91742b | 144 | def _user_consume(self): |
a01b452b SM |
145 | pass |
146 | ||
761e1890 | 147 | @classmethod |
7c14d641 | 148 | def _user_query(cls, priv_query_exec, obj, params, method_obj): |
761e1890 | 149 | nonlocal query_log_level |
3c729b9a | 150 | query_log_level = priv_query_exec.logging_level |
761e1890 PP |
151 | |
152 | query_log_level = None | |
f5567ea8 | 153 | query_exec = bt2.QueryExecutor(MySink, "obj", None) |
3c729b9a PP |
154 | query_exec.logging_level = bt2.LoggingLevel.INFO |
155 | query_exec.query() | |
761e1890 PP |
156 | self.assertEqual(query_log_level, bt2.LoggingLevel.INFO) |
157 | del query_log_level | |
158 | ||
c7eee084 PP |
159 | def test_query_gen_error(self): |
160 | class MySink(bt2._UserSinkComponent): | |
6a91742b | 161 | def _user_consume(self): |
a01b452b SM |
162 | pass |
163 | ||
c7eee084 | 164 | @classmethod |
7c14d641 | 165 | def _user_query(cls, priv_query_exec, obj, params, method_obj): |
c7eee084 PP |
166 | raise ValueError |
167 | ||
694c792b | 168 | with self.assertRaises(bt2._Error) as ctx: |
f5567ea8 | 169 | bt2.QueryExecutor(MySink, "obj", [17, 23]).query() |
c7eee084 | 170 | |
ce4923b0 | 171 | exc = ctx.exception |
561e7049 | 172 | self.assertEqual(len(exc), 3) |
ce4923b0 | 173 | cause = exc[0] |
3fb99a22 | 174 | self.assertIsInstance(cause, bt2._ComponentClassErrorCause) |
f5567ea8 | 175 | self.assertIn("raise ValueError", cause.message) |
ce4923b0 | 176 | self.assertEqual(cause.component_class_type, bt2.ComponentClassType.SINK) |
f5567ea8 | 177 | self.assertEqual(cause.component_class_name, "MySink") |
ce4923b0 | 178 | |
76b6c2f7 | 179 | def test_query_unknown_object(self): |
c7eee084 | 180 | class MySink(bt2._UserSinkComponent): |
6a91742b | 181 | def _user_consume(self): |
a01b452b SM |
182 | pass |
183 | ||
c7eee084 | 184 | @classmethod |
7c14d641 | 185 | def _user_query(cls, priv_query_exec, obj, params, method_obj): |
76b6c2f7 | 186 | raise bt2.UnknownObject |
c7eee084 | 187 | |
76b6c2f7 | 188 | with self.assertRaises(bt2.UnknownObject): |
f5567ea8 | 189 | bt2.QueryExecutor(MySink, "obj", [17, 23]).query() |
c7eee084 | 190 | |
761e1890 PP |
191 | def test_query_logging_level_invalid_type(self): |
192 | class MySink(bt2._UserSinkComponent): | |
6a91742b | 193 | def _user_consume(self): |
a01b452b SM |
194 | pass |
195 | ||
761e1890 | 196 | @classmethod |
7c14d641 | 197 | def _user_query(cls, priv_query_exec, obj, params, method_obj): |
761e1890 PP |
198 | pass |
199 | ||
f5567ea8 | 200 | query_exec = bt2.QueryExecutor(MySink, "obj", [17, 23]) |
3c729b9a | 201 | |
761e1890 | 202 | with self.assertRaises(TypeError): |
f5567ea8 | 203 | query_exec.logging_level = "yeah" |
761e1890 PP |
204 | |
205 | def test_query_logging_level_invalid_value(self): | |
206 | class MySink(bt2._UserSinkComponent): | |
6a91742b | 207 | def _user_consume(self): |
a01b452b SM |
208 | pass |
209 | ||
761e1890 | 210 | @classmethod |
7c14d641 | 211 | def _user_query(cls, priv_query_exec, obj, params, method_obj): |
761e1890 PP |
212 | pass |
213 | ||
f5567ea8 | 214 | query_exec = bt2.QueryExecutor(MySink, "obj", [17, 23]) |
3c729b9a | 215 | |
761e1890 | 216 | with self.assertRaises(ValueError): |
3c729b9a | 217 | query_exec.logging_level = 12345 |
761e1890 | 218 | |
c7eee084 PP |
219 | def test_query_try_again(self): |
220 | class MySink(bt2._UserSinkComponent): | |
6a91742b | 221 | def _user_consume(self): |
a01b452b SM |
222 | pass |
223 | ||
c7eee084 | 224 | @classmethod |
7c14d641 | 225 | def _user_query(cls, priv_query_exec, obj, params, method_obj): |
c7eee084 PP |
226 | raise bt2.TryAgain |
227 | ||
228 | with self.assertRaises(bt2.TryAgain): | |
f5567ea8 | 229 | bt2.QueryExecutor(MySink, "obj", [17, 23]).query() |
c7eee084 | 230 | |
9b4f9b42 PP |
231 | def test_query_add_interrupter(self): |
232 | class MySink(bt2._UserSinkComponent): | |
6a91742b | 233 | def _user_consume(self): |
9b4f9b42 PP |
234 | pass |
235 | ||
236 | @classmethod | |
7c14d641 | 237 | def _user_query(cls, priv_query_exec, obj, params, method_obj): |
9b4f9b42 PP |
238 | nonlocal interrupter2 |
239 | test_self.assertFalse(query_exec.is_interrupted) | |
240 | interrupter2.set() | |
241 | test_self.assertTrue(query_exec.is_interrupted) | |
242 | interrupter2.reset() | |
243 | test_self.assertFalse(query_exec.is_interrupted) | |
244 | ||
245 | interrupter1 = bt2.Interrupter() | |
246 | interrupter2 = bt2.Interrupter() | |
247 | test_self = self | |
f5567ea8 | 248 | query_exec = bt2.QueryExecutor(MySink, "obj", [17, 23]) |
9b4f9b42 PP |
249 | query_exec.add_interrupter(interrupter1) |
250 | query_exec.add_interrupter(interrupter2) | |
3c729b9a | 251 | query_exec.query() |
c7eee084 | 252 | |
9b4f9b42 | 253 | def test_query_interrupt(self): |
c7eee084 | 254 | class MySink(bt2._UserSinkComponent): |
6a91742b | 255 | def _user_consume(self): |
a01b452b SM |
256 | pass |
257 | ||
c7eee084 | 258 | @classmethod |
7c14d641 | 259 | def _user_query(cls, priv_query_exec, obj, params, method_obj): |
9b4f9b42 | 260 | test_self.assertFalse(query_exec.is_interrupted) |
88d1a0b7 | 261 | query_exec.default_interrupter.set() |
9b4f9b42 | 262 | test_self.assertTrue(query_exec.is_interrupted) |
c7eee084 | 263 | |
9b4f9b42 | 264 | test_self = self |
f5567ea8 | 265 | query_exec = bt2.QueryExecutor(MySink, "obj", [17, 23]) |
3c729b9a PP |
266 | query_exec.query() |
267 | ||
268 | def test_query_priv_executor_invalid_after(self): | |
269 | class MySink(bt2._UserSinkComponent): | |
270 | def _user_consume(self): | |
271 | pass | |
272 | ||
273 | @classmethod | |
7c14d641 | 274 | def _user_query(cls, priv_query_exec, obj, params, method_obj): |
3c729b9a PP |
275 | nonlocal test_priv_query_exec |
276 | test_priv_query_exec = priv_query_exec | |
277 | ||
278 | test_priv_query_exec = None | |
f5567ea8 | 279 | query_exec = bt2.QueryExecutor(MySink, "obj", [17, 23]) |
3c729b9a PP |
280 | query_exec.query() |
281 | assert test_priv_query_exec is not None | |
282 | ||
283 | with self.assertRaises(RuntimeError): | |
284 | test_priv_query_exec.logging_level | |
285 | ||
286 | del test_priv_query_exec | |
d14ddbba SM |
287 | |
288 | ||
f5567ea8 | 289 | if __name__ == "__main__": |
d14ddbba | 290 | unittest.main() |