1 # SPDX-License-Identifier: GPL-2.0-only
3 # Copyright (C) 2019 EfficiOS Inc.
6 from bt2
import native_bt
11 class FailingIter(bt2
._UserMessageIterator
):
13 raise ValueError('User message iterator is failing')
16 class SourceWithFailingIter(
17 bt2
._UserSourceComponent
, message_iterator_class
=FailingIter
19 def __init__(self
, config
, params
, obj
):
20 self
._add
_output
_port
('out')
23 class SourceWithFailingInit(
24 bt2
._UserSourceComponent
, message_iterator_class
=FailingIter
26 def __init__(self
, config
, params
, obj
):
27 raise ValueError('Source is failing')
30 class WorkingSink(bt2
._UserSinkComponent
):
31 def __init__(self
, config
, params
, obj
):
32 self
._in
= self
._add
_input
_port
('in')
34 def _user_graph_is_configured(self
):
35 self
._iter
= self
._create
_message
_iterator
(self
._in
)
37 def _user_consume(self
):
41 class SinkWithExceptionChaining(bt2
._UserSinkComponent
):
42 def __init__(self
, config
, params
, obj
):
43 self
._in
= self
._add
_input
_port
('in')
45 def _user_graph_is_configured(self
):
46 self
._iter
= self
._create
_message
_iterator
(self
._in
)
48 def _user_consume(self
):
51 except bt2
._Error
as e
:
52 raise ValueError('oops') from e
55 class SinkWithFailingQuery(bt2
._UserSinkComponent
):
56 def _user_consume(self
):
60 def _user_query(priv_executor
, obj
, params
, method_obj
):
61 raise ValueError('Query is failing')
64 class ErrorTestCase(unittest
.TestCase
):
65 def _run_failing_graph(self
, source_cc
, sink_cc
):
66 with self
.assertRaises(bt2
._Error
) as ctx
:
68 src
= graph
.add_component(source_cc
, 'src')
69 snk
= graph
.add_component(sink_cc
, 'snk')
70 graph
.connect_ports(src
.output_ports
['out'], snk
.input_ports
['in'])
75 def test_current_thread_error_none(self
):
76 # When a bt2._Error is raised, it steals the current thread's error.
77 # Verify that it is now NULL.
78 self
._run
_failing
_graph
(SourceWithFailingInit
, WorkingSink
)
79 self
.assertIsNone(native_bt
.current_thread_take_error())
82 exc
= self
._run
_failing
_graph
(SourceWithFailingIter
, WorkingSink
)
84 # The exact number of causes is not too important (it can change if we
85 # append more or less causes along the way), but the idea is to verify is
86 # has a value that makes sense.
87 self
.assertEqual(len(exc
), 4)
90 exc
= self
._run
_failing
_graph
(SourceWithFailingIter
, WorkingSink
)
93 # Each cause is an instance of _ErrorCause (including subclasses).
94 self
.assertIsInstance(c
, bt2
._ErrorCause
)
96 def test_getitem(self
):
97 exc
= self
._run
_failing
_graph
(SourceWithFailingIter
, WorkingSink
)
99 for i
in range(len(exc
)):
101 # Each cause is an instance of _ErrorCause (including subclasses).
102 self
.assertIsInstance(c
, bt2
._ErrorCause
)
104 def test_getitem_indexerror(self
):
105 exc
= self
._run
_failing
_graph
(SourceWithFailingIter
, WorkingSink
)
107 with self
.assertRaises(IndexError):
110 def test_exception_chaining(self
):
111 # Test that if we do:
115 # except bt2._Error as exc:
116 # raise ValueError('oh noes') from exc
118 # We are able to fetch the causes of the original bt2._Error in the
119 # exception chain. Also, each exception in the chain should become one
121 exc
= self
._run
_failing
_graph
(SourceWithFailingIter
, SinkWithExceptionChaining
)
123 self
.assertEqual(len(exc
), 5)
125 self
.assertIsInstance(exc
[0], bt2
._MessageIteratorErrorCause
)
126 self
.assertEqual(exc
[0].component_class_name
, 'SourceWithFailingIter')
127 self
.assertIn('ValueError: User message iterator is failing', exc
[0].message
)
129 self
.assertIsInstance(exc
[1], bt2
._ErrorCause
)
131 self
.assertIsInstance(exc
[2], bt2
._ComponentErrorCause
)
132 self
.assertEqual(exc
[2].component_class_name
, 'SinkWithExceptionChaining')
134 'unexpected error: cannot advance the message iterator', exc
[2].message
137 self
.assertIsInstance(exc
[3], bt2
._ComponentErrorCause
)
138 self
.assertEqual(exc
[3].component_class_name
, 'SinkWithExceptionChaining')
139 self
.assertIn('ValueError: oops', exc
[3].message
)
141 self
.assertIsInstance(exc
[4], bt2
._ErrorCause
)
143 def _common_cause_tests(self
, cause
):
144 self
.assertIsInstance(cause
.module_name
, str)
145 self
.assertIsInstance(cause
.file_name
, str)
146 self
.assertIsInstance(cause
.line_number
, int)
148 def test_unknown_error_cause(self
):
149 exc
= self
._run
_failing
_graph
(SourceWithFailingIter
, SinkWithExceptionChaining
)
151 self
.assertIs(type(cause
), bt2
._ErrorCause
)
152 self
._common
_cause
_tests
(cause
)
154 def test_component_error_cause(self
):
155 exc
= self
._run
_failing
_graph
(SourceWithFailingInit
, SinkWithExceptionChaining
)
157 self
.assertIs(type(cause
), bt2
._ComponentErrorCause
)
158 self
._common
_cause
_tests
(cause
)
160 self
.assertIn('Source is failing', cause
.message
)
161 self
.assertEqual(cause
.component_name
, 'src')
162 self
.assertEqual(cause
.component_class_type
, bt2
.ComponentClassType
.SOURCE
)
163 self
.assertEqual(cause
.component_class_name
, 'SourceWithFailingInit')
164 self
.assertIsNone(cause
.plugin_name
)
166 def test_component_class_error_cause(self
):
167 q
= bt2
.QueryExecutor(SinkWithFailingQuery
, 'hello')
169 with self
.assertRaises(bt2
._Error
) as ctx
:
172 cause
= ctx
.exception
[0]
173 self
.assertIs(type(cause
), bt2
._ComponentClassErrorCause
)
174 self
._common
_cause
_tests
(cause
)
176 self
.assertIn('Query is failing', cause
.message
)
178 self
.assertEqual(cause
.component_class_type
, bt2
.ComponentClassType
.SINK
)
179 self
.assertEqual(cause
.component_class_name
, 'SinkWithFailingQuery')
180 self
.assertIsNone(cause
.plugin_name
)
182 def test_message_iterator_error_cause(self
):
183 exc
= self
._run
_failing
_graph
(SourceWithFailingIter
, SinkWithExceptionChaining
)
185 self
.assertIs(type(cause
), bt2
._MessageIteratorErrorCause
)
186 self
._common
_cause
_tests
(cause
)
188 self
.assertIn('User message iterator is failing', cause
.message
)
189 self
.assertEqual(cause
.component_name
, 'src')
190 self
.assertEqual(cause
.component_output_port_name
, 'out')
191 self
.assertEqual(cause
.component_class_type
, bt2
.ComponentClassType
.SOURCE
)
192 self
.assertEqual(cause
.component_class_name
, 'SourceWithFailingIter')
193 self
.assertIsNone(cause
.plugin_name
)
196 # Test __str__. We don't need to test the precise format used, but
197 # just that it doesn't miserably crash and that it contains some
199 exc
= self
._run
_failing
_graph
(SourceWithFailingIter
, SinkWithExceptionChaining
)
201 self
.assertIn("[src (out): 'source.SourceWithFailingIter']", s
)
202 self
.assertIn('ValueError: oops', s
)
205 if __name__
== '__main__':