2 # Copyright (C) 2019 EfficiOS Inc.
4 # This program is free software; you can redistribute it and/or
5 # modify it under the terms of the GNU General Public License
6 # as published by the Free Software Foundation; only version 2
9 # This program is distributed in the hope that it will be useful,
10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 # GNU General Public License for more details.
14 # You should have received a copy of the GNU General Public License
15 # along with this program; if not, write to the Free Software
16 # Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
24 class QueryExecutorTestCase(unittest
.TestCase
):
25 def test_default_interrupter(self
):
26 class MySink(bt2
._UserSinkComponent
):
27 def _user_consume(self
):
30 query_exec
= bt2
.QueryExecutor(MySink
, 'obj')
31 interrupter
= query_exec
.default_interrupter
32 self
.assertIs(type(interrupter
), bt2
.Interrupter
)
35 class MySink(bt2
._UserSinkComponent
):
36 def _user_consume(self
):
40 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
43 return {'null': None, 'bt2': 'BT2'}
47 'array': ['coucou', 23, None],
48 'other_map': {'yes': 'yeah', '19': 19, 'minus 1.5': -1.5},
52 res
= bt2
.QueryExecutor(MySink
, 'obj', params
).query()
53 self
.assertIs(type(res
), bt2
._MapValueConst
)
54 self
.assertIs(type(res
['bt2']), bt2
._StringValueConst
)
55 self
.assertEqual(query_params
, params
)
56 self
.assertEqual(res
, {'null': None, 'bt2': 'BT2'})
59 def test_query_params_none(self
):
60 class MySink(bt2
._UserSinkComponent
):
61 def _user_consume(self
):
65 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
70 bt2
.QueryExecutor(MySink
, 'obj', None).query()
71 self
.assertIs(query_params
, None)
74 def test_query_no_params(self
):
75 class MySink(bt2
._UserSinkComponent
):
76 def _user_consume(self
):
80 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
85 bt2
.QueryExecutor(MySink
, 'obj').query()
86 self
.assertIs(query_params
, None)
89 def test_query_with_method_obj(self
):
90 class MySink(bt2
._UserSinkComponent
):
91 def _user_consume(self
):
95 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
96 nonlocal query_method_obj
97 query_method_obj
= method_obj
99 query_method_obj
= None
100 method_obj
= object()
101 bt2
.QueryExecutor(MySink
, 'obj', method_obj
=method_obj
).query()
102 self
.assertIs(query_method_obj
, method_obj
)
105 def test_query_with_method_obj_del_ref(self
):
106 class MySink(bt2
._UserSinkComponent
):
107 def _user_consume(self
):
111 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
112 nonlocal query_method_obj
113 query_method_obj
= method_obj
118 query_method_obj
= None
119 method_obj
= Custom()
120 method_obj
.hola
= 'hello'
121 query_exec
= bt2
.QueryExecutor(MySink
, 'obj', method_obj
=method_obj
)
124 self
.assertIsInstance(query_method_obj
, Custom
)
125 self
.assertEqual(query_method_obj
.hola
, 'hello')
128 def test_query_with_none_method_obj(self
):
129 class MySink(bt2
._UserSinkComponent
):
130 def _user_consume(self
):
134 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
135 nonlocal query_method_obj
136 query_method_obj
= method_obj
138 query_method_obj
= object()
139 bt2
.QueryExecutor(MySink
, 'obj').query()
140 self
.assertIsNone(query_method_obj
)
143 def test_query_with_method_obj_non_python_comp_cls(self
):
144 plugin
= bt2
.find_plugin('text', find_in_user_dir
=False, find_in_sys_dir
=False)
145 assert plugin
is not None
146 cc
= plugin
.source_component_classes
['dmesg']
147 assert cc
is not None
149 with self
.assertRaisesRegex(
151 re
.escape(r
'cannot pass a Python object to a non-Python component class'),
153 bt2
.QueryExecutor(cc
, 'obj', method_obj
=object()).query()
155 def test_query_logging_level(self
):
156 class MySink(bt2
._UserSinkComponent
):
157 def _user_consume(self
):
161 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
162 nonlocal query_log_level
163 query_log_level
= priv_query_exec
.logging_level
165 query_log_level
= None
166 query_exec
= bt2
.QueryExecutor(MySink
, 'obj', None)
167 query_exec
.logging_level
= bt2
.LoggingLevel
.INFO
169 self
.assertEqual(query_log_level
, bt2
.LoggingLevel
.INFO
)
172 def test_query_gen_error(self
):
173 class MySink(bt2
._UserSinkComponent
):
174 def _user_consume(self
):
178 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
181 with self
.assertRaises(bt2
._Error
) as ctx
:
182 bt2
.QueryExecutor(MySink
, 'obj', [17, 23]).query()
185 self
.assertEqual(len(exc
), 3)
187 self
.assertIsInstance(cause
, bt2
._ComponentClassErrorCause
)
188 self
.assertIn('raise ValueError', cause
.message
)
189 self
.assertEqual(cause
.component_class_type
, bt2
.ComponentClassType
.SINK
)
190 self
.assertEqual(cause
.component_class_name
, 'MySink')
192 def test_query_unknown_object(self
):
193 class MySink(bt2
._UserSinkComponent
):
194 def _user_consume(self
):
198 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
199 raise bt2
.UnknownObject
201 with self
.assertRaises(bt2
.UnknownObject
):
202 bt2
.QueryExecutor(MySink
, 'obj', [17, 23]).query()
204 def test_query_logging_level_invalid_type(self
):
205 class MySink(bt2
._UserSinkComponent
):
206 def _user_consume(self
):
210 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
213 query_exec
= bt2
.QueryExecutor(MySink
, 'obj', [17, 23])
215 with self
.assertRaises(TypeError):
216 query_exec
.logging_level
= 'yeah'
218 def test_query_logging_level_invalid_value(self
):
219 class MySink(bt2
._UserSinkComponent
):
220 def _user_consume(self
):
224 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
227 query_exec
= bt2
.QueryExecutor(MySink
, 'obj', [17, 23])
229 with self
.assertRaises(ValueError):
230 query_exec
.logging_level
= 12345
232 def test_query_try_again(self
):
233 class MySink(bt2
._UserSinkComponent
):
234 def _user_consume(self
):
238 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
241 with self
.assertRaises(bt2
.TryAgain
):
242 bt2
.QueryExecutor(MySink
, 'obj', [17, 23]).query()
244 def test_query_add_interrupter(self
):
245 class MySink(bt2
._UserSinkComponent
):
246 def _user_consume(self
):
250 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
251 nonlocal interrupter2
252 test_self
.assertFalse(query_exec
.is_interrupted
)
254 test_self
.assertTrue(query_exec
.is_interrupted
)
256 test_self
.assertFalse(query_exec
.is_interrupted
)
258 interrupter1
= bt2
.Interrupter()
259 interrupter2
= bt2
.Interrupter()
261 query_exec
= bt2
.QueryExecutor(MySink
, 'obj', [17, 23])
262 query_exec
.add_interrupter(interrupter1
)
263 query_exec
.add_interrupter(interrupter2
)
266 def test_query_interrupt(self
):
267 class MySink(bt2
._UserSinkComponent
):
268 def _user_consume(self
):
272 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
273 test_self
.assertFalse(query_exec
.is_interrupted
)
274 query_exec
.default_interrupter
.set()
275 test_self
.assertTrue(query_exec
.is_interrupted
)
278 query_exec
= bt2
.QueryExecutor(MySink
, 'obj', [17, 23])
281 def test_query_priv_executor_invalid_after(self
):
282 class MySink(bt2
._UserSinkComponent
):
283 def _user_consume(self
):
287 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
288 nonlocal test_priv_query_exec
289 test_priv_query_exec
= priv_query_exec
291 test_priv_query_exec
= None
292 query_exec
= bt2
.QueryExecutor(MySink
, 'obj', [17, 23])
294 assert test_priv_query_exec
is not None
296 with self
.assertRaises(RuntimeError):
297 test_priv_query_exec
.logging_level
299 del test_priv_query_exec
302 if __name__
== '__main__':