tests/bindings/python/bt2/test_component.py

   1 # SPDX-License-Identifier: GPL-2.0-only
   2 #
   3 # Copyright (C) 2019 EfficiOS Inc.
   4 #
   5
   6 import unittest
   7 import bt2
   8 from bt2 import component as bt2_component
   9
  10
  11 class UserComponentTestCase(unittest.TestCase):
  12     @staticmethod
  13     def _create_comp(comp_cls, name=None, log_level=bt2.LoggingLevel.NONE):
  14         graph = bt2.Graph()
  15
  16         if name is None:
  17             name = 'comp'
  18
  19         return graph.add_component(comp_cls, name, logging_level=log_level)
  20
  21     def test_name(self):
  22         class MySink(bt2._UserSinkComponent):
  23             def __init__(comp_self, config, params, obj):
  24                 self.assertEqual(comp_self.name, 'yaes')
  25
  26             def _user_consume(self):
  27                 pass
  28
  29         self._create_comp(MySink, 'yaes')
  30
  31     def test_logging_level(self):
  32         class MySink(bt2._UserSinkComponent):
  33             def __init__(comp_self, config, params, obj):
  34                 self.assertEqual(comp_self.logging_level, bt2.LoggingLevel.INFO)
  35
  36             def _user_consume(self):
  37                 pass
  38
  39         self._create_comp(MySink, 'yaes', bt2.LoggingLevel.INFO)
  40
  41     def test_graph_mip_version(self):
  42         class MySink(bt2._UserSinkComponent):
  43             def __init__(comp_self, config, params, obj):
  44                 self.assertEqual(comp_self._graph_mip_version, 0)
  45
  46             def _user_consume(self):
  47                 pass
  48
  49         self._create_comp(MySink, 'yaes', bt2.LoggingLevel.INFO)
  50
  51     def test_class(self):
  52         class MySink(bt2._UserSinkComponent):
  53             def __init__(comp_self, config, params, obj):
  54                 self.assertEqual(comp_self.cls, MySink)
  55
  56             def _user_consume(self):
  57                 pass
  58
  59         self._create_comp(MySink)
  60
  61     def test_addr(self):
  62         class MySink(bt2._UserSinkComponent):
  63             def __init__(comp_self, config, params, obj):
  64                 self.assertIsInstance(comp_self.addr, int)
  65                 self.assertNotEqual(comp_self.addr, 0)
  66
  67             def _user_consume(self):
  68                 pass
  69
  70         self._create_comp(MySink)
  71
  72     def test_finalize(self):
  73         finalized = False
  74
  75         class MySink(bt2._UserSinkComponent):
  76             def _user_consume(self):
  77                 pass
  78
  79             def _user_finalize(comp_self):
  80                 nonlocal finalized
  81                 finalized = True
  82
  83         graph = bt2.Graph()
  84         comp = graph.add_component(MySink, 'lel')
  85
  86         del graph
  87         del comp
  88         self.assertTrue(finalized)
  89
  90     def test_source_component_config(self):
  91         class MySource(
  92             bt2._UserSourceComponent, message_iterator_class=bt2._UserMessageIterator
  93         ):
  94             def __init__(comp_self, config, params, obj):
  95                 nonlocal cfg_type
  96                 cfg_type = type(config)
  97
  98         cfg_type = None
  99         self._create_comp(MySource)
 100         self.assertIs(cfg_type, bt2_component._UserSourceComponentConfiguration)
 101
 102     def test_filter_component_config(self):
 103         class MyFilter(
 104             bt2._UserFilterComponent, message_iterator_class=bt2._UserMessageIterator
 105         ):
 106             def __init__(comp_self, config, params, obj):
 107                 nonlocal cfg_type
 108                 cfg_type = type(config)
 109
 110         cfg_type = None
 111         self._create_comp(MyFilter)
 112         self.assertIs(cfg_type, bt2_component._UserFilterComponentConfiguration)
 113
 114     def test_sink_component_config(self):
 115         class MySink(bt2._UserSinkComponent):
 116             def __init__(comp_self, config, params, obj):
 117                 nonlocal cfg_type
 118                 cfg_type = type(config)
 119
 120             def _user_consume(self):
 121                 pass
 122
 123         cfg_type = None
 124         self._create_comp(MySink)
 125         self.assertIs(cfg_type, bt2_component._UserSinkComponentConfiguration)
 126
 127
 128 class GenericComponentTestCase(unittest.TestCase):
 129     @staticmethod
 130     def _create_comp(comp_cls, name=None, log_level=bt2.LoggingLevel.NONE):
 131         graph = bt2.Graph()
 132
 133         if name is None:
 134             name = 'comp'
 135
 136         return graph.add_component(comp_cls, name, logging_level=log_level)
 137
 138     def test_name(self):
 139         class MySink(bt2._UserSinkComponent):
 140             def _user_consume(self):
 141                 pass
 142
 143         comp = self._create_comp(MySink, 'yaes')
 144         self.assertEqual(comp.name, 'yaes')
 145
 146     def test_logging_level(self):
 147         class MySink(bt2._UserSinkComponent):
 148             def _user_consume(self):
 149                 pass
 150
 151         comp = self._create_comp(MySink, 'yaes', bt2.LoggingLevel.WARNING)
 152         self.assertEqual(comp.logging_level, bt2.LoggingLevel.WARNING)
 153
 154     def test_class(self):
 155         class MySink(bt2._UserSinkComponent):
 156             def _user_consume(self):
 157                 pass
 158
 159         comp = self._create_comp(MySink)
 160         self.assertEqual(comp.cls, MySink)
 161
 162     def test_addr(self):
 163         class MySink(bt2._UserSinkComponent):
 164             def _user_consume(self):
 165                 pass
 166
 167         comp = self._create_comp(MySink)
 168         self.assertIsInstance(comp.addr, int)
 169         self.assertNotEqual(comp.addr, 0)
 170
 171
 172 if __name__ == '__main__':
 173     unittest.main()