tests/bindings/python/bt2/test_query_executor.py

   1 from bt2 import values
   2 import unittest
   3 import copy
   4 import bt2
   5
   6
   7 class QueryExecutorTestCase(unittest.TestCase):
   8     def test_query(self):
   9         class MySink(bt2._UserSinkComponent):
  10             def _consume(self):
  11                 pass
  12
  13             @classmethod
  14             def _query(cls, query_exec, obj, params):
  15                 nonlocal query_params
  16                 query_params = params
  17                 return {
  18                     'null': None,
  19                     'bt2': 'BT2',
  20                 }
  21
  22         query_params = None
  23         params = {
  24             'array': ['coucou', 23, None],
  25             'other_map': {
  26                 'yes': 'yeah',
  27                 '19': 19,
  28                 'minus 1.5': -1.5,
  29             },
  30             'null': None,
  31         }
  32
  33         res = bt2.QueryExecutor().query(MySink, 'obj', params)
  34         self.assertEqual(query_params, params)
  35         self.assertEqual(res, {
  36             'null': None,
  37             'bt2': 'BT2',
  38         })
  39         del query_params
  40
  41     def test_query_params_none(self):
  42         class MySink(bt2._UserSinkComponent):
  43             def _consume(self):
  44                 pass
  45
  46             @classmethod
  47             def _query(cls, query_exec, obj, params):
  48                 nonlocal query_params
  49                 query_params = params
  50
  51         query_params = 23
  52         res = bt2.QueryExecutor().query(MySink, 'obj', None)
  53         self.assertEqual(query_params, None)
  54         del query_params
  55
  56     def test_query_gen_error(self):
  57         class MySink(bt2._UserSinkComponent):
  58             def _consume(self):
  59                 pass
  60
  61             @classmethod
  62             def _query(cls, query_exec, obj, params):
  63                 raise ValueError
  64
  65         with self.assertRaises(bt2.Error):
  66             res = bt2.QueryExecutor().query(MySink, 'obj', [17, 23])
  67
  68     def test_query_invalid_object(self):
  69         class MySink(bt2._UserSinkComponent):
  70             def _consume(self):
  71                 pass
  72
  73             @classmethod
  74             def _query(cls, query_exec, obj, params):
  75                 raise bt2.InvalidQueryObject
  76
  77         with self.assertRaises(bt2.InvalidQueryObject):
  78             res = bt2.QueryExecutor().query(MySink, 'obj', [17, 23])
  79
  80     def test_query_invalid_params(self):
  81         class MySink(bt2._UserSinkComponent):
  82             def _consume(self):
  83                 pass
  84
  85             @classmethod
  86             def _query(cls, query_exec, obj, params):
  87                 raise bt2.InvalidQueryParams
  88
  89         with self.assertRaises(bt2.InvalidQueryParams):
  90             res = bt2.QueryExecutor().query(MySink, 'obj', [17, 23])
  91
  92     def test_query_try_again(self):
  93         class MySink(bt2._UserSinkComponent):
  94             def _consume(self):
  95                 pass
  96
  97             @classmethod
  98             def _query(cls, query_exec, obj, params):
  99                 raise bt2.TryAgain
 100
 101         with self.assertRaises(bt2.TryAgain):
 102             res = bt2.QueryExecutor().query(MySink, 'obj', [17, 23])
 103
 104     def test_cancel_no_query(self):
 105         query_exec = bt2.QueryExecutor()
 106         self.assertFalse(query_exec.is_canceled)
 107         query_exec.cancel()
 108         self.assertTrue(query_exec.is_canceled)
 109
 110     def test_query_canceled(self):
 111         class MySink(bt2._UserSinkComponent):
 112             def _consume(self):
 113                 pass
 114
 115             @classmethod
 116             def _query(cls, query_exec, obj, params):
 117                 raise bt2.TryAgain
 118
 119         query_exec = bt2.QueryExecutor()
 120         query_exec.cancel()
 121
 122         with self.assertRaises(bt2.QueryExecutorCanceled):
 123             res = query_exec.query(MySink, 'obj', [17, 23])
 124
 125     def test_eq(self):
 126         query_exec = bt2.QueryExecutor()
 127         self.assertEqual(query_exec, query_exec)
 128
 129     def test_eq_invalid(self):
 130         query_exec = bt2.QueryExecutor()
 131         self.assertNotEqual(query_exec, 23)