tests/bindings/python/bt2/test_query_executor.py

   1 #
   2 # Copyright (C) 2019 EfficiOS Inc.
   3 #
   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
   7 # of the License.
   8 #
   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.
  13 #
  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.
  17 #
  18
  19 import unittest
  20 import bt2
  21 import re
  22
  23
  24 class QueryExecutorTestCase(unittest.TestCase):
  25     def test_default_interrupter(self):
  26         class MySink(bt2._UserSinkComponent):
  27             def _user_consume(self):
  28                 pass
  29
  30         query_exec = bt2.QueryExecutor(MySink, 'obj')
  31         interrupter = query_exec.default_interrupter
  32         self.assertIs(type(interrupter), bt2.Interrupter)
  33
  34     def test_query(self):
  35         class MySink(bt2._UserSinkComponent):
  36             def _user_consume(self):
  37                 pass
  38
  39             @classmethod
  40             def _user_query(cls, priv_query_exec, obj, params, method_obj):
  41                 nonlocal query_params
  42                 query_params = params
  43                 return {'null': None, 'bt2': 'BT2'}
  44
  45         query_params = None
  46         params = {
  47             'array': ['coucou', 23, None],
  48             'other_map': {'yes': 'yeah', '19': 19, 'minus 1.5': -1.5},
  49             'null': None,
  50         }
  51
  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'})
  57         del query_params
  58
  59     def test_query_params_none(self):
  60         class MySink(bt2._UserSinkComponent):
  61             def _user_consume(self):
  62                 pass
  63
  64             @classmethod
  65             def _user_query(cls, priv_query_exec, obj, params, method_obj):
  66                 nonlocal query_params
  67                 query_params = params
  68
  69         query_params = 23
  70         bt2.QueryExecutor(MySink, 'obj', None).query()
  71         self.assertIs(query_params, None)
  72         del query_params
  73
  74     def test_query_no_params(self):
  75         class MySink(bt2._UserSinkComponent):
  76             def _user_consume(self):
  77                 pass
  78
  79             @classmethod
  80             def _user_query(cls, priv_query_exec, obj, params, method_obj):
  81                 nonlocal query_params
  82                 query_params = params
  83
  84         query_params = 23
  85         bt2.QueryExecutor(MySink, 'obj').query()
  86         self.assertIs(query_params, None)
  87         del query_params
  88
  89     def test_query_with_method_obj(self):
  90         class MySink(bt2._UserSinkComponent):
  91             def _user_consume(self):
  92                 pass
  93
  94             @classmethod
  95             def _user_query(cls, priv_query_exec, obj, params, method_obj):
  96                 nonlocal query_method_obj
  97                 query_method_obj = method_obj
  98
  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)
 103         del query_method_obj
 104
 105     def test_query_with_method_obj_del_ref(self):
 106         class MySink(bt2._UserSinkComponent):
 107             def _user_consume(self):
 108                 pass
 109
 110             @classmethod
 111             def _user_query(cls, priv_query_exec, obj, params, method_obj):
 112                 nonlocal query_method_obj
 113                 query_method_obj = method_obj
 114
 115         class Custom:
 116             pass
 117
 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)
 122         del method_obj
 123         query_exec.query()
 124         self.assertIsInstance(query_method_obj, Custom)
 125         self.assertEqual(query_method_obj.hola, 'hello')
 126         del query_method_obj
 127
 128     def test_query_with_none_method_obj(self):
 129         class MySink(bt2._UserSinkComponent):
 130             def _user_consume(self):
 131                 pass
 132
 133             @classmethod
 134             def _user_query(cls, priv_query_exec, obj, params, method_obj):
 135                 nonlocal query_method_obj
 136                 query_method_obj = method_obj
 137
 138         query_method_obj = object()
 139         bt2.QueryExecutor(MySink, 'obj').query()
 140         self.assertIsNone(query_method_obj)
 141         del query_method_obj
 142
 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
 148
 149         with self.assertRaisesRegex(
 150             ValueError,
 151             re.escape(r'cannot pass a Python object to a non-Python component class'),
 152         ):
 153             bt2.QueryExecutor(cc, 'obj', method_obj=object()).query()
 154
 155     def test_query_logging_level(self):
 156         class MySink(bt2._UserSinkComponent):
 157             def _user_consume(self):
 158                 pass
 159
 160             @classmethod
 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
 164
 165         query_log_level = None
 166         query_exec = bt2.QueryExecutor(MySink, 'obj', None)
 167         query_exec.logging_level = bt2.LoggingLevel.INFO
 168         query_exec.query()
 169         self.assertEqual(query_log_level, bt2.LoggingLevel.INFO)
 170         del query_log_level
 171
 172     def test_query_gen_error(self):
 173         class MySink(bt2._UserSinkComponent):
 174             def _user_consume(self):
 175                 pass
 176
 177             @classmethod
 178             def _user_query(cls, priv_query_exec, obj, params, method_obj):
 179                 raise ValueError
 180
 181         with self.assertRaises(bt2._Error) as ctx:
 182             bt2.QueryExecutor(MySink, 'obj', [17, 23]).query()
 183
 184         exc = ctx.exception
 185         self.assertEqual(len(exc), 3)
 186         cause = exc[0]
 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')
 191
 192     def test_query_unknown_object(self):
 193         class MySink(bt2._UserSinkComponent):
 194             def _user_consume(self):
 195                 pass
 196
 197             @classmethod
 198             def _user_query(cls, priv_query_exec, obj, params, method_obj):
 199                 raise bt2.UnknownObject
 200
 201         with self.assertRaises(bt2.UnknownObject):
 202             bt2.QueryExecutor(MySink, 'obj', [17, 23]).query()
 203
 204     def test_query_logging_level_invalid_type(self):
 205         class MySink(bt2._UserSinkComponent):
 206             def _user_consume(self):
 207                 pass
 208
 209             @classmethod
 210             def _user_query(cls, priv_query_exec, obj, params, method_obj):
 211                 pass
 212
 213         query_exec = bt2.QueryExecutor(MySink, 'obj', [17, 23])
 214
 215         with self.assertRaises(TypeError):
 216             query_exec.logging_level = 'yeah'
 217
 218     def test_query_logging_level_invalid_value(self):
 219         class MySink(bt2._UserSinkComponent):
 220             def _user_consume(self):
 221                 pass
 222
 223             @classmethod
 224             def _user_query(cls, priv_query_exec, obj, params, method_obj):
 225                 pass
 226
 227         query_exec = bt2.QueryExecutor(MySink, 'obj', [17, 23])
 228
 229         with self.assertRaises(ValueError):
 230             query_exec.logging_level = 12345
 231
 232     def test_query_try_again(self):
 233         class MySink(bt2._UserSinkComponent):
 234             def _user_consume(self):
 235                 pass
 236
 237             @classmethod
 238             def _user_query(cls, priv_query_exec, obj, params, method_obj):
 239                 raise bt2.TryAgain
 240
 241         with self.assertRaises(bt2.TryAgain):
 242             bt2.QueryExecutor(MySink, 'obj', [17, 23]).query()
 243
 244     def test_query_add_interrupter(self):
 245         class MySink(bt2._UserSinkComponent):
 246             def _user_consume(self):
 247                 pass
 248
 249             @classmethod
 250             def _user_query(cls, priv_query_exec, obj, params, method_obj):
 251                 nonlocal interrupter2
 252                 test_self.assertFalse(query_exec.is_interrupted)
 253                 interrupter2.set()
 254                 test_self.assertTrue(query_exec.is_interrupted)
 255                 interrupter2.reset()
 256                 test_self.assertFalse(query_exec.is_interrupted)
 257
 258         interrupter1 = bt2.Interrupter()
 259         interrupter2 = bt2.Interrupter()
 260         test_self = self
 261         query_exec = bt2.QueryExecutor(MySink, 'obj', [17, 23])
 262         query_exec.add_interrupter(interrupter1)
 263         query_exec.add_interrupter(interrupter2)
 264         query_exec.query()
 265
 266     def test_query_interrupt(self):
 267         class MySink(bt2._UserSinkComponent):
 268             def _user_consume(self):
 269                 pass
 270
 271             @classmethod
 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)
 276
 277         test_self = self
 278         query_exec = bt2.QueryExecutor(MySink, 'obj', [17, 23])
 279         query_exec.query()
 280
 281     def test_query_priv_executor_invalid_after(self):
 282         class MySink(bt2._UserSinkComponent):
 283             def _user_consume(self):
 284                 pass
 285
 286             @classmethod
 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
 290
 291         test_priv_query_exec = None
 292         query_exec = bt2.QueryExecutor(MySink, 'obj', [17, 23])
 293         query_exec.query()
 294         assert test_priv_query_exec is not None
 295
 296         with self.assertRaises(RuntimeError):
 297             test_priv_query_exec.logging_level
 298
 299         del test_priv_query_exec
 300
 301
 302 if __name__ == '__main__':
 303     unittest.main()