fix: test_message_iterator.py hangs on Python 3.12
[babeltrace.git] / tests / bindings / python / bt2 / test_event_class.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 utils import get_default_trace_class
22 from bt2 import stream_class as bt2_stream_class
23 from bt2 import event_class as bt2_event_class
24 from bt2 import field_class as bt2_field_class
25 from bt2 import value as bt2_value
26 from utils import TestOutputPortMessageIterator
27
28
29 def _create_const_event_class(tc, stream_class):
30 fc1 = tc.create_structure_field_class()
31 fc2 = tc.create_structure_field_class()
32 event_class = stream_class.create_event_class(
33 payload_field_class=fc1, specific_context_field_class=fc2
34 )
35
36 class MyIter(bt2._UserMessageIterator):
37 def __init__(self, config, self_port_output):
38
39 trace = tc()
40 stream = trace.create_stream(stream_class)
41 self._msgs = [
42 self._create_stream_beginning_message(stream),
43 self._create_event_message(event_class, stream),
44 ]
45
46 def __next__(self):
47 if len(self._msgs) == 0:
48 raise StopIteration
49
50 return self._msgs.pop(0)
51
52 class MySrc(bt2._UserSourceComponent, message_iterator_class=MyIter):
53 def __init__(self, config, params, obj):
54 self._add_output_port('out', params)
55
56 graph = bt2.Graph()
57 src_comp = graph.add_component(MySrc, 'my_source', None)
58 msg_iter = TestOutputPortMessageIterator(graph, src_comp.output_ports['out'])
59
60 # Ignore first message, stream beginning
61 _ = next(msg_iter)
62
63 event_msg = next(msg_iter)
64
65 return event_msg.event.cls
66
67
68 class EventClassTestCase(unittest.TestCase):
69 def setUp(self):
70 self._tc = get_default_trace_class()
71
72 self._context_fc = self._tc.create_structure_field_class()
73 self._context_fc.append_member('allo', self._tc.create_string_field_class())
74 self._context_fc.append_member(
75 'zola', self._tc.create_signed_integer_field_class(18)
76 )
77
78 self._payload_fc = self._tc.create_structure_field_class()
79 self._payload_fc.append_member('zoom', self._tc.create_string_field_class())
80
81 self._stream_class = self._tc.create_stream_class(
82 assigns_automatic_event_class_id=True
83 )
84
85 def test_create_default(self):
86 ec = self._stream_class.create_event_class()
87
88 self.assertIs(type(ec), bt2_event_class._EventClass)
89 self.assertIsNone(ec.name, 'my_event')
90 self.assertTrue(type(ec.id), int)
91 self.assertIsNone(ec.specific_context_field_class)
92 self.assertIsNone(ec.payload_field_class)
93 self.assertIsNone(ec.emf_uri)
94 self.assertIsNone(ec.log_level)
95 self.assertEqual(len(ec.user_attributes), 0)
96
97 def test_create_invalid_id(self):
98 sc = self._tc.create_stream_class(assigns_automatic_event_class_id=False)
99 with self.assertRaises(TypeError):
100 sc.create_event_class(id='lel')
101
102 self.assertEqual(len(sc), 0)
103
104 def test_create_specific_context_field_class(self):
105 fc = self._tc.create_structure_field_class()
106 ec = self._stream_class.create_event_class(specific_context_field_class=fc)
107 self.assertEqual(ec.specific_context_field_class.addr, fc.addr)
108 self.assertIs(
109 type(ec.specific_context_field_class), bt2_field_class._StructureFieldClass
110 )
111
112 def test_const_create_specific_context_field_class(self):
113 ec_const = _create_const_event_class(self._tc, self._stream_class)
114 self.assertIs(
115 type(ec_const.specific_context_field_class),
116 bt2_field_class._StructureFieldClassConst,
117 )
118
119 def test_create_invalid_specific_context_field_class(self):
120 with self.assertRaises(TypeError):
121 self._stream_class.create_event_class(specific_context_field_class='lel')
122
123 self.assertEqual(len(self._stream_class), 0)
124
125 def test_create_payload_field_class(self):
126 fc = self._tc.create_structure_field_class()
127 ec = self._stream_class.create_event_class(payload_field_class=fc)
128 self.assertEqual(ec.payload_field_class.addr, fc.addr)
129 self.assertIs(
130 type(ec.payload_field_class), bt2_field_class._StructureFieldClass
131 )
132
133 def test_const_create_payload_field_class(self):
134 ec_const = _create_const_event_class(self._tc, self._stream_class)
135 self.assertIs(
136 type(ec_const.payload_field_class),
137 bt2_field_class._StructureFieldClassConst,
138 )
139
140 def test_create_invalid_payload_field_class(self):
141 with self.assertRaises(TypeError):
142 self._stream_class.create_event_class(payload_field_class='lel')
143
144 self.assertEqual(len(self._stream_class), 0)
145
146 def test_create_name(self):
147 ec = self._stream_class.create_event_class(name='viande à chien')
148 self.assertEqual(ec.name, 'viande à chien')
149
150 def test_create_invalid_name(self):
151 with self.assertRaises(TypeError):
152 self._stream_class.create_event_class(name=2)
153
154 self.assertEqual(len(self._stream_class), 0)
155
156 def test_emf_uri(self):
157 ec = self._stream_class.create_event_class(emf_uri='salut')
158 self.assertEqual(ec.emf_uri, 'salut')
159
160 def test_create_invalid_emf_uri(self):
161 with self.assertRaises(TypeError):
162 self._stream_class.create_event_class(emf_uri=23)
163
164 self.assertEqual(len(self._stream_class), 0)
165
166 def test_create_log_level(self):
167 ec = self._stream_class.create_event_class(
168 log_level=bt2.EventClassLogLevel.EMERGENCY
169 )
170 self.assertEqual(ec.log_level, bt2.EventClassLogLevel.EMERGENCY)
171
172 def test_create_invalid_log_level(self):
173 with self.assertRaises(ValueError):
174 self._stream_class.create_event_class(log_level='zoom')
175
176 self.assertEqual(len(self._stream_class), 0)
177
178 def test_create_user_attributes(self):
179 ec = self._stream_class.create_event_class(user_attributes={'salut': 23})
180 self.assertEqual(ec.user_attributes, {'salut': 23})
181 self.assertIs(type(ec.user_attributes), bt2_value.MapValue)
182
183 def test_const_create_user_attributes(self):
184 ec_const = _create_const_event_class(self._tc, self._stream_class)
185 self.assertIs(type(ec_const.user_attributes), bt2_value._MapValueConst)
186
187 def test_create_invalid_user_attributes(self):
188 with self.assertRaises(TypeError):
189 self._stream_class.create_event_class(user_attributes=object())
190
191 self.assertEqual(len(self._stream_class), 0)
192
193 def test_create_invalid_user_attributes_value_type(self):
194 with self.assertRaises(TypeError):
195 self._stream_class.create_event_class(user_attributes=23)
196
197 self.assertEqual(len(self._stream_class), 0)
198
199 def test_stream_class(self):
200 ec = self._stream_class.create_event_class()
201 self.assertEqual(ec.stream_class.addr, self._stream_class.addr)
202 self.assertIs(type(ec.stream_class), bt2_stream_class._StreamClass)
203
204 def test_const_stream_class(self):
205 ec_const = _create_const_event_class(self._tc, self._stream_class)
206 self.assertIs(type(ec_const.stream_class), bt2_stream_class._StreamClassConst)
207
208
209 if __name__ == '__main__':
210 unittest.main()
This page took 0.039604 seconds and 4 git commands to generate.