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