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.
26 class QueryExecutorTestCase(unittest
.TestCase
):
28 class MySink(bt2
._UserSinkComponent
):
29 def _user_consume(self
):
33 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
36 return {'null': None, 'bt2': 'BT2'}
40 'array': ['coucou', 23, None],
41 'other_map': {'yes': 'yeah', '19': 19, 'minus 1.5': -1.5},
45 res
= bt2
.QueryExecutor(MySink
, 'obj', params
).query()
46 self
.assertEqual(query_params
, params
)
47 self
.assertEqual(res
, {'null': None, 'bt2': 'BT2'})
50 def test_query_params_none(self
):
51 class MySink(bt2
._UserSinkComponent
):
52 def _user_consume(self
):
56 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
61 res
= bt2
.QueryExecutor(MySink
, 'obj', None).query()
62 self
.assertIs(query_params
, None)
65 def test_query_no_params(self
):
66 class MySink(bt2
._UserSinkComponent
):
67 def _user_consume(self
):
71 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
76 res
= bt2
.QueryExecutor(MySink
, 'obj').query()
77 self
.assertIs(query_params
, None)
80 def test_query_with_method_obj(self
):
81 class MySink(bt2
._UserSinkComponent
):
82 def _user_consume(self
):
86 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
87 nonlocal query_method_obj
88 query_method_obj
= method_obj
90 query_method_obj
= None
92 res
= bt2
.QueryExecutor(MySink
, 'obj', method_obj
=method_obj
).query()
93 self
.assertIs(query_method_obj
, method_obj
)
96 def test_query_with_method_obj_del_ref(self
):
97 class MySink(bt2
._UserSinkComponent
):
98 def _user_consume(self
):
102 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
103 nonlocal query_method_obj
104 query_method_obj
= method_obj
109 query_method_obj
= None
110 method_obj
= Custom()
111 method_obj
.hola
= 'hello'
112 query_exec
= bt2
.QueryExecutor(MySink
, 'obj', method_obj
=method_obj
)
115 self
.assertIsInstance(query_method_obj
, Custom
)
116 self
.assertEqual(query_method_obj
.hola
, 'hello')
119 def test_query_with_none_method_obj(self
):
120 class MySink(bt2
._UserSinkComponent
):
121 def _user_consume(self
):
125 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
126 nonlocal query_method_obj
127 query_method_obj
= method_obj
129 query_method_obj
= object()
130 res
= bt2
.QueryExecutor(MySink
, 'obj').query()
131 self
.assertIsNone(query_method_obj
)
134 def test_query_with_method_obj_non_python_comp_cls(self
):
135 plugin
= bt2
.find_plugin('text', find_in_user_dir
=False, find_in_sys_dir
=False)
136 assert plugin
is not None
137 cc
= plugin
.source_component_classes
['dmesg']
138 assert cc
is not None
140 with self
.assertRaisesRegex(
142 re
.escape(r
'cannot pass a Python object to a non-Python component class'),
144 bt2
.QueryExecutor(cc
, 'obj', method_obj
=object()).query()
146 def test_query_logging_level(self
):
147 class MySink(bt2
._UserSinkComponent
):
148 def _user_consume(self
):
152 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
153 nonlocal query_log_level
154 query_log_level
= priv_query_exec
.logging_level
156 query_log_level
= None
157 query_exec
= bt2
.QueryExecutor(MySink
, 'obj', None)
158 query_exec
.logging_level
= bt2
.LoggingLevel
.INFO
160 self
.assertEqual(query_log_level
, bt2
.LoggingLevel
.INFO
)
163 def test_query_gen_error(self
):
164 class MySink(bt2
._UserSinkComponent
):
165 def _user_consume(self
):
169 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
172 with self
.assertRaises(bt2
._Error
) as ctx
:
173 res
= bt2
.QueryExecutor(MySink
, 'obj', [17, 23]).query()
176 self
.assertEqual(len(exc
), 2)
178 self
.assertIsInstance(cause
, bt2
._ComponentClassErrorCause
)
179 self
.assertIn('raise ValueError', cause
.message
)
180 self
.assertEqual(cause
.component_class_type
, bt2
.ComponentClassType
.SINK
)
181 self
.assertEqual(cause
.component_class_name
, 'MySink')
183 def test_query_unknown_object(self
):
184 class MySink(bt2
._UserSinkComponent
):
185 def _user_consume(self
):
189 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
190 raise bt2
.UnknownObject
192 with self
.assertRaises(bt2
.UnknownObject
):
193 res
= bt2
.QueryExecutor(MySink
, 'obj', [17, 23]).query()
195 def test_query_logging_level_invalid_type(self
):
196 class MySink(bt2
._UserSinkComponent
):
197 def _user_consume(self
):
201 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
204 query_exec
= bt2
.QueryExecutor(MySink
, 'obj', [17, 23])
206 with self
.assertRaises(TypeError):
207 query_exec
.logging_level
= 'yeah'
209 def test_query_logging_level_invalid_value(self
):
210 class MySink(bt2
._UserSinkComponent
):
211 def _user_consume(self
):
215 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
218 query_exec
= bt2
.QueryExecutor(MySink
, 'obj', [17, 23])
220 with self
.assertRaises(ValueError):
221 query_exec
.logging_level
= 12345
223 def test_query_try_again(self
):
224 class MySink(bt2
._UserSinkComponent
):
225 def _user_consume(self
):
229 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
232 with self
.assertRaises(bt2
.TryAgain
):
233 res
= bt2
.QueryExecutor(MySink
, 'obj', [17, 23]).query()
235 def test_query_add_interrupter(self
):
236 class MySink(bt2
._UserSinkComponent
):
237 def _user_consume(self
):
241 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
242 nonlocal interrupter2
243 test_self
.assertFalse(query_exec
.is_interrupted
)
245 test_self
.assertTrue(query_exec
.is_interrupted
)
247 test_self
.assertFalse(query_exec
.is_interrupted
)
249 interrupter1
= bt2
.Interrupter()
250 interrupter2
= bt2
.Interrupter()
252 query_exec
= bt2
.QueryExecutor(MySink
, 'obj', [17, 23])
253 query_exec
.add_interrupter(interrupter1
)
254 query_exec
.add_interrupter(interrupter2
)
257 def test_query_interrupt(self
):
258 class MySink(bt2
._UserSinkComponent
):
259 def _user_consume(self
):
263 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
264 test_self
.assertFalse(query_exec
.is_interrupted
)
265 query_exec
.interrupt()
266 test_self
.assertTrue(query_exec
.is_interrupted
)
269 query_exec
= bt2
.QueryExecutor(MySink
, 'obj', [17, 23])
272 def test_query_priv_executor_invalid_after(self
):
273 class MySink(bt2
._UserSinkComponent
):
274 def _user_consume(self
):
278 def _user_query(cls
, priv_query_exec
, obj
, params
, method_obj
):
279 nonlocal test_priv_query_exec
280 test_priv_query_exec
= priv_query_exec
282 test_priv_query_exec
= None
283 query_exec
= bt2
.QueryExecutor(MySink
, 'obj', [17, 23])
285 assert test_priv_query_exec
is not None
287 with self
.assertRaises(RuntimeError):
288 test_priv_query_exec
.logging_level
290 del test_priv_query_exec