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 | ||
17 | query_exec = bt2.QueryExecutor(MySink, 'obj') | |
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 | |
cfbd7cf3 | 30 | return {'null': None, 'bt2': 'BT2'} |
c7eee084 PP |
31 | |
32 | query_params = None | |
33 | params = { | |
34 | 'array': ['coucou', 23, None], | |
cfbd7cf3 | 35 | 'other_map': {'yes': 'yeah', '19': 19, 'minus 1.5': -1.5}, |
c7eee084 PP |
36 | 'null': None, |
37 | } | |
38 | ||
3c729b9a | 39 | res = bt2.QueryExecutor(MySink, 'obj', params).query() |
e42e1587 FD |
40 | self.assertIs(type(res), bt2._MapValueConst) |
41 | self.assertIs(type(res['bt2']), bt2._StringValueConst) | |
c7eee084 | 42 | self.assertEqual(query_params, params) |
cfbd7cf3 | 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 | |
082db648 | 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 | |
082db648 | 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() | |
082db648 | 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() | |
107 | method_obj.hola = 'hello' | |
108 | query_exec = bt2.QueryExecutor(MySink, 'obj', method_obj=method_obj) | |
109 | del method_obj | |
110 | query_exec.query() | |
111 | self.assertIsInstance(query_method_obj, Custom) | |
112 | self.assertEqual(query_method_obj.hola, 'hello') | |
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() | |
082db648 | 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): | |
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 | |
135 | ||
136 | with self.assertRaisesRegex( | |
137 | ValueError, | |
138 | re.escape(r'cannot pass a Python object to a non-Python component class'), | |
139 | ): | |
140 | bt2.QueryExecutor(cc, 'obj', method_obj=object()).query() | |
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 | |
3c729b9a PP |
153 | query_exec = bt2.QueryExecutor(MySink, 'obj', None) |
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: |
082db648 | 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) |
ce4923b0 SM |
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') | |
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): |
082db648 | 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 | ||
3c729b9a PP |
200 | query_exec = bt2.QueryExecutor(MySink, 'obj', [17, 23]) |
201 | ||
761e1890 | 202 | with self.assertRaises(TypeError): |
3c729b9a | 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 | ||
3c729b9a PP |
214 | query_exec = bt2.QueryExecutor(MySink, 'obj', [17, 23]) |
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): | |
082db648 | 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 | |
3c729b9a | 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 |
3c729b9a PP |
265 | query_exec = bt2.QueryExecutor(MySink, 'obj', [17, 23]) |
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 | |
279 | query_exec = bt2.QueryExecutor(MySink, 'obj', [17, 23]) | |
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 | ||
289 | if __name__ == '__main__': | |
290 | unittest.main() |