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_query(self):
  26         class MySink(bt2._UserSinkComponent):
  27             def _user_consume(self):
  28                 pass
  29
  30             @classmethod
  31             def _user_query(cls, priv_query_exec, obj, params, method_obj):
  32                 nonlocal query_params
  33                 query_params = params
  34                 return {'null': None, 'bt2': 'BT2'}
  35
  36         query_params = None
  37         params = {
  38             'array': ['coucou', 23, None],
  39             'other_map': {'yes': 'yeah', '19': 19, 'minus 1.5': -1.5},
  40             'null': None,
  41         }
  42
  43         res = bt2.QueryExecutor(MySink, 'obj', params).query()
  44         self.assertIs(type(res), bt2._MapValueConst)
  45         self.assertIs(type(res['bt2']), bt2._StringValueConst)
  46         self.assertEqual(query_params, params)
  47         self.assertEqual(res, {'null': None, 'bt2': 'BT2'})
  48         del query_params
  49
  50     def test_query_params_none(self):
  51         class MySink(bt2._UserSinkComponent):
  52             def _user_consume(self):
  53                 pass
  54
  55             @classmethod
  56             def _user_query(cls, priv_query_exec, obj, params, method_obj):
  57                 nonlocal query_params
  58                 query_params = params
  59
  60         query_params = 23
  61         bt2.QueryExecutor(MySink, 'obj', None).query()
  62         self.assertIs(query_params, None)
  63         del query_params
  64
  65     def test_query_no_params(self):
  66         class MySink(bt2._UserSinkComponent):
  67             def _user_consume(self):
  68                 pass
  69
  70             @classmethod
  71             def _user_query(cls, priv_query_exec, obj, params, method_obj):
  72                 nonlocal query_params
  73                 query_params = params
  74
  75         query_params = 23
  76         bt2.QueryExecutor(MySink, 'obj').query()
  77         self.assertIs(query_params, None)
  78         del query_params
  79
  80     def test_query_with_method_obj(self):
  81         class MySink(bt2._UserSinkComponent):
  82             def _user_consume(self):
  83                 pass
  84
  85             @classmethod
  86             def _user_query(cls, priv_query_exec, obj, params, method_obj):
  87                 nonlocal query_method_obj
  88                 query_method_obj = method_obj
  89
  90         query_method_obj = None
  91         method_obj = object()
  92         bt2.QueryExecutor(MySink, 'obj', method_obj=method_obj).query()
  93         self.assertIs(query_method_obj, method_obj)
  94         del query_method_obj
  95
  96     def test_query_with_method_obj_del_ref(self):
  97         class MySink(bt2._UserSinkComponent):
  98             def _user_consume(self):
  99                 pass
 100
 101             @classmethod
 102             def _user_query(cls, priv_query_exec, obj, params, method_obj):
 103                 nonlocal query_method_obj
 104                 query_method_obj = method_obj
 105
 106         class Custom:
 107             pass
 108
 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)
 113         del method_obj
 114         query_exec.query()
 115         self.assertIsInstance(query_method_obj, Custom)
 116         self.assertEqual(query_method_obj.hola, 'hello')
 117         del query_method_obj
 118
 119     def test_query_with_none_method_obj(self):
 120         class MySink(bt2._UserSinkComponent):
 121             def _user_consume(self):
 122                 pass
 123
 124             @classmethod
 125             def _user_query(cls, priv_query_exec, obj, params, method_obj):
 126                 nonlocal query_method_obj
 127                 query_method_obj = method_obj
 128
 129         query_method_obj = object()
 130         bt2.QueryExecutor(MySink, 'obj').query()
 131         self.assertIsNone(query_method_obj)
 132         del query_method_obj
 133
 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
 139
 140         with self.assertRaisesRegex(
 141             ValueError,
 142             re.escape(r'cannot pass a Python object to a non-Python component class'),
 143         ):
 144             bt2.QueryExecutor(cc, 'obj', method_obj=object()).query()
 145
 146     def test_query_logging_level(self):
 147         class MySink(bt2._UserSinkComponent):
 148             def _user_consume(self):
 149                 pass
 150
 151             @classmethod
 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
 155
 156         query_log_level = None
 157         query_exec = bt2.QueryExecutor(MySink, 'obj', None)
 158         query_exec.logging_level = bt2.LoggingLevel.INFO
 159         query_exec.query()
 160         self.assertEqual(query_log_level, bt2.LoggingLevel.INFO)
 161         del query_log_level
 162
 163     def test_query_gen_error(self):
 164         class MySink(bt2._UserSinkComponent):
 165             def _user_consume(self):
 166                 pass
 167
 168             @classmethod
 169             def _user_query(cls, priv_query_exec, obj, params, method_obj):
 170                 raise ValueError
 171
 172         with self.assertRaises(bt2._Error) as ctx:
 173             bt2.QueryExecutor(MySink, 'obj', [17, 23]).query()
 174
 175         exc = ctx.exception
 176         self.assertEqual(len(exc), 3)
 177         cause = exc[0]
 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')
 182
 183     def test_query_unknown_object(self):
 184         class MySink(bt2._UserSinkComponent):
 185             def _user_consume(self):
 186                 pass
 187
 188             @classmethod
 189             def _user_query(cls, priv_query_exec, obj, params, method_obj):
 190                 raise bt2.UnknownObject
 191
 192         with self.assertRaises(bt2.UnknownObject):
 193             bt2.QueryExecutor(MySink, 'obj', [17, 23]).query()
 194
 195     def test_query_logging_level_invalid_type(self):
 196         class MySink(bt2._UserSinkComponent):
 197             def _user_consume(self):
 198                 pass
 199
 200             @classmethod
 201             def _user_query(cls, priv_query_exec, obj, params, method_obj):
 202                 pass
 203
 204         query_exec = bt2.QueryExecutor(MySink, 'obj', [17, 23])
 205
 206         with self.assertRaises(TypeError):
 207             query_exec.logging_level = 'yeah'
 208
 209     def test_query_logging_level_invalid_value(self):
 210         class MySink(bt2._UserSinkComponent):
 211             def _user_consume(self):
 212                 pass
 213
 214             @classmethod
 215             def _user_query(cls, priv_query_exec, obj, params, method_obj):
 216                 pass
 217
 218         query_exec = bt2.QueryExecutor(MySink, 'obj', [17, 23])
 219
 220         with self.assertRaises(ValueError):
 221             query_exec.logging_level = 12345
 222
 223     def test_query_try_again(self):
 224         class MySink(bt2._UserSinkComponent):
 225             def _user_consume(self):
 226                 pass
 227
 228             @classmethod
 229             def _user_query(cls, priv_query_exec, obj, params, method_obj):
 230                 raise bt2.TryAgain
 231
 232         with self.assertRaises(bt2.TryAgain):
 233             bt2.QueryExecutor(MySink, 'obj', [17, 23]).query()
 234
 235     def test_query_add_interrupter(self):
 236         class MySink(bt2._UserSinkComponent):
 237             def _user_consume(self):
 238                 pass
 239
 240             @classmethod
 241             def _user_query(cls, priv_query_exec, obj, params, method_obj):
 242                 nonlocal interrupter2
 243                 test_self.assertFalse(query_exec.is_interrupted)
 244                 interrupter2.set()
 245                 test_self.assertTrue(query_exec.is_interrupted)
 246                 interrupter2.reset()
 247                 test_self.assertFalse(query_exec.is_interrupted)
 248
 249         interrupter1 = bt2.Interrupter()
 250         interrupter2 = bt2.Interrupter()
 251         test_self = self
 252         query_exec = bt2.QueryExecutor(MySink, 'obj', [17, 23])
 253         query_exec.add_interrupter(interrupter1)
 254         query_exec.add_interrupter(interrupter2)
 255         query_exec.query()
 256
 257     def test_query_interrupt(self):
 258         class MySink(bt2._UserSinkComponent):
 259             def _user_consume(self):
 260                 pass
 261
 262             @classmethod
 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)
 267
 268         test_self = self
 269         query_exec = bt2.QueryExecutor(MySink, 'obj', [17, 23])
 270         query_exec.query()
 271
 272     def test_query_priv_executor_invalid_after(self):
 273         class MySink(bt2._UserSinkComponent):
 274             def _user_consume(self):
 275                 pass
 276
 277             @classmethod
 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
 281
 282         test_priv_query_exec = None
 283         query_exec = bt2.QueryExecutor(MySink, 'obj', [17, 23])
 284         query_exec.query()
 285         assert test_priv_query_exec is not None
 286
 287         with self.assertRaises(RuntimeError):
 288             test_priv_query_exec.logging_level
 289
 290         del test_priv_query_exec
 291
 292
 293 if __name__ == '__main__':
 294     unittest.main()