tests/bindings/python/bt2/test_query_executor.py

   1 # SPDX-License-Identifier: GPL-2.0-only
   2 #
   3 # Copyright (C) 2019 EfficiOS Inc.
   4 #
   5
   6 import unittest
   7 import bt2
   8 import re
   9
  10
  11 class QueryExecutorTestCase(unittest.TestCase):
  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
  21     def test_query(self):
  22         class MySink(bt2._UserSinkComponent):
  23             def _user_consume(self):
  24                 pass
  25
  26             @classmethod
  27             def _user_query(cls, priv_query_exec, obj, params, method_obj):
  28                 nonlocal query_params
  29                 query_params = params
  30                 return {'null': None, 'bt2': 'BT2'}
  31
  32         query_params = None
  33         params = {
  34             'array': ['coucou', 23, None],
  35             'other_map': {'yes': 'yeah', '19': 19, 'minus 1.5': -1.5},
  36             'null': None,
  37         }
  38
  39         res = bt2.QueryExecutor(MySink, 'obj', params).query()
  40         self.assertIs(type(res), bt2._MapValueConst)
  41         self.assertIs(type(res['bt2']), bt2._StringValueConst)
  42         self.assertEqual(query_params, params)
  43         self.assertEqual(res, {'null': None, 'bt2': 'BT2'})
  44         del query_params
  45
  46     def test_query_params_none(self):
  47         class MySink(bt2._UserSinkComponent):
  48             def _user_consume(self):
  49                 pass
  50
  51             @classmethod
  52             def _user_query(cls, priv_query_exec, obj, params, method_obj):
  53                 nonlocal query_params
  54                 query_params = params
  55
  56         query_params = 23
  57         bt2.QueryExecutor(MySink, 'obj', None).query()
  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
  67             def _user_query(cls, priv_query_exec, obj, params, method_obj):
  68                 nonlocal query_params
  69                 query_params = params
  70
  71         query_params = 23
  72         bt2.QueryExecutor(MySink, 'obj').query()
  73         self.assertIs(query_params, None)
  74         del query_params
  75
  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()
  88         bt2.QueryExecutor(MySink, 'obj', method_obj=method_obj).query()
  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()
 126         bt2.QueryExecutor(MySink, 'obj').query()
 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
 142     def test_query_logging_level(self):
 143         class MySink(bt2._UserSinkComponent):
 144             def _user_consume(self):
 145                 pass
 146
 147             @classmethod
 148             def _user_query(cls, priv_query_exec, obj, params, method_obj):
 149                 nonlocal query_log_level
 150                 query_log_level = priv_query_exec.logging_level
 151
 152         query_log_level = None
 153         query_exec = bt2.QueryExecutor(MySink, 'obj', None)
 154         query_exec.logging_level = bt2.LoggingLevel.INFO
 155         query_exec.query()
 156         self.assertEqual(query_log_level, bt2.LoggingLevel.INFO)
 157         del query_log_level
 158
 159     def test_query_gen_error(self):
 160         class MySink(bt2._UserSinkComponent):
 161             def _user_consume(self):
 162                 pass
 163
 164             @classmethod
 165             def _user_query(cls, priv_query_exec, obj, params, method_obj):
 166                 raise ValueError
 167
 168         with self.assertRaises(bt2._Error) as ctx:
 169             bt2.QueryExecutor(MySink, 'obj', [17, 23]).query()
 170
 171         exc = ctx.exception
 172         self.assertEqual(len(exc), 3)
 173         cause = exc[0]
 174         self.assertIsInstance(cause, bt2._ComponentClassErrorCause)
 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
 179     def test_query_unknown_object(self):
 180         class MySink(bt2._UserSinkComponent):
 181             def _user_consume(self):
 182                 pass
 183
 184             @classmethod
 185             def _user_query(cls, priv_query_exec, obj, params, method_obj):
 186                 raise bt2.UnknownObject
 187
 188         with self.assertRaises(bt2.UnknownObject):
 189             bt2.QueryExecutor(MySink, 'obj', [17, 23]).query()
 190
 191     def test_query_logging_level_invalid_type(self):
 192         class MySink(bt2._UserSinkComponent):
 193             def _user_consume(self):
 194                 pass
 195
 196             @classmethod
 197             def _user_query(cls, priv_query_exec, obj, params, method_obj):
 198                 pass
 199
 200         query_exec = bt2.QueryExecutor(MySink, 'obj', [17, 23])
 201
 202         with self.assertRaises(TypeError):
 203             query_exec.logging_level = 'yeah'
 204
 205     def test_query_logging_level_invalid_value(self):
 206         class MySink(bt2._UserSinkComponent):
 207             def _user_consume(self):
 208                 pass
 209
 210             @classmethod
 211             def _user_query(cls, priv_query_exec, obj, params, method_obj):
 212                 pass
 213
 214         query_exec = bt2.QueryExecutor(MySink, 'obj', [17, 23])
 215
 216         with self.assertRaises(ValueError):
 217             query_exec.logging_level = 12345
 218
 219     def test_query_try_again(self):
 220         class MySink(bt2._UserSinkComponent):
 221             def _user_consume(self):
 222                 pass
 223
 224             @classmethod
 225             def _user_query(cls, priv_query_exec, obj, params, method_obj):
 226                 raise bt2.TryAgain
 227
 228         with self.assertRaises(bt2.TryAgain):
 229             bt2.QueryExecutor(MySink, 'obj', [17, 23]).query()
 230
 231     def test_query_add_interrupter(self):
 232         class MySink(bt2._UserSinkComponent):
 233             def _user_consume(self):
 234                 pass
 235
 236             @classmethod
 237             def _user_query(cls, priv_query_exec, obj, params, method_obj):
 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
 248         query_exec = bt2.QueryExecutor(MySink, 'obj', [17, 23])
 249         query_exec.add_interrupter(interrupter1)
 250         query_exec.add_interrupter(interrupter2)
 251         query_exec.query()
 252
 253     def test_query_interrupt(self):
 254         class MySink(bt2._UserSinkComponent):
 255             def _user_consume(self):
 256                 pass
 257
 258             @classmethod
 259             def _user_query(cls, priv_query_exec, obj, params, method_obj):
 260                 test_self.assertFalse(query_exec.is_interrupted)
 261                 query_exec.default_interrupter.set()
 262                 test_self.assertTrue(query_exec.is_interrupted)
 263
 264         test_self = self
 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
 274             def _user_query(cls, priv_query_exec, obj, params, method_obj):
 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
 287
 288
 289 if __name__ == '__main__':
 290     unittest.main()