tests/bindings/python/bt2/test_component_class.py

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