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