lib: add precond. check for begin <= end on pkt./ev. disc. msg. creation
[babeltrace.git] / 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()
This page took 0.03367 seconds and 4 git commands to generate.