tests/bindings/python/bt2/test_error.py

   1 # SPDX-License-Identifier: GPL-2.0-only
   2 #
   3 # Copyright (C) 2019 EfficiOS Inc.
   4 #
   5
   6 import unittest
   7
   8 import bt2
   9 from bt2 import native_bt
  10
  11
  12 class FailingIter(bt2._UserMessageIterator):
  13     def __next__(self):
  14         raise ValueError("User message iterator is failing")
  15
  16
  17 class SourceWithFailingIter(
  18     bt2._UserSourceComponent, message_iterator_class=FailingIter
  19 ):
  20     def __init__(self, config, params, obj):
  21         self._add_output_port("out")
  22
  23
  24 class SourceWithFailingInit(
  25     bt2._UserSourceComponent, message_iterator_class=FailingIter
  26 ):
  27     def __init__(self, config, params, obj):
  28         raise ValueError("Source is failing")
  29
  30
  31 class WorkingSink(bt2._UserSinkComponent):
  32     def __init__(self, config, params, obj):
  33         self._in = self._add_input_port("in")
  34
  35     def _user_graph_is_configured(self):
  36         self._iter = self._create_message_iterator(self._in)
  37
  38     def _user_consume(self):
  39         next(self._iter)
  40
  41
  42 class SinkWithExceptionChaining(bt2._UserSinkComponent):
  43     def __init__(self, config, params, obj):
  44         self._in = self._add_input_port("in")
  45
  46     def _user_graph_is_configured(self):
  47         self._iter = self._create_message_iterator(self._in)
  48
  49     def _user_consume(self):
  50         try:
  51             next(self._iter)
  52         except bt2._Error as e:
  53             raise ValueError("oops") from e
  54
  55
  56 class SinkWithFailingQuery(bt2._UserSinkComponent):
  57     def _user_consume(self):
  58         pass
  59
  60     @staticmethod
  61     def _user_query(priv_executor, obj, params, method_obj):
  62         raise ValueError("Query is failing")
  63
  64
  65 class ErrorTestCase(unittest.TestCase):
  66     def _run_failing_graph(self, source_cc, sink_cc):
  67         with self.assertRaises(bt2._Error) as ctx:
  68             graph = bt2.Graph()
  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"])
  72             graph.run()
  73
  74         return ctx.exception
  75
  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())
  81
  82     def test_len(self):
  83         exc = self._run_failing_graph(SourceWithFailingIter, WorkingSink)
  84
  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)
  89
  90     def test_iter(self):
  91         exc = self._run_failing_graph(SourceWithFailingIter, WorkingSink)
  92
  93         for c in exc:
  94             # Each cause is an instance of _ErrorCause (including subclasses).
  95             self.assertIsInstance(c, bt2._ErrorCause)
  96
  97     def test_getitem(self):
  98         exc = self._run_failing_graph(SourceWithFailingIter, WorkingSink)
  99
 100         for i in range(len(exc)):
 101             c = exc[i]
 102             # Each cause is an instance of _ErrorCause (including subclasses).
 103             self.assertIsInstance(c, bt2._ErrorCause)
 104
 105     def test_getitem_indexerror(self):
 106         exc = self._run_failing_graph(SourceWithFailingIter, WorkingSink)
 107
 108         with self.assertRaises(IndexError):
 109             exc[len(exc)]
 110
 111     def test_exception_chaining(self):
 112         # Test that if we do:
 113         #
 114         # try:
 115         #     ...
 116         # except bt2._Error as exc:
 117         #     raise ValueError('oh noes') from exc
 118         #
 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
 121         # cause once caught.
 122         exc = self._run_failing_graph(SourceWithFailingIter, SinkWithExceptionChaining)
 123
 124         self.assertEqual(len(exc), 5)
 125
 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)
 129
 130         self.assertIsInstance(exc[1], bt2._ErrorCause)
 131
 132         self.assertIsInstance(exc[2], bt2._ComponentErrorCause)
 133         self.assertEqual(exc[2].component_class_name, "SinkWithExceptionChaining")
 134         self.assertIn(
 135             "unexpected error: cannot advance the message iterator", exc[2].message
 136         )
 137
 138         self.assertIsInstance(exc[3], bt2._ComponentErrorCause)
 139         self.assertEqual(exc[3].component_class_name, "SinkWithExceptionChaining")
 140         self.assertIn("ValueError: oops", exc[3].message)
 141
 142         self.assertIsInstance(exc[4], bt2._ErrorCause)
 143
 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)
 148
 149     def test_unknown_error_cause(self):
 150         exc = self._run_failing_graph(SourceWithFailingIter, SinkWithExceptionChaining)
 151         cause = exc[-1]
 152         self.assertIs(type(cause), bt2._ErrorCause)
 153         self._common_cause_tests(cause)
 154
 155     def test_component_error_cause(self):
 156         exc = self._run_failing_graph(SourceWithFailingInit, SinkWithExceptionChaining)
 157         cause = exc[0]
 158         self.assertIs(type(cause), bt2._ComponentErrorCause)
 159         self._common_cause_tests(cause)
 160
 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)
 166
 167     def test_component_class_error_cause(self):
 168         q = bt2.QueryExecutor(SinkWithFailingQuery, "hello")
 169
 170         with self.assertRaises(bt2._Error) as ctx:
 171             q.query()
 172
 173         cause = ctx.exception[0]
 174         self.assertIs(type(cause), bt2._ComponentClassErrorCause)
 175         self._common_cause_tests(cause)
 176
 177         self.assertIn("Query is failing", cause.message)
 178
 179         self.assertEqual(cause.component_class_type, bt2.ComponentClassType.SINK)
 180         self.assertEqual(cause.component_class_name, "SinkWithFailingQuery")
 181         self.assertIsNone(cause.plugin_name)
 182
 183     def test_message_iterator_error_cause(self):
 184         exc = self._run_failing_graph(SourceWithFailingIter, SinkWithExceptionChaining)
 185         cause = exc[0]
 186         self.assertIs(type(cause), bt2._MessageIteratorErrorCause)
 187         self._common_cause_tests(cause)
 188
 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)
 195
 196     def test_str(self):
 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
 199         # expected bits.
 200         exc = self._run_failing_graph(SourceWithFailingIter, SinkWithExceptionChaining)
 201         s = str(exc)
 202         self.assertIn("[src (out): 'source.SourceWithFailingIter']", s)
 203         self.assertIn("ValueError: oops", s)
 204
 205
 206 if __name__ == "__main__":
 207     unittest.main()