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             @classmethod
  32             def _query(cls, query_exec, obj, params, log_level):
  33                 nonlocal query_params
  34                 query_params = params
  35                 return {
  36                     'null': None,
  37                     'bt2': 'BT2',
  38                 }
  39
  40         query_params = None
  41         params = {
  42             'array': ['coucou', 23, None],
  43             'other_map': {
  44                 'yes': 'yeah',
  45                 '19': 19,
  46                 'minus 1.5': -1.5,
  47             },
  48             'null': None,
  49         }
  50
  51         res = bt2.QueryExecutor().query(MySink, 'obj', params)
  52         self.assertEqual(query_params, params)
  53         self.assertEqual(res, {
  54             'null': None,
  55             'bt2': 'BT2',
  56         })
  57         del query_params
  58
  59     def test_query_params_none(self):
  60         class MySink(bt2._UserSinkComponent):
  61             def _consume(self):
  62                 pass
  63
  64             @classmethod
  65             def _query(cls, query_exec, obj, params, log_level):
  66                 nonlocal query_params
  67                 query_params = params
  68
  69         query_params = 23
  70         res = bt2.QueryExecutor().query(MySink, 'obj', None)
  71         self.assertEqual(query_params, None)
  72         del query_params
  73
  74     def test_query_logging_level(self):
  75         class MySink(bt2._UserSinkComponent):
  76             def _consume(self):
  77                 pass
  78
  79             @classmethod
  80             def _query(cls, query_exec, obj, params, log_level):
  81                 nonlocal query_log_level
  82                 query_log_level = log_level
  83
  84         query_log_level = None
  85         res = bt2.QueryExecutor().query(MySink, 'obj', None,
  86                                         bt2.LoggingLevel.INFO)
  87         self.assertEqual(query_log_level, bt2.LoggingLevel.INFO)
  88         del query_log_level
  89
  90     def test_query_gen_error(self):
  91         class MySink(bt2._UserSinkComponent):
  92             def _consume(self):
  93                 pass
  94
  95             @classmethod
  96             def _query(cls, query_exec, obj, params, log_level):
  97                 raise ValueError
  98
  99         with self.assertRaises(bt2.Error):
 100             res = bt2.QueryExecutor().query(MySink, 'obj', [17, 23])
 101
 102     def test_query_invalid_object(self):
 103         class MySink(bt2._UserSinkComponent):
 104             def _consume(self):
 105                 pass
 106
 107             @classmethod
 108             def _query(cls, query_exec, obj, params, log_level):
 109                 raise bt2.InvalidObject
 110
 111         with self.assertRaises(bt2.InvalidObject):
 112             res = bt2.QueryExecutor().query(MySink, 'obj', [17, 23])
 113
 114     def test_query_logging_level_invalid_type(self):
 115         class MySink(bt2._UserSinkComponent):
 116             def _consume(self):
 117                 pass
 118
 119             @classmethod
 120             def _query(cls, query_exec, obj, params, log_level):
 121                 pass
 122
 123         with self.assertRaises(TypeError):
 124             res = bt2.QueryExecutor().query(MySink, 'obj', [17, 23], 'yeah')
 125
 126     def test_query_logging_level_invalid_value(self):
 127         class MySink(bt2._UserSinkComponent):
 128             def _consume(self):
 129                 pass
 130
 131             @classmethod
 132             def _query(cls, query_exec, obj, params, log_level):
 133                 pass
 134
 135         with self.assertRaises(ValueError):
 136             res = bt2.QueryExecutor().query(MySink, 'obj', [17, 23], 12345)
 137
 138     def test_query_invalid_params(self):
 139         class MySink(bt2._UserSinkComponent):
 140             def _consume(self):
 141                 pass
 142
 143             @classmethod
 144             def _query(cls, query_exec, obj, params, log_level):
 145                 raise bt2.InvalidParams
 146
 147         with self.assertRaises(bt2.InvalidParams):
 148             res = bt2.QueryExecutor().query(MySink, 'obj', [17, 23])
 149
 150     def test_query_try_again(self):
 151         class MySink(bt2._UserSinkComponent):
 152             def _consume(self):
 153                 pass
 154
 155             @classmethod
 156             def _query(cls, query_exec, obj, params, log_level):
 157                 raise bt2.TryAgain
 158
 159         with self.assertRaises(bt2.TryAgain):
 160             res = bt2.QueryExecutor().query(MySink, 'obj', [17, 23])
 161
 162     def test_cancel_no_query(self):
 163         query_exec = bt2.QueryExecutor()
 164         self.assertFalse(query_exec.is_canceled)
 165         query_exec.cancel()
 166         self.assertTrue(query_exec.is_canceled)
 167
 168     def test_query_canceled(self):
 169         class MySink(bt2._UserSinkComponent):
 170             def _consume(self):
 171                 pass
 172
 173             @classmethod
 174             def _query(cls, query_exec, obj, params, log_level):
 175                 raise bt2.TryAgain
 176
 177         query_exec = bt2.QueryExecutor()
 178         query_exec.cancel()
 179
 180         with self.assertRaises(bt2.Canceled):
 181             res = query_exec.query(MySink, 'obj', [17, 23])