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_no_lines(self):
 131         class MySink(bt2._UserSinkComponent):
 132             # fmt: off
 133             """"""
 134             # fmt: on
 135
 136             def _user_consume(self):
 137                 pass
 138
 139         self.assertIsNone(MySink.description)
 140
 141     def test_empty_description_no_contents(self):
 142         class MySink(bt2._UserSinkComponent):
 143             # fmt: off
 144             """
 145             """
 146             # fmt: on
 147
 148             def _user_consume(self):
 149                 pass
 150
 151         self.assertIsNone(MySink.description)
 152
 153     def test_empty_description_single_line(self):
 154         class MySink(bt2._UserSinkComponent):
 155             """my description"""
 156
 157             def _user_consume(self):
 158                 pass
 159
 160         self.assertEqual(MySink.description, "my description")
 161
 162     def test_help(self):
 163         class MySink(bt2._UserSinkComponent):
 164             """
 165             The description.
 166
 167             The help
 168             text is
 169             here.
 170             """
 171
 172             def _user_consume(self):
 173                 pass
 174
 175         self.assertEqual(MySink.help, 'The help\ntext is\nhere.')
 176
 177     def test_addr(self):
 178         class MySink(bt2._UserSinkComponent):
 179             def _user_consume(self):
 180                 pass
 181
 182         self.assertIsInstance(MySink.addr, int)
 183         self.assertNotEqual(MySink.addr, 0)
 184
 185     def test_query_not_implemented(self):
 186         class MySink(bt2._UserSinkComponent):
 187             def _user_consume(self):
 188                 pass
 189
 190         with self.assertRaises(bt2.UnknownObject):
 191             bt2.QueryExecutor(MySink, 'obj', 23).query()
 192
 193     def test_query_raises(self):
 194         class MySink(bt2._UserSinkComponent):
 195             def _user_consume(self):
 196                 pass
 197
 198             @classmethod
 199             def _user_query(cls, priv_query_exec, obj, params):
 200                 raise ValueError
 201
 202         with self.assertRaises(bt2._Error):
 203             bt2.QueryExecutor(MySink, 'obj', 23).query()
 204
 205     def test_query_wrong_return_type(self):
 206         class MySink(bt2._UserSinkComponent):
 207             def _user_consume(self):
 208                 pass
 209
 210             @classmethod
 211             def _user_query(cls, priv_query_exec, obj, params, method_obj):
 212                 return ...
 213
 214         with self.assertRaises(bt2._Error):
 215             bt2.QueryExecutor(MySink, 'obj', 23).query()
 216
 217     def test_query_params_none(self):
 218         class MySink(bt2._UserSinkComponent):
 219             def _user_consume(self):
 220                 pass
 221
 222             @classmethod
 223             def _user_query(cls, priv_query_exec, obj, params, method_obj):
 224                 nonlocal query_params
 225                 query_params = params
 226                 return None
 227
 228         query_params = None
 229         params = None
 230         res = bt2.QueryExecutor(MySink, 'obj', params).query()
 231         self.assertEqual(query_params, params)
 232         self.assertIsNone(res)
 233         del query_params
 234
 235     def test_query_logging_level(self):
 236         class MySink(bt2._UserSinkComponent):
 237             def _user_consume(self):
 238                 pass
 239
 240             @classmethod
 241             def _user_query(cls, priv_query_exec, obj, params, method_obj):
 242                 nonlocal query_log_level
 243                 query_log_level = priv_query_exec.logging_level
 244
 245         query_log_level = None
 246         query_exec = bt2.QueryExecutor(MySink, 'obj', None)
 247         query_exec.logging_level = bt2.LoggingLevel.WARNING
 248         query_exec.query()
 249         self.assertEqual(query_log_level, bt2.LoggingLevel.WARNING)
 250         del query_log_level
 251
 252     def test_query_returns_none(self):
 253         class MySink(bt2._UserSinkComponent):
 254             def _user_consume(self):
 255                 pass
 256
 257             @staticmethod
 258             def _user_query(priv_query_exec, obj, params, method_obj):
 259                 return
 260
 261         res = bt2.QueryExecutor(MySink, 'obj', None).query()
 262         self.assertIsNone(res)
 263
 264     def test_query_simple(self):
 265         class MySink(bt2._UserSinkComponent):
 266             def _user_consume(self):
 267                 pass
 268
 269             @classmethod
 270             def _user_query(cls, priv_query_exec, obj, params, method_obj):
 271                 nonlocal query_params
 272                 query_params = params
 273                 return 17.5
 274
 275         query_params = None
 276         params = ['coucou', 23, None]
 277         res = bt2.QueryExecutor(MySink, 'obj', params).query()
 278         self.assertEqual(query_params, params)
 279         self.assertEqual(res, 17.5)
 280         del query_params
 281
 282     def test_query_complex(self):
 283         class MySink(bt2._UserSinkComponent):
 284             def _user_consume(self):
 285                 pass
 286
 287             @classmethod
 288             def _user_query(cls, priv_query_exec, obj, params, method_obj):
 289                 nonlocal query_params
 290                 query_params = params
 291                 return {'null': None, 'bt2': 'BT2'}
 292
 293         query_params = None
 294         params = {
 295             'array': ['coucou', 23, None],
 296             'other_map': {'yes': 'yeah', '19': 19, 'minus 1.5': -1.5},
 297             'null': None,
 298         }
 299
 300         res = bt2.QueryExecutor(MySink, 'obj', params).query()
 301         self.assertEqual(query_params, params)
 302         self.assertEqual(res, {'null': None, 'bt2': 'BT2'})
 303         del query_params
 304
 305     def test_eq(self):
 306         class MySink(bt2._UserSinkComponent):
 307             def _user_consume(self):
 308                 pass
 309
 310         self.assertEqual(MySink, MySink)
 311
 312
 313 class ComponentClassTestCase(unittest.TestCase):
 314     def setUp(self):
 315         class MySink(bt2._UserSinkComponent):
 316             """
 317             The description.
 318
 319             The help.
 320             """
 321
 322             def _user_consume(self):
 323                 pass
 324
 325             @classmethod
 326             def _user_query(cls, priv_query_exec, obj, params, method_obj):
 327                 return [obj, params, 23]
 328
 329         self._py_comp_cls = MySink
 330         graph = bt2.Graph()
 331         comp = graph.add_component(MySink, 'salut')
 332         self._comp_cls = comp.cls
 333         self.assertIs(type(self._comp_cls), bt2._SinkComponentClassConst)
 334
 335     def tearDown(self):
 336         del self._py_comp_cls
 337         del self._comp_cls
 338
 339     def test_description(self):
 340         self.assertEqual(self._comp_cls.description, 'The description.')
 341
 342     def test_help(self):
 343         self.assertEqual(self._comp_cls.help, 'The help.')
 344
 345     def test_name(self):
 346         self.assertEqual(self._comp_cls.name, 'MySink')
 347
 348     def test_addr(self):
 349         self.assertIsInstance(self._comp_cls.addr, int)
 350         self.assertNotEqual(self._comp_cls.addr, 0)
 351
 352     def test_eq_invalid(self):
 353         self.assertFalse(self._comp_cls == 23)
 354
 355     def test_eq(self):
 356         self.assertEqual(self._comp_cls, self._comp_cls)
 357         self.assertEqual(self._py_comp_cls, self._comp_cls)
 358
 359     def test_query(self):
 360         res = bt2.QueryExecutor(
 361             self._comp_cls, 'an object', {'yes': 'no', 'book': -17}
 362         ).query()
 363         expected = ['an object', {'yes': 'no', 'book': -17}, 23]
 364         self.assertEqual(res, expected)
 365
 366
 367 if __name__ == '__main__':
 368     unittest.main()