tests/bindings/python/bt2/test_component_class.py

   1 from bt2 import value
   2 import unittest
   3 import bt2
   4
   5
   6 class UserComponentClassTestCase(unittest.TestCase):
   7     def _test_no_init(self, cls):
   8         with self.assertRaises(bt2.Error):
   9             cls()
  10
  11     def test_no_init_source(self):
  12         class MyIter(bt2._UserMessageIterator):
  13             def __next__(self):
  14                 raise bt2.Stop
  15
  16         class MySource(bt2._UserSourceComponent,
  17                        message_iterator_class=MyIter):
  18             pass
  19
  20         self._test_no_init(MySource)
  21
  22     def test_no_init_filter(self):
  23         class MyIter(bt2._UserMessageIterator):
  24             def __next__(self):
  25                 raise bt2.Stop
  26
  27         class MyFilter(bt2._UserFilterComponent,
  28                        message_iterator_class=MyIter):
  29             pass
  30
  31         self._test_no_init(MyFilter)
  32
  33     def test_no_init_sink(self):
  34         class MySink(bt2._UserSinkComponent):
  35             def _consume(self):
  36                 pass
  37
  38         self._test_no_init(MySink)
  39
  40     def test_incomplete_source_no_msg_iter_cls(self):
  41         class MyIter(bt2._UserMessageIterator):
  42             pass
  43
  44         with self.assertRaises(bt2.IncompleteUserClass):
  45             class MySource(bt2._UserSourceComponent):
  46                 pass
  47
  48     def test_incomplete_source_wrong_msg_iter_cls_type(self):
  49         class MyIter(bt2._UserMessageIterator):
  50             pass
  51
  52         with self.assertRaises(bt2.IncompleteUserClass):
  53             class MySource(bt2._UserSourceComponent,
  54                            message_iterator_class=int):
  55                 pass
  56
  57     def test_incomplete_filter_no_msg_iter_cls(self):
  58         class MyIter(bt2._UserMessageIterator):
  59             pass
  60
  61         with self.assertRaises(bt2.IncompleteUserClass):
  62             class MyFilter(bt2._UserFilterComponent):
  63                 pass
  64
  65     def test_incomplete_sink_no_consume_method(self):
  66         class MyIter(bt2._UserMessageIterator):
  67             pass
  68
  69         with self.assertRaises(bt2.IncompleteUserClass):
  70             class MySink(bt2._UserSinkComponent):
  71                 pass
  72
  73     def test_minimal_source(self):
  74         class MyIter(bt2._UserMessageIterator):
  75             pass
  76
  77         class MySource(bt2._UserSourceComponent,
  78                        message_iterator_class=MyIter):
  79             pass
  80
  81     def test_minimal_filter(self):
  82         class MyIter(bt2._UserMessageIterator):
  83             pass
  84
  85         class MyFilter(bt2._UserFilterComponent,
  86                        message_iterator_class=MyIter):
  87             pass
  88
  89     def test_minimal_sink(self):
  90         class MySink(bt2._UserSinkComponent):
  91             def _consume(self):
  92                 pass
  93
  94     def test_default_name(self):
  95         class MySink(bt2._UserSinkComponent):
  96             def _consume(self):
  97                 pass
  98
  99         self.assertEqual(MySink.name, 'MySink')
 100
 101     def test_custom_name(self):
 102         class MySink(bt2._UserSinkComponent, name='salut'):
 103             def _consume(self):
 104                 pass
 105
 106         self.assertEqual(MySink.name, 'salut')
 107
 108     def test_invalid_custom_name(self):
 109         with self.assertRaises(TypeError):
 110             class MySink(bt2._UserSinkComponent, name=23):
 111                 def _consume(self):
 112                     pass
 113
 114     def test_description(self):
 115         class MySink(bt2._UserSinkComponent):
 116             """
 117             The description.
 118
 119             Bacon ipsum dolor amet ribeye t-bone corned beef, beef jerky
 120             porchetta burgdoggen prosciutto chicken frankfurter boudin
 121             hamburger doner bacon turducken. Sirloin shank sausage,
 122             boudin meatloaf alcatra meatball t-bone tongue pastrami
 123             cupim flank tenderloin.
 124             """
 125
 126             def _consume(self):
 127                 pass
 128
 129         self.assertEqual(MySink.description, 'The description.')
 130
 131     def test_empty_description(self):
 132         class MySink(bt2._UserSinkComponent):
 133             """
 134             """
 135
 136             def _consume(self):
 137                 pass
 138
 139         self.assertIsNone(MySink.description)
 140
 141     def test_help(self):
 142         class MySink(bt2._UserSinkComponent):
 143             """
 144             The description.
 145
 146             The help
 147             text is
 148             here.
 149             """
 150
 151             def _consume(self):
 152                 pass
 153
 154         self.assertEqual(MySink.help, 'The help\ntext is\nhere.')
 155
 156     def test_addr(self):
 157         class MySink(bt2._UserSinkComponent):
 158             def _consume(self):
 159                 pass
 160
 161         self.assertIsInstance(MySink.addr, int)
 162         self.assertNotEqual(MySink.addr, 0)
 163
 164     def test_query_not_implemented(self):
 165         class MySink(bt2._UserSinkComponent):
 166             def _consume(self):
 167                 pass
 168
 169         with self.assertRaises(bt2.Error):
 170             bt2.QueryExecutor().query(MySink, 'obj', 23)
 171
 172     def test_query_raises(self):
 173         class MySink(bt2._UserSinkComponent):
 174             def _consume(self):
 175                 pass
 176
 177             @classmethod
 178             def _query(cls, query_exec, obj, params):
 179                 raise ValueError
 180
 181         with self.assertRaises(bt2.Error):
 182             bt2.QueryExecutor().query(MySink, 'obj', 23)
 183
 184     def test_query_wrong_return_type(self):
 185         class MySink(bt2._UserSinkComponent):
 186             def _consume(self):
 187                 pass
 188
 189             @classmethod
 190             def _query(cls, query_exec, obj, params):
 191                 return ...
 192
 193         with self.assertRaises(bt2.Error):
 194             bt2.QueryExecutor().query(MySink, 'obj', 23)
 195
 196     def test_query_params_none(self):
 197         class MySink(bt2._UserSinkComponent):
 198             def _consume(self):
 199                 pass
 200
 201             @classmethod
 202             def _query(cls, query_exec, obj, params):
 203                 nonlocal query_params
 204                 query_params = params
 205                 return None
 206
 207         query_params = None
 208         params = None
 209         res = bt2.QueryExecutor().query(MySink, 'obj', params)
 210         self.assertEqual(query_params, params)
 211         self.assertIsNone(res)
 212         del query_params
 213
 214     def test_query_simple(self):
 215         class MySink(bt2._UserSinkComponent):
 216             def _consume(self):
 217                 pass
 218
 219             @classmethod
 220             def _query(cls, query_exec, obj, params):
 221                 nonlocal query_params
 222                 query_params = params
 223                 return 17.5
 224
 225         query_params = None
 226         params = ['coucou', 23, None]
 227         res = bt2.QueryExecutor().query(MySink, 'obj', params)
 228         self.assertEqual(query_params, params)
 229         self.assertEqual(res, 17.5)
 230         del query_params
 231
 232     def test_query_complex(self):
 233         class MySink(bt2._UserSinkComponent):
 234             def _consume(self):
 235                 pass
 236
 237             @classmethod
 238             def _query(cls, query_exec, obj, params):
 239                 nonlocal query_params
 240                 query_params = params
 241                 return {
 242                     'null': None,
 243                     'bt2': 'BT2',
 244                 }
 245
 246         query_params = None
 247         params = {
 248             'array': ['coucou', 23, None],
 249             'other_map': {
 250                 'yes': 'yeah',
 251                 '19': 19,
 252                 'minus 1.5': -1.5,
 253             },
 254             'null': None,
 255         }
 256
 257         res = bt2.QueryExecutor().query(MySink, 'obj', params)
 258         self.assertEqual(query_params, params)
 259         self.assertEqual(res, {
 260             'null': None,
 261             'bt2': 'BT2',
 262         })
 263         del query_params
 264
 265     def test_eq(self):
 266         class MySink(bt2._UserSinkComponent):
 267             def _consume(self):
 268                 pass
 269
 270         self.assertEqual(MySink, MySink)
 271
 272
 273 class GenericComponentClassTestCase(unittest.TestCase):
 274     def setUp(self):
 275         class MySink(bt2._UserSinkComponent):
 276             '''
 277             The description.
 278
 279             The help.
 280             '''
 281             def _consume(self):
 282                 pass
 283
 284             @classmethod
 285             def _query(cls, query_exec, obj, params):
 286                 return [obj, params, 23]
 287
 288         self._py_comp_cls = MySink
 289         graph = bt2.Graph()
 290         comp = graph.add_sink_component(MySink, 'salut')
 291         self._comp_cls = comp.component_class
 292         self.assertTrue(issubclass(type(self._comp_cls),
 293                                    bt2.component._GenericComponentClass))
 294
 295     def tearDown(self):
 296         del self._py_comp_cls
 297         del self._comp_cls
 298
 299     def test_description(self):
 300         self.assertEqual(self._comp_cls.description, 'The description.')
 301
 302     def test_help(self):
 303         self.assertEqual(self._comp_cls.help, 'The help.')
 304
 305     def test_name(self):
 306         self.assertEqual(self._comp_cls.name, 'MySink')
 307
 308     def test_addr(self):
 309         self.assertIsInstance(self._comp_cls.addr, int)
 310         self.assertNotEqual(self._comp_cls.addr, 0)
 311
 312     def test_eq_invalid(self):
 313         self.assertFalse(self._comp_cls == 23)
 314
 315     def test_eq(self):
 316         self.assertEqual(self._comp_cls, self._comp_cls)
 317         self.assertEqual(self._py_comp_cls, self._comp_cls)
 318
 319     def test_query(self):
 320         res = bt2.QueryExecutor().query(self._comp_cls, 'an object',
 321                                         {'yes': 'no', 'book': -17})
 322         expected = ['an object', {'yes': 'no', 'book': -17}, 23]
 323         self.assertEqual(res, expected)