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