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