tests/bindings/python/bt2/test_component_class.py

   1 #
   2 # Copyright (C) 2019 EfficiOS Inc.
   3 #
   4 # This program is free software; you can redistribute it and/or
   5 # modify it under the terms of the GNU General Public License
   6 # as published by the Free Software Foundation; only version 2
   7 # of the License.
   8 #
   9 # This program is distributed in the hope that it will be useful,
  10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 # GNU General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU General Public License
  15 # along with this program; if not, write to the Free Software
  16 # Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
  17 #
  18
  19 import unittest
  20 import bt2
  21
  22
  23 class UserComponentClassTestCase(unittest.TestCase):
  24     def _test_no_init(self, cls):
  25         with self.assertRaises(RuntimeError):
  26             cls()
  27
  28     def test_no_init_source(self):
  29         class MySource(
  30             bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
  31         ):
  32             pass
  33
  34         self._test_no_init(MySource)
  35
  36     def test_no_init_filter(self):
  37         class MyFilter(
  38             bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
  39         ):
  40             pass
  41
  42         self._test_no_init(MyFilter)
  43
  44     def test_no_init_sink(self):
  45         class MySink(bt2._UserSinkComponent):
  46             def _user_consume(self):
  47                 pass
  48
  49         self._test_no_init(MySink)
  50
  51     def test_incomplete_source_no_msg_iter_cls(self):
  52         with self.assertRaises(bt2._IncompleteUserClass):
  53
  54             class MySource(bt2._UserSourceComponent):
  55                 pass
  56
  57     def test_incomplete_source_wrong_msg_iter_cls_type(self):
  58         with self.assertRaises(bt2._IncompleteUserClass):
  59
  60             class MySource(bt2._UserSourceComponent, message_iterator_class=int):
  61                 pass
  62
  63     def test_incomplete_filter_no_msg_iter_cls(self):
  64         with self.assertRaises(bt2._IncompleteUserClass):
  65
  66             class MyFilter(bt2._UserFilterComponent):
  67                 pass
  68
  69     def test_incomplete_sink_no_consume_method(self):
  70         with self.assertRaises(bt2._IncompleteUserClass):
  71
  72             class MySink(bt2._UserSinkComponent):
  73                 pass
  74
  75     def test_minimal_source(self):
  76         class MySource(
  77             bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
  78         ):
  79             pass
  80
  81     def test_minimal_filter(self):
  82         class MyFilter(
  83             bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
  84         ):
  85             pass
  86
  87     def test_minimal_sink(self):
  88         class MySink(bt2._UserSinkComponent):
  89             def _user_consume(self):
  90                 pass
  91
  92     def test_default_name(self):
  93         class MySink(bt2._UserSinkComponent):
  94             def _user_consume(self):
  95                 pass
  96
  97         self.assertEqual(MySink.name, 'MySink')
  98
  99     def test_custom_name(self):
 100         class MySink(bt2._UserSinkComponent, name='salut'):
 101             def _user_consume(self):
 102                 pass
 103
 104         self.assertEqual(MySink.name, 'salut')
 105
 106     def test_invalid_custom_name(self):
 107         with self.assertRaises(TypeError):
 108
 109             class MySink(bt2._UserSinkComponent, name=23):
 110                 def _user_consume(self):
 111                     pass
 112
 113     def test_description(self):
 114         class MySink(bt2._UserSinkComponent):
 115             """
 116             The description.
 117
 118             Bacon ipsum dolor amet ribeye t-bone corned beef, beef jerky
 119             porchetta burgdoggen prosciutto chicken frankfurter boudin
 120             hamburger doner bacon turducken. Sirloin shank sausage,
 121             boudin meatloaf alcatra meatball t-bone tongue pastrami
 122             cupim flank tenderloin.
 123             """
 124
 125             def _user_consume(self):
 126                 pass
 127
 128         self.assertEqual(MySink.description, 'The description.')
 129
 130     def test_empty_description(self):
 131         class MySink(bt2._UserSinkComponent):
 132             """
 133             """
 134
 135             def _user_consume(self):
 136                 pass
 137
 138         self.assertIsNone(MySink.description)
 139
 140     def test_help(self):
 141         class MySink(bt2._UserSinkComponent):
 142             """
 143             The description.
 144
 145             The help
 146             text is
 147             here.
 148             """
 149
 150             def _user_consume(self):
 151                 pass
 152
 153         self.assertEqual(MySink.help, 'The help\ntext is\nhere.')
 154
 155     def test_addr(self):
 156         class MySink(bt2._UserSinkComponent):
 157             def _user_consume(self):
 158                 pass
 159
 160         self.assertIsInstance(MySink.addr, int)
 161         self.assertNotEqual(MySink.addr, 0)
 162
 163     def test_query_not_implemented(self):
 164         class MySink(bt2._UserSinkComponent):
 165             def _user_consume(self):
 166                 pass
 167
 168         with self.assertRaises(bt2.UnknownObject):
 169             bt2.QueryExecutor(MySink, 'obj', 23).query()
 170
 171     def test_query_raises(self):
 172         class MySink(bt2._UserSinkComponent):
 173             def _user_consume(self):
 174                 pass
 175
 176             @classmethod
 177             def _user_query(cls, priv_query_exec, obj, params):
 178                 raise ValueError
 179
 180         with self.assertRaises(bt2._Error):
 181             bt2.QueryExecutor(MySink, 'obj', 23).query()
 182
 183     def test_query_wrong_return_type(self):
 184         class MySink(bt2._UserSinkComponent):
 185             def _user_consume(self):
 186                 pass
 187
 188             @classmethod
 189             def _user_query(cls, priv_query_exec, obj, params, method_obj):
 190                 return ...
 191
 192         with self.assertRaises(bt2._Error):
 193             bt2.QueryExecutor(MySink, 'obj', 23).query()
 194
 195     def test_query_params_none(self):
 196         class MySink(bt2._UserSinkComponent):
 197             def _user_consume(self):
 198                 pass
 199
 200             @classmethod
 201             def _user_query(cls, priv_query_exec, obj, params, method_obj):
 202                 nonlocal query_params
 203                 query_params = params
 204                 return None
 205
 206         query_params = None
 207         params = None
 208         res = bt2.QueryExecutor(MySink, 'obj', params).query()
 209         self.assertEqual(query_params, params)
 210         self.assertIsNone(res)
 211         del query_params
 212
 213     def test_query_logging_level(self):
 214         class MySink(bt2._UserSinkComponent):
 215             def _user_consume(self):
 216                 pass
 217
 218             @classmethod
 219             def _user_query(cls, priv_query_exec, obj, params, method_obj):
 220                 nonlocal query_log_level
 221                 query_log_level = priv_query_exec.logging_level
 222
 223         query_log_level = None
 224         query_exec = bt2.QueryExecutor(MySink, 'obj', None)
 225         query_exec.logging_level = bt2.LoggingLevel.WARNING
 226         query_exec.query()
 227         self.assertEqual(query_log_level, bt2.LoggingLevel.WARNING)
 228         del query_log_level
 229
 230     def test_query_returns_none(self):
 231         class MySink(bt2._UserSinkComponent):
 232             def _user_consume(self):
 233                 pass
 234
 235             @staticmethod
 236             def _user_query(priv_query_exec, obj, params, method_obj):
 237                 return
 238
 239         res = bt2.QueryExecutor(MySink, 'obj', None).query()
 240         self.assertIsNone(res)
 241
 242     def test_query_simple(self):
 243         class MySink(bt2._UserSinkComponent):
 244             def _user_consume(self):
 245                 pass
 246
 247             @classmethod
 248             def _user_query(cls, priv_query_exec, obj, params, method_obj):
 249                 nonlocal query_params
 250                 query_params = params
 251                 return 17.5
 252
 253         query_params = None
 254         params = ['coucou', 23, None]
 255         res = bt2.QueryExecutor(MySink, 'obj', params).query()
 256         self.assertEqual(query_params, params)
 257         self.assertEqual(res, 17.5)
 258         del query_params
 259
 260     def test_query_complex(self):
 261         class MySink(bt2._UserSinkComponent):
 262             def _user_consume(self):
 263                 pass
 264
 265             @classmethod
 266             def _user_query(cls, priv_query_exec, obj, params, method_obj):
 267                 nonlocal query_params
 268                 query_params = params
 269                 return {'null': None, 'bt2': 'BT2'}
 270
 271         query_params = None
 272         params = {
 273             'array': ['coucou', 23, None],
 274             'other_map': {'yes': 'yeah', '19': 19, 'minus 1.5': -1.5},
 275             'null': None,
 276         }
 277
 278         res = bt2.QueryExecutor(MySink, 'obj', params).query()
 279         self.assertEqual(query_params, params)
 280         self.assertEqual(res, {'null': None, 'bt2': 'BT2'})
 281         del query_params
 282
 283     def test_eq(self):
 284         class MySink(bt2._UserSinkComponent):
 285             def _user_consume(self):
 286                 pass
 287
 288         self.assertEqual(MySink, MySink)
 289
 290
 291 class ComponentClassTestCase(unittest.TestCase):
 292     def setUp(self):
 293         class MySink(bt2._UserSinkComponent):
 294             '''
 295             The description.
 296
 297             The help.
 298             '''
 299
 300             def _user_consume(self):
 301                 pass
 302
 303             @classmethod
 304             def _user_query(cls, priv_query_exec, obj, params, method_obj):
 305                 return [obj, params, 23]
 306
 307         self._py_comp_cls = MySink
 308         graph = bt2.Graph()
 309         comp = graph.add_component(MySink, 'salut')
 310         self._comp_cls = comp.cls
 311         self.assertIs(type(self._comp_cls), bt2._SinkComponentClassConst)
 312
 313     def tearDown(self):
 314         del self._py_comp_cls
 315         del self._comp_cls
 316
 317     def test_description(self):
 318         self.assertEqual(self._comp_cls.description, 'The description.')
 319
 320     def test_help(self):
 321         self.assertEqual(self._comp_cls.help, 'The help.')
 322
 323     def test_name(self):
 324         self.assertEqual(self._comp_cls.name, 'MySink')
 325
 326     def test_addr(self):
 327         self.assertIsInstance(self._comp_cls.addr, int)
 328         self.assertNotEqual(self._comp_cls.addr, 0)
 329
 330     def test_eq_invalid(self):
 331         self.assertFalse(self._comp_cls == 23)
 332
 333     def test_eq(self):
 334         self.assertEqual(self._comp_cls, self._comp_cls)
 335         self.assertEqual(self._py_comp_cls, self._comp_cls)
 336
 337     def test_query(self):
 338         res = bt2.QueryExecutor(
 339             self._comp_cls, 'an object', {'yes': 'no', 'book': -17}
 340         ).query()
 341         expected = ['an object', {'yes': 'no', 'book': -17}, 23]
 342         self.assertEqual(res, expected)
 343
 344
 345 if __name__ == '__main__':
 346     unittest.main()