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_input_port_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_input_port_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()