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