fix: test_message_iterator.py hangs on Python 3.12
[babeltrace.git] / 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()
This page took 0.032865 seconds and 4 git commands to generate.