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 from bt2 import value
  20 import unittest
  21 import copy
  22 import bt2
  23
  24
  25 class QueryExecutorTestCase(unittest.TestCase):
  26     def test_query(self):
  27         class MySink(bt2._UserSinkComponent):
  28             def _user_consume(self):
  29                 pass
  30
  31             @classmethod
  32             def _user_query(cls, priv_query_exec, obj, params):
  33                 nonlocal query_params
  34                 query_params = params
  35                 return {'null': None, 'bt2': 'BT2'}
  36
  37         query_params = None
  38         params = {
  39             'array': ['coucou', 23, None],
  40             'other_map': {'yes': 'yeah', '19': 19, 'minus 1.5': -1.5},
  41             'null': None,
  42         }
  43
  44         res = bt2.QueryExecutor(MySink, 'obj', params).query()
  45         self.assertEqual(query_params, params)
  46         self.assertEqual(res, {'null': None, 'bt2': 'BT2'})
  47         del query_params
  48
  49     def test_query_params_none(self):
  50         class MySink(bt2._UserSinkComponent):
  51             def _user_consume(self):
  52                 pass
  53
  54             @classmethod
  55             def _user_query(cls, priv_query_exec, obj, params):
  56                 nonlocal query_params
  57                 query_params = params
  58
  59         query_params = 23
  60         res = bt2.QueryExecutor(MySink, 'obj', None).query()
  61         self.assertIs(query_params, None)
  62         del query_params
  63
  64     def test_query_no_params(self):
  65         class MySink(bt2._UserSinkComponent):
  66             def _user_consume(self):
  67                 pass
  68
  69             @classmethod
  70             def _user_query(cls, priv_query_exec, obj, params):
  71                 nonlocal query_params
  72                 query_params = params
  73
  74         query_params = 23
  75         res = bt2.QueryExecutor(MySink, 'obj').query()
  76         self.assertIs(query_params, None)
  77         del query_params
  78
  79     def test_query_logging_level(self):
  80         class MySink(bt2._UserSinkComponent):
  81             def _user_consume(self):
  82                 pass
  83
  84             @classmethod
  85             def _user_query(cls, priv_query_exec, obj, params):
  86                 nonlocal query_log_level
  87                 query_log_level = priv_query_exec.logging_level
  88
  89         query_log_level = None
  90         query_exec = bt2.QueryExecutor(MySink, 'obj', None)
  91         query_exec.logging_level = bt2.LoggingLevel.INFO
  92         query_exec.query()
  93         self.assertEqual(query_log_level, bt2.LoggingLevel.INFO)
  94         del query_log_level
  95
  96     def test_query_gen_error(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):
 103                 raise ValueError
 104
 105         with self.assertRaises(bt2._Error) as ctx:
 106             res = bt2.QueryExecutor(MySink, 'obj', [17, 23]).query()
 107
 108         exc = ctx.exception
 109         self.assertEqual(len(exc), 2)
 110         cause = exc[0]
 111         self.assertIsInstance(cause, bt2._ComponentClassErrorCause)
 112         self.assertIn('raise ValueError', cause.message)
 113         self.assertEqual(cause.component_class_type, bt2.ComponentClassType.SINK)
 114         self.assertEqual(cause.component_class_name, 'MySink')
 115
 116     def test_query_unknown_object(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):
 123                 raise bt2.UnknownObject
 124
 125         with self.assertRaises(bt2.UnknownObject):
 126             res = bt2.QueryExecutor(MySink, 'obj', [17, 23]).query()
 127
 128     def test_query_logging_level_invalid_type(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):
 135                 pass
 136
 137         query_exec = bt2.QueryExecutor(MySink, 'obj', [17, 23])
 138
 139         with self.assertRaises(TypeError):
 140             query_exec.logging_level = 'yeah'
 141
 142     def test_query_logging_level_invalid_value(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):
 149                 pass
 150
 151         query_exec = bt2.QueryExecutor(MySink, 'obj', [17, 23])
 152
 153         with self.assertRaises(ValueError):
 154             query_exec.logging_level = 12345
 155
 156     def test_query_try_again(self):
 157         class MySink(bt2._UserSinkComponent):
 158             def _user_consume(self):
 159                 pass
 160
 161             @classmethod
 162             def _user_query(cls, priv_query_exec, obj, params):
 163                 raise bt2.TryAgain
 164
 165         with self.assertRaises(bt2.TryAgain):
 166             res = bt2.QueryExecutor(MySink, 'obj', [17, 23]).query()
 167
 168     def test_query_add_interrupter(self):
 169         class MySink(bt2._UserSinkComponent):
 170             def _user_consume(self):
 171                 pass
 172
 173             @classmethod
 174             def _user_query(cls, priv_query_exec, obj, params):
 175                 nonlocal interrupter2
 176                 test_self.assertFalse(query_exec.is_interrupted)
 177                 interrupter2.set()
 178                 test_self.assertTrue(query_exec.is_interrupted)
 179                 interrupter2.reset()
 180                 test_self.assertFalse(query_exec.is_interrupted)
 181
 182         interrupter1 = bt2.Interrupter()
 183         interrupter2 = bt2.Interrupter()
 184         test_self = self
 185         query_exec = bt2.QueryExecutor(MySink, 'obj', [17, 23])
 186         query_exec.add_interrupter(interrupter1)
 187         query_exec.add_interrupter(interrupter2)
 188         query_exec.query()
 189
 190     def test_query_interrupt(self):
 191         class MySink(bt2._UserSinkComponent):
 192             def _user_consume(self):
 193                 pass
 194
 195             @classmethod
 196             def _user_query(cls, priv_query_exec, obj, params):
 197                 test_self.assertFalse(query_exec.is_interrupted)
 198                 query_exec.interrupt()
 199                 test_self.assertTrue(query_exec.is_interrupted)
 200
 201         test_self = self
 202         query_exec = bt2.QueryExecutor(MySink, 'obj', [17, 23])
 203         query_exec.query()
 204
 205     def test_query_priv_executor_invalid_after(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):
 212                 nonlocal test_priv_query_exec
 213                 test_priv_query_exec = priv_query_exec
 214
 215         test_priv_query_exec = None
 216         query_exec = bt2.QueryExecutor(MySink, 'obj', [17, 23])
 217         query_exec.query()
 218         assert test_priv_query_exec is not None
 219
 220         with self.assertRaises(RuntimeError):
 221             test_priv_query_exec.logging_level
 222
 223         del test_priv_query_exec