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 _consume(self):
  29                 pass
  30
  31             def _graph_is_configured(self):
  32                 pass
  33
  34             @classmethod
  35             def _query(cls, query_exec, obj, params, log_level):
  36                 nonlocal query_params
  37                 query_params = params
  38                 return {'null': None, 'bt2': 'BT2'}
  39
  40         query_params = None
  41         params = {
  42             'array': ['coucou', 23, None],
  43             'other_map': {'yes': 'yeah', '19': 19, 'minus 1.5': -1.5},
  44             'null': None,
  45         }
  46
  47         res = bt2.QueryExecutor().query(MySink, 'obj', params)
  48         self.assertEqual(query_params, params)
  49         self.assertEqual(res, {'null': None, 'bt2': 'BT2'})
  50         del query_params
  51
  52     def test_query_params_none(self):
  53         class MySink(bt2._UserSinkComponent):
  54             def _consume(self):
  55                 pass
  56
  57             def _graph_is_configured(self):
  58                 pass
  59
  60             @classmethod
  61             def _query(cls, query_exec, obj, params, log_level):
  62                 nonlocal query_params
  63                 query_params = params
  64
  65         query_params = 23
  66         res = bt2.QueryExecutor().query(MySink, 'obj', None)
  67         self.assertEqual(query_params, None)
  68         del query_params
  69
  70     def test_query_logging_level(self):
  71         class MySink(bt2._UserSinkComponent):
  72             def _consume(self):
  73                 pass
  74
  75             def _graph_is_configured(self):
  76                 pass
  77
  78             @classmethod
  79             def _query(cls, query_exec, obj, params, log_level):
  80                 nonlocal query_log_level
  81                 query_log_level = log_level
  82
  83         query_log_level = None
  84         res = bt2.QueryExecutor().query(MySink, 'obj', None, bt2.LoggingLevel.INFO)
  85         self.assertEqual(query_log_level, bt2.LoggingLevel.INFO)
  86         del query_log_level
  87
  88     def test_query_gen_error(self):
  89         class MySink(bt2._UserSinkComponent):
  90             def _consume(self):
  91                 pass
  92
  93             def _graph_is_configured(self):
  94                 pass
  95
  96             @classmethod
  97             def _query(cls, query_exec, obj, params, log_level):
  98                 raise ValueError
  99
 100         with self.assertRaises(bt2._Error) as ctx:
 101             res = bt2.QueryExecutor().query(MySink, 'obj', [17, 23])
 102
 103         exc = ctx.exception
 104         self.assertEqual(len(exc), 1)
 105         cause = exc[0]
 106         self.assertIsInstance(cause, bt2.error._ComponentClassErrorCause)
 107         self.assertIn('raise ValueError', cause.message)
 108         self.assertEqual(cause.component_class_type, bt2.ComponentClassType.SINK)
 109         self.assertEqual(cause.component_class_name, 'MySink')
 110
 111     def test_query_invalid_object(self):
 112         class MySink(bt2._UserSinkComponent):
 113             def _consume(self):
 114                 pass
 115
 116             def _graph_is_configured(self):
 117                 pass
 118
 119             @classmethod
 120             def _query(cls, query_exec, obj, params, log_level):
 121                 raise bt2.InvalidObject
 122
 123         with self.assertRaises(bt2.InvalidObject):
 124             res = bt2.QueryExecutor().query(MySink, 'obj', [17, 23])
 125
 126     def test_query_logging_level_invalid_type(self):
 127         class MySink(bt2._UserSinkComponent):
 128             def _consume(self):
 129                 pass
 130
 131             def _graph_is_configured(self):
 132                 pass
 133
 134             @classmethod
 135             def _query(cls, query_exec, obj, params, log_level):
 136                 pass
 137
 138         with self.assertRaises(TypeError):
 139             res = bt2.QueryExecutor().query(MySink, 'obj', [17, 23], 'yeah')
 140
 141     def test_query_logging_level_invalid_value(self):
 142         class MySink(bt2._UserSinkComponent):
 143             def _consume(self):
 144                 pass
 145
 146             def _graph_is_configured(self):
 147                 pass
 148
 149             @classmethod
 150             def _query(cls, query_exec, obj, params, log_level):
 151                 pass
 152
 153         with self.assertRaises(ValueError):
 154             res = bt2.QueryExecutor().query(MySink, 'obj', [17, 23], 12345)
 155
 156     def test_query_invalid_params(self):
 157         class MySink(bt2._UserSinkComponent):
 158             def _consume(self):
 159                 pass
 160
 161             def _graph_is_configured(self):
 162                 pass
 163
 164             @classmethod
 165             def _query(cls, query_exec, obj, params, log_level):
 166                 raise bt2.InvalidParams
 167
 168         with self.assertRaises(bt2.InvalidParams):
 169             res = bt2.QueryExecutor().query(MySink, 'obj', [17, 23])
 170
 171     def test_query_try_again(self):
 172         class MySink(bt2._UserSinkComponent):
 173             def _consume(self):
 174                 pass
 175
 176             def _graph_is_configured(self):
 177                 pass
 178
 179             @classmethod
 180             def _query(cls, query_exec, obj, params, log_level):
 181                 raise bt2.TryAgain
 182
 183         with self.assertRaises(bt2.TryAgain):
 184             res = bt2.QueryExecutor().query(MySink, 'obj', [17, 23])
 185
 186     def test_query_add_interrupter(self):
 187         class MySink(bt2._UserSinkComponent):
 188             def _consume(self):
 189                 pass
 190
 191             def _graph_is_configured(self):
 192                 pass
 193
 194             @classmethod
 195             def _query(cls, query_exec, obj, params, log_level):
 196                 nonlocal interrupter2
 197                 test_self.assertFalse(query_exec.is_interrupted)
 198                 interrupter2.set()
 199                 test_self.assertTrue(query_exec.is_interrupted)
 200                 interrupter2.reset()
 201                 test_self.assertFalse(query_exec.is_interrupted)
 202
 203         interrupter1 = bt2.Interrupter()
 204         interrupter2 = bt2.Interrupter()
 205         test_self = self
 206         query_exec = bt2.QueryExecutor()
 207         query_exec.add_interrupter(interrupter1)
 208         query_exec.add_interrupter(interrupter2)
 209         query_exec.query(MySink, 'obj', [17, 23])
 210
 211     def test_query_interrupt(self):
 212         class MySink(bt2._UserSinkComponent):
 213             def _consume(self):
 214                 pass
 215
 216             def _graph_is_configured(self):
 217                 pass
 218
 219             @classmethod
 220             def _query(cls, query_exec, obj, params, log_level):
 221                 test_self.assertFalse(query_exec.is_interrupted)
 222                 query_exec.interrupt()
 223                 test_self.assertTrue(query_exec.is_interrupted)
 224
 225         test_self = self
 226         query_exec = bt2.QueryExecutor()
 227         query_exec.query(MySink, 'obj', [17, 23])