1 # SPDX-License-Identifier: GPL-2.0-only
3 # Copyright (C) 2019 EfficiOS Inc.
9 from bt2
import native_bt
12 class FailingIter(bt2
._UserMessageIterator
):
14 raise ValueError("User message iterator is failing")
17 class SourceWithFailingIter(
18 bt2
._UserSourceComponent
, message_iterator_class
=FailingIter
20 def __init__(self
, config
, params
, obj
):
21 self
._add
_output
_port
("out")
24 class SourceWithFailingInit(
25 bt2
._UserSourceComponent
, message_iterator_class
=FailingIter
27 def __init__(self
, config
, params
, obj
):
28 raise ValueError("Source is failing")
31 class WorkingSink(bt2
._UserSinkComponent
):
32 def __init__(self
, config
, params
, obj
):
33 self
._in
= self
._add
_input
_port
("in")
35 def _user_graph_is_configured(self
):
36 self
._iter
= self
._create
_message
_iterator
(self
._in
)
38 def _user_consume(self
):
42 class SinkWithExceptionChaining(bt2
._UserSinkComponent
):
43 def __init__(self
, config
, params
, obj
):
44 self
._in
= self
._add
_input
_port
("in")
46 def _user_graph_is_configured(self
):
47 self
._iter
= self
._create
_message
_iterator
(self
._in
)
49 def _user_consume(self
):
52 except bt2
._Error
as e
:
53 raise ValueError("oops") from e
56 class SinkWithFailingQuery(bt2
._UserSinkComponent
):
57 def _user_consume(self
):
61 def _user_query(priv_executor
, obj
, params
, method_obj
):
62 raise ValueError("Query is failing")
65 class ErrorTestCase(unittest
.TestCase
):
66 def _run_failing_graph(self
, source_cc
, sink_cc
):
67 with self
.assertRaises(bt2
._Error
) as ctx
:
69 src
= graph
.add_component(source_cc
, "src")
70 snk
= graph
.add_component(sink_cc
, "snk")
71 graph
.connect_ports(src
.output_ports
["out"], snk
.input_ports
["in"])
76 def test_current_thread_error_none(self
):
77 # When a bt2._Error is raised, it steals the current thread's error.
78 # Verify that it is now NULL.
79 self
._run
_failing
_graph
(SourceWithFailingInit
, WorkingSink
)
80 self
.assertIsNone(native_bt
.current_thread_take_error())
83 exc
= self
._run
_failing
_graph
(SourceWithFailingIter
, WorkingSink
)
85 # The exact number of causes is not too important (it can change if we
86 # append more or less causes along the way), but the idea is to verify is
87 # has a value that makes sense.
88 self
.assertEqual(len(exc
), 4)
91 exc
= self
._run
_failing
_graph
(SourceWithFailingIter
, WorkingSink
)
94 # Each cause is an instance of _ErrorCause (including subclasses).
95 self
.assertIsInstance(c
, bt2
._ErrorCause
)
97 def test_getitem(self
):
98 exc
= self
._run
_failing
_graph
(SourceWithFailingIter
, WorkingSink
)
100 for i
in range(len(exc
)):
102 # Each cause is an instance of _ErrorCause (including subclasses).
103 self
.assertIsInstance(c
, bt2
._ErrorCause
)
105 def test_getitem_indexerror(self
):
106 exc
= self
._run
_failing
_graph
(SourceWithFailingIter
, WorkingSink
)
108 with self
.assertRaises(IndexError):
111 def test_exception_chaining(self
):
112 # Test that if we do:
116 # except bt2._Error as exc:
117 # raise ValueError('oh noes') from exc
119 # We are able to fetch the causes of the original bt2._Error in the
120 # exception chain. Also, each exception in the chain should become one
122 exc
= self
._run
_failing
_graph
(SourceWithFailingIter
, SinkWithExceptionChaining
)
124 self
.assertEqual(len(exc
), 5)
126 self
.assertIsInstance(exc
[0], bt2
._MessageIteratorErrorCause
)
127 self
.assertEqual(exc
[0].component_class_name
, "SourceWithFailingIter")
128 self
.assertIn("ValueError: User message iterator is failing", exc
[0].message
)
130 self
.assertIsInstance(exc
[1], bt2
._ErrorCause
)
132 self
.assertIsInstance(exc
[2], bt2
._ComponentErrorCause
)
133 self
.assertEqual(exc
[2].component_class_name
, "SinkWithExceptionChaining")
135 "unexpected error: cannot advance the message iterator", exc
[2].message
138 self
.assertIsInstance(exc
[3], bt2
._ComponentErrorCause
)
139 self
.assertEqual(exc
[3].component_class_name
, "SinkWithExceptionChaining")
140 self
.assertIn("ValueError: oops", exc
[3].message
)
142 self
.assertIsInstance(exc
[4], bt2
._ErrorCause
)
144 def _common_cause_tests(self
, cause
):
145 self
.assertIsInstance(cause
.module_name
, str)
146 self
.assertIsInstance(cause
.file_name
, str)
147 self
.assertIsInstance(cause
.line_number
, int)
149 def test_unknown_error_cause(self
):
150 exc
= self
._run
_failing
_graph
(SourceWithFailingIter
, SinkWithExceptionChaining
)
152 self
.assertIs(type(cause
), bt2
._ErrorCause
)
153 self
._common
_cause
_tests
(cause
)
155 def test_component_error_cause(self
):
156 exc
= self
._run
_failing
_graph
(SourceWithFailingInit
, SinkWithExceptionChaining
)
158 self
.assertIs(type(cause
), bt2
._ComponentErrorCause
)
159 self
._common
_cause
_tests
(cause
)
161 self
.assertIn("Source is failing", cause
.message
)
162 self
.assertEqual(cause
.component_name
, "src")
163 self
.assertEqual(cause
.component_class_type
, bt2
.ComponentClassType
.SOURCE
)
164 self
.assertEqual(cause
.component_class_name
, "SourceWithFailingInit")
165 self
.assertIsNone(cause
.plugin_name
)
167 def test_component_class_error_cause(self
):
168 q
= bt2
.QueryExecutor(SinkWithFailingQuery
, "hello")
170 with self
.assertRaises(bt2
._Error
) as ctx
:
173 cause
= ctx
.exception
[0]
174 self
.assertIs(type(cause
), bt2
._ComponentClassErrorCause
)
175 self
._common
_cause
_tests
(cause
)
177 self
.assertIn("Query is failing", cause
.message
)
179 self
.assertEqual(cause
.component_class_type
, bt2
.ComponentClassType
.SINK
)
180 self
.assertEqual(cause
.component_class_name
, "SinkWithFailingQuery")
181 self
.assertIsNone(cause
.plugin_name
)
183 def test_message_iterator_error_cause(self
):
184 exc
= self
._run
_failing
_graph
(SourceWithFailingIter
, SinkWithExceptionChaining
)
186 self
.assertIs(type(cause
), bt2
._MessageIteratorErrorCause
)
187 self
._common
_cause
_tests
(cause
)
189 self
.assertIn("User message iterator is failing", cause
.message
)
190 self
.assertEqual(cause
.component_name
, "src")
191 self
.assertEqual(cause
.component_output_port_name
, "out")
192 self
.assertEqual(cause
.component_class_type
, bt2
.ComponentClassType
.SOURCE
)
193 self
.assertEqual(cause
.component_class_name
, "SourceWithFailingIter")
194 self
.assertIsNone(cause
.plugin_name
)
197 # Test __str__. We don't need to test the precise format used, but
198 # just that it doesn't miserably crash and that it contains some
200 exc
= self
._run
_failing
_graph
(SourceWithFailingIter
, SinkWithExceptionChaining
)
202 self
.assertIn("[src (out): 'source.SourceWithFailingIter']", s
)
203 self
.assertIn("ValueError: oops", s
)
206 if __name__
== "__main__":