fix: test_message_iterator.py hangs on Python 3.12
[babeltrace.git] / tests / bindings / python / bt2 / test_error.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 from bt2 import native_bt
20 import bt2
21 import unittest
22
23
24 class FailingIter(bt2._UserMessageIterator):
25 def __next__(self):
26 raise ValueError('User message iterator is failing')
27
28
29 class SourceWithFailingIter(
30 bt2._UserSourceComponent, message_iterator_class=FailingIter
31 ):
32 def __init__(self, config, params, obj):
33 self._add_output_port('out')
34
35
36 class SourceWithFailingInit(
37 bt2._UserSourceComponent, message_iterator_class=FailingIter
38 ):
39 def __init__(self, config, params, obj):
40 raise ValueError('Source is failing')
41
42
43 class WorkingSink(bt2._UserSinkComponent):
44 def __init__(self, config, params, obj):
45 self._in = self._add_input_port('in')
46
47 def _user_graph_is_configured(self):
48 self._iter = self._create_message_iterator(self._in)
49
50 def _user_consume(self):
51 next(self._iter)
52
53
54 class SinkWithExceptionChaining(bt2._UserSinkComponent):
55 def __init__(self, config, params, obj):
56 self._in = self._add_input_port('in')
57
58 def _user_graph_is_configured(self):
59 self._iter = self._create_message_iterator(self._in)
60
61 def _user_consume(self):
62 try:
63 next(self._iter)
64 except bt2._Error as e:
65 raise ValueError('oops') from e
66
67
68 class SinkWithFailingQuery(bt2._UserSinkComponent):
69 def _user_consume(self):
70 pass
71
72 @staticmethod
73 def _user_query(priv_executor, obj, params, method_obj):
74 raise ValueError('Query is failing')
75
76
77 class ErrorTestCase(unittest.TestCase):
78 def _run_failing_graph(self, source_cc, sink_cc):
79 with self.assertRaises(bt2._Error) as ctx:
80 graph = bt2.Graph()
81 src = graph.add_component(source_cc, 'src')
82 snk = graph.add_component(sink_cc, 'snk')
83 graph.connect_ports(src.output_ports['out'], snk.input_ports['in'])
84 graph.run()
85
86 return ctx.exception
87
88 def test_current_thread_error_none(self):
89 # When a bt2._Error is raised, it steals the current thread's error.
90 # Verify that it is now NULL.
91 self._run_failing_graph(SourceWithFailingInit, WorkingSink)
92 self.assertIsNone(native_bt.current_thread_take_error())
93
94 def test_len(self):
95 exc = self._run_failing_graph(SourceWithFailingIter, WorkingSink)
96
97 # The exact number of causes is not too important (it can change if we
98 # append more or less causes along the way), but the idea is to verify is
99 # has a value that makes sense.
100 self.assertEqual(len(exc), 4)
101
102 def test_iter(self):
103 exc = self._run_failing_graph(SourceWithFailingIter, WorkingSink)
104
105 for c in exc:
106 # Each cause is an instance of _ErrorCause (including subclasses).
107 self.assertIsInstance(c, bt2._ErrorCause)
108
109 def test_getitem(self):
110 exc = self._run_failing_graph(SourceWithFailingIter, WorkingSink)
111
112 for i in range(len(exc)):
113 c = exc[i]
114 # Each cause is an instance of _ErrorCause (including subclasses).
115 self.assertIsInstance(c, bt2._ErrorCause)
116
117 def test_getitem_indexerror(self):
118 exc = self._run_failing_graph(SourceWithFailingIter, WorkingSink)
119
120 with self.assertRaises(IndexError):
121 exc[len(exc)]
122
123 def test_exception_chaining(self):
124 # Test that if we do:
125 #
126 # try:
127 # ...
128 # except bt2._Error as exc:
129 # raise ValueError('oh noes') from exc
130 #
131 # We are able to fetch the causes of the original bt2._Error in the
132 # exception chain. Also, each exception in the chain should become one
133 # cause once caught.
134 exc = self._run_failing_graph(SourceWithFailingIter, SinkWithExceptionChaining)
135
136 self.assertEqual(len(exc), 5)
137
138 self.assertIsInstance(exc[0], bt2._MessageIteratorErrorCause)
139 self.assertEqual(exc[0].component_class_name, 'SourceWithFailingIter')
140 self.assertIn('ValueError: User message iterator is failing', exc[0].message)
141
142 self.assertIsInstance(exc[1], bt2._ErrorCause)
143
144 self.assertIsInstance(exc[2], bt2._ComponentErrorCause)
145 self.assertEqual(exc[2].component_class_name, 'SinkWithExceptionChaining')
146 self.assertIn(
147 'unexpected error: cannot advance the message iterator', exc[2].message
148 )
149
150 self.assertIsInstance(exc[3], bt2._ComponentErrorCause)
151 self.assertEqual(exc[3].component_class_name, 'SinkWithExceptionChaining')
152 self.assertIn('ValueError: oops', exc[3].message)
153
154 self.assertIsInstance(exc[4], bt2._ErrorCause)
155
156 def _common_cause_tests(self, cause):
157 self.assertIsInstance(cause.module_name, str)
158 self.assertIsInstance(cause.file_name, str)
159 self.assertIsInstance(cause.line_number, int)
160
161 def test_unknown_error_cause(self):
162 exc = self._run_failing_graph(SourceWithFailingIter, SinkWithExceptionChaining)
163 cause = exc[-1]
164 self.assertIs(type(cause), bt2._ErrorCause)
165 self._common_cause_tests(cause)
166
167 def test_component_error_cause(self):
168 exc = self._run_failing_graph(SourceWithFailingInit, SinkWithExceptionChaining)
169 cause = exc[0]
170 self.assertIs(type(cause), bt2._ComponentErrorCause)
171 self._common_cause_tests(cause)
172
173 self.assertIn('Source is failing', cause.message)
174 self.assertEqual(cause.component_name, 'src')
175 self.assertEqual(cause.component_class_type, bt2.ComponentClassType.SOURCE)
176 self.assertEqual(cause.component_class_name, 'SourceWithFailingInit')
177 self.assertIsNone(cause.plugin_name)
178
179 def test_component_class_error_cause(self):
180 q = bt2.QueryExecutor(SinkWithFailingQuery, 'hello')
181
182 with self.assertRaises(bt2._Error) as ctx:
183 q.query()
184
185 cause = ctx.exception[0]
186 self.assertIs(type(cause), bt2._ComponentClassErrorCause)
187 self._common_cause_tests(cause)
188
189 self.assertIn('Query is failing', cause.message)
190
191 self.assertEqual(cause.component_class_type, bt2.ComponentClassType.SINK)
192 self.assertEqual(cause.component_class_name, 'SinkWithFailingQuery')
193 self.assertIsNone(cause.plugin_name)
194
195 def test_message_iterator_error_cause(self):
196 exc = self._run_failing_graph(SourceWithFailingIter, SinkWithExceptionChaining)
197 cause = exc[0]
198 self.assertIs(type(cause), bt2._MessageIteratorErrorCause)
199 self._common_cause_tests(cause)
200
201 self.assertIn('User message iterator is failing', cause.message)
202 self.assertEqual(cause.component_name, 'src')
203 self.assertEqual(cause.component_output_port_name, 'out')
204 self.assertEqual(cause.component_class_type, bt2.ComponentClassType.SOURCE)
205 self.assertEqual(cause.component_class_name, 'SourceWithFailingIter')
206 self.assertIsNone(cause.plugin_name)
207
208 def test_str(self):
209 # Test __str__. We don't need to test the precise format used, but
210 # just that it doesn't miserably crash and that it contains some
211 # expected bits.
212 exc = self._run_failing_graph(SourceWithFailingIter, SinkWithExceptionChaining)
213 s = str(exc)
214 self.assertIn("[src (out): 'source.SourceWithFailingIter']", s)
215 self.assertIn('ValueError: oops', s)
216
217
218 if __name__ == '__main__':
219 unittest.main()
This page took 0.038829 seconds and 4 git commands to generate.