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
):
26 class MySink(bt2
._UserSinkComponent
):
27 def _user_consume(self
):
31 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
34 return {'null': None, 'bt2': 'BT2'}
38 'array': ['coucou', 23, None],
39 'other_map': {'yes': 'yeah', '19': 19, 'minus 1.5': -1.5},
43 res
= bt2
.QueryExecutor(MySink
, 'obj', params
).query()
44 self
.assertEqual(query_params
, params
)
45 self
.assertEqual(res
, {'null': None, 'bt2': 'BT2'})
48 def test_query_params_none(self
):
49 class MySink(bt2
._UserSinkComponent
):
50 def _user_consume(self
):
54 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
59 bt2
.QueryExecutor(MySink
, 'obj', None).query()
60 self
.assertIs(query_params
, None)
63 def test_query_no_params(self
):
64 class MySink(bt2
._UserSinkComponent
):
65 def _user_consume(self
):
69 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
74 bt2
.QueryExecutor(MySink
, 'obj').query()
75 self
.assertIs(query_params
, None)
78 def test_query_with_method_obj(self
):
79 class MySink(bt2
._UserSinkComponent
):
80 def _user_consume(self
):
84 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
85 nonlocal query_method_obj
86 query_method_obj
= method_obj
88 query_method_obj
= None
90 bt2
.QueryExecutor(MySink
, 'obj', method_obj
=method_obj
).query()
91 self
.assertIs(query_method_obj
, method_obj
)
94 def test_query_with_method_obj_del_ref(self
):
95 class MySink(bt2
._UserSinkComponent
):
96 def _user_consume(self
):
100 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
101 nonlocal query_method_obj
102 query_method_obj
= method_obj
107 query_method_obj
= None
108 method_obj
= Custom()
109 method_obj
.hola
= 'hello'
110 query_exec
= bt2
.QueryExecutor(MySink
, 'obj', method_obj
=method_obj
)
113 self
.assertIsInstance(query_method_obj
, Custom
)
114 self
.assertEqual(query_method_obj
.hola
, 'hello')
117 def test_query_with_none_method_obj(self
):
118 class MySink(bt2
._UserSinkComponent
):
119 def _user_consume(self
):
123 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
124 nonlocal query_method_obj
125 query_method_obj
= method_obj
127 query_method_obj
= object()
128 bt2
.QueryExecutor(MySink
, 'obj').query()
129 self
.assertIsNone(query_method_obj
)
132 def test_query_with_method_obj_non_python_comp_cls(self
):
133 plugin
= bt2
.find_plugin('text', find_in_user_dir
=False, find_in_sys_dir
=False)
134 assert plugin
is not None
135 cc
= plugin
.source_component_classes
['dmesg']
136 assert cc
is not None
138 with self
.assertRaisesRegex(
140 re
.escape(r
'cannot pass a Python object to a non-Python component class'),
142 bt2
.QueryExecutor(cc
, 'obj', method_obj
=object()).query()
144 def test_query_logging_level(self
):
145 class MySink(bt2
._UserSinkComponent
):
146 def _user_consume(self
):
150 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
151 nonlocal query_log_level
152 query_log_level
= priv_query_exec
.logging_level
154 query_log_level
= None
155 query_exec
= bt2
.QueryExecutor(MySink
, 'obj', None)
156 query_exec
.logging_level
= bt2
.LoggingLevel
.INFO
158 self
.assertEqual(query_log_level
, bt2
.LoggingLevel
.INFO
)
161 def test_query_gen_error(self
):
162 class MySink(bt2
._UserSinkComponent
):
163 def _user_consume(self
):
167 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
170 with self
.assertRaises(bt2
._Error
) as ctx
:
171 bt2
.QueryExecutor(MySink
, 'obj', [17, 23]).query()
174 self
.assertEqual(len(exc
), 3)
176 self
.assertIsInstance(cause
, bt2
._ComponentClassErrorCause
)
177 self
.assertIn('raise ValueError', cause
.message
)
178 self
.assertEqual(cause
.component_class_type
, bt2
.ComponentClassType
.SINK
)
179 self
.assertEqual(cause
.component_class_name
, 'MySink')
181 def test_query_unknown_object(self
):
182 class MySink(bt2
._UserSinkComponent
):
183 def _user_consume(self
):
187 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
188 raise bt2
.UnknownObject
190 with self
.assertRaises(bt2
.UnknownObject
):
191 bt2
.QueryExecutor(MySink
, 'obj', [17, 23]).query()
193 def test_query_logging_level_invalid_type(self
):
194 class MySink(bt2
._UserSinkComponent
):
195 def _user_consume(self
):
199 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
202 query_exec
= bt2
.QueryExecutor(MySink
, 'obj', [17, 23])
204 with self
.assertRaises(TypeError):
205 query_exec
.logging_level
= 'yeah'
207 def test_query_logging_level_invalid_value(self
):
208 class MySink(bt2
._UserSinkComponent
):
209 def _user_consume(self
):
213 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
216 query_exec
= bt2
.QueryExecutor(MySink
, 'obj', [17, 23])
218 with self
.assertRaises(ValueError):
219 query_exec
.logging_level
= 12345
221 def test_query_try_again(self
):
222 class MySink(bt2
._UserSinkComponent
):
223 def _user_consume(self
):
227 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
230 with self
.assertRaises(bt2
.TryAgain
):
231 bt2
.QueryExecutor(MySink
, 'obj', [17, 23]).query()
233 def test_query_add_interrupter(self
):
234 class MySink(bt2
._UserSinkComponent
):
235 def _user_consume(self
):
239 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
240 nonlocal interrupter2
241 test_self
.assertFalse(query_exec
.is_interrupted
)
243 test_self
.assertTrue(query_exec
.is_interrupted
)
245 test_self
.assertFalse(query_exec
.is_interrupted
)
247 interrupter1
= bt2
.Interrupter()
248 interrupter2
= bt2
.Interrupter()
250 query_exec
= bt2
.QueryExecutor(MySink
, 'obj', [17, 23])
251 query_exec
.add_interrupter(interrupter1
)
252 query_exec
.add_interrupter(interrupter2
)
255 def test_query_interrupt(self
):
256 class MySink(bt2
._UserSinkComponent
):
257 def _user_consume(self
):
261 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
262 test_self
.assertFalse(query_exec
.is_interrupted
)
263 query_exec
.interrupt()
264 test_self
.assertTrue(query_exec
.is_interrupted
)
267 query_exec
= bt2
.QueryExecutor(MySink
, 'obj', [17, 23])
270 def test_query_priv_executor_invalid_after(self
):
271 class MySink(bt2
._UserSinkComponent
):
272 def _user_consume(self
):
276 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
277 nonlocal test_priv_query_exec
278 test_priv_query_exec
= priv_query_exec
280 test_priv_query_exec
= None
281 query_exec
= bt2
.QueryExecutor(MySink
, 'obj', [17, 23])
283 assert test_priv_query_exec
is not None
285 with self
.assertRaises(RuntimeError):
286 test_priv_query_exec
.logging_level
288 del test_priv_query_exec