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, log_level):
 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, log_level):
 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, log_level):
 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_logging_level(self):
 215         class MySink(bt2._UserSinkComponent):
 216             def _consume(self):
 217                 pass
 218
 219             @classmethod
 220             def _query(cls, query_exec, obj, params, log_level):
 221                 nonlocal query_log_level
 222                 query_log_level = log_level
 223
 224         query_log_level = None
 225         res = bt2.QueryExecutor().query(MySink, 'obj', None,
 226                                         bt2.LoggingLevel.WARN)
 227         self.assertEqual(query_log_level, bt2.LoggingLevel.WARN)
 228         del query_log_level
 229
 230     def test_query_simple(self):
 231         class MySink(bt2._UserSinkComponent):
 232             def _consume(self):
 233                 pass
 234
 235             @classmethod
 236             def _query(cls, query_exec, obj, params, log_level):
 237                 nonlocal query_params
 238                 query_params = params
 239                 return 17.5
 240
 241         query_params = None
 242         params = ['coucou', 23, None]
 243         res = bt2.QueryExecutor().query(MySink, 'obj', params)
 244         self.assertEqual(query_params, params)
 245         self.assertEqual(res, 17.5)
 246         del query_params
 247
 248     def test_query_complex(self):
 249         class MySink(bt2._UserSinkComponent):
 250             def _consume(self):
 251                 pass
 252
 253             @classmethod
 254             def _query(cls, query_exec, obj, params, log_level):
 255                 nonlocal query_params
 256                 query_params = params
 257                 return {
 258                     'null': None,
 259                     'bt2': 'BT2',
 260                 }
 261
 262         query_params = None
 263         params = {
 264             'array': ['coucou', 23, None],
 265             'other_map': {
 266                 'yes': 'yeah',
 267                 '19': 19,
 268                 'minus 1.5': -1.5,
 269             },
 270             'null': None,
 271         }
 272
 273         res = bt2.QueryExecutor().query(MySink, 'obj', params)
 274         self.assertEqual(query_params, params)
 275         self.assertEqual(res, {
 276             'null': None,
 277             'bt2': 'BT2',
 278         })
 279         del query_params
 280
 281     def test_eq(self):
 282         class MySink(bt2._UserSinkComponent):
 283             def _consume(self):
 284                 pass
 285
 286         self.assertEqual(MySink, MySink)
 287
 288
 289 class GenericComponentClassTestCase(unittest.TestCase):
 290     def setUp(self):
 291         class MySink(bt2._UserSinkComponent):
 292             '''
 293             The description.
 294
 295             The help.
 296             '''
 297             def _consume(self):
 298                 pass
 299
 300             @classmethod
 301             def _query(cls, query_exec, obj, params, log_level):
 302                 return [obj, params, 23]
 303
 304         self._py_comp_cls = MySink
 305         graph = bt2.Graph()
 306         comp = graph.add_component(MySink, 'salut')
 307         self._comp_cls = comp.cls
 308         self.assertTrue(issubclass(type(self._comp_cls),
 309                                    bt2.component._GenericComponentClass))
 310
 311     def tearDown(self):
 312         del self._py_comp_cls
 313         del self._comp_cls
 314
 315     def test_description(self):
 316         self.assertEqual(self._comp_cls.description, 'The description.')
 317
 318     def test_help(self):
 319         self.assertEqual(self._comp_cls.help, 'The help.')
 320
 321     def test_name(self):
 322         self.assertEqual(self._comp_cls.name, 'MySink')
 323
 324     def test_addr(self):
 325         self.assertIsInstance(self._comp_cls.addr, int)
 326         self.assertNotEqual(self._comp_cls.addr, 0)
 327
 328     def test_eq_invalid(self):
 329         self.assertFalse(self._comp_cls == 23)
 330
 331     def test_eq(self):
 332         self.assertEqual(self._comp_cls, self._comp_cls)
 333         self.assertEqual(self._py_comp_cls, self._comp_cls)
 334
 335     def test_query(self):
 336         res = bt2.QueryExecutor().query(self._comp_cls, 'an object',
 337                                         {'yes': 'no', 'book': -17})
 338         expected = ['an object', {'yes': 'no', 'book': -17}, 23]
 339         self.assertEqual(res, expected)