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