bt2c::Logger: remove unused cLevel() method

[babeltrace.git] / tests / bindings / python / bt2 / test_error.py

# SPDX-License-Identifier: GPL-2.0-only
#
# Copyright (C) 2019 EfficiOS Inc.
#

import unittest

import bt2
from bt2 import native_bt


class FailingIter(bt2._UserMessageIterator):
    def __next__(self):
        raise ValueError("User message iterator is failing")


class SourceWithFailingIter(
    bt2._UserSourceComponent, message_iterator_class=FailingIter
):
    def __init__(self, config, params, obj):
        self._add_output_port("out")


class SourceWithFailingInit(
    bt2._UserSourceComponent, message_iterator_class=FailingIter
):
    def __init__(self, config, params, obj):
        raise ValueError("Source is failing")


class WorkingSink(bt2._UserSinkComponent):
    def __init__(self, config, params, obj):
        self._in = self._add_input_port("in")

    def _user_graph_is_configured(self):
        self._iter = self._create_message_iterator(self._in)

    def _user_consume(self):
        next(self._iter)


class SinkWithExceptionChaining(bt2._UserSinkComponent):
    def __init__(self, config, params, obj):
        self._in = self._add_input_port("in")

    def _user_graph_is_configured(self):
        self._iter = self._create_message_iterator(self._in)

    def _user_consume(self):
        try:
            next(self._iter)
        except bt2._Error as e:
            raise ValueError("oops") from e


class SinkWithFailingQuery(bt2._UserSinkComponent):
    def _user_consume(self):
        pass

    @staticmethod
    def _user_query(priv_executor, obj, params, method_obj):
        raise ValueError("Query is failing")


class ErrorTestCase(unittest.TestCase):
    def _run_failing_graph(self, source_cc, sink_cc):
        with self.assertRaises(bt2._Error) as ctx:
            graph = bt2.Graph()
            src = graph.add_component(source_cc, "src")
            snk = graph.add_component(sink_cc, "snk")
            graph.connect_ports(src.output_ports["out"], snk.input_ports["in"])
            graph.run()

        return ctx.exception

    def test_current_thread_error_none(self):
        # When a bt2._Error is raised, it steals the current thread's error.
        # Verify that it is now NULL.
        self._run_failing_graph(SourceWithFailingInit, WorkingSink)
        self.assertIsNone(native_bt.current_thread_take_error())

    def test_len(self):
        exc = self._run_failing_graph(SourceWithFailingIter, WorkingSink)

        # The exact number of causes is not too important (it can change if we
        # append more or less causes along the way), but the idea is to verify is
        # has a value that makes sense.
        self.assertEqual(len(exc), 4)

    def test_iter(self):
        exc = self._run_failing_graph(SourceWithFailingIter, WorkingSink)

        for c in exc:
            # Each cause is an instance of _ErrorCause (including subclasses).
            self.assertIsInstance(c, bt2._ErrorCause)

    def test_getitem(self):
        exc = self._run_failing_graph(SourceWithFailingIter, WorkingSink)

        for i in range(len(exc)):
            c = exc[i]
            # Each cause is an instance of _ErrorCause (including subclasses).
            self.assertIsInstance(c, bt2._ErrorCause)

    def test_getitem_indexerror(self):
        exc = self._run_failing_graph(SourceWithFailingIter, WorkingSink)

        with self.assertRaises(IndexError):
            exc[len(exc)]

    def test_exception_chaining(self):
        # Test that if we do:
        #
        # try:
        #     ...
        # except bt2._Error as exc:
        #     raise ValueError('oh noes') from exc
        #
        # We are able to fetch the causes of the original bt2._Error in the
        # exception chain.  Also, each exception in the chain should become one
        # cause once caught.
        exc = self._run_failing_graph(SourceWithFailingIter, SinkWithExceptionChaining)

        self.assertEqual(len(exc), 5)

        self.assertIsInstance(exc[0], bt2._MessageIteratorErrorCause)
        self.assertEqual(exc[0].component_class_name, "SourceWithFailingIter")
        self.assertIn("ValueError: User message iterator is failing", exc[0].message)

        self.assertIsInstance(exc[1], bt2._ErrorCause)

        self.assertIsInstance(exc[2], bt2._ComponentErrorCause)
        self.assertEqual(exc[2].component_class_name, "SinkWithExceptionChaining")
        self.assertIn(
            "unexpected error: cannot advance the message iterator", exc[2].message
        )

        self.assertIsInstance(exc[3], bt2._ComponentErrorCause)
        self.assertEqual(exc[3].component_class_name, "SinkWithExceptionChaining")
        self.assertIn("ValueError: oops", exc[3].message)

        self.assertIsInstance(exc[4], bt2._ErrorCause)

    def _common_cause_tests(self, cause):
        self.assertIsInstance(cause.module_name, str)
        self.assertIsInstance(cause.file_name, str)
        self.assertIsInstance(cause.line_number, int)

    def test_unknown_error_cause(self):
        exc = self._run_failing_graph(SourceWithFailingIter, SinkWithExceptionChaining)
        cause = exc[-1]
        self.assertIs(type(cause), bt2._ErrorCause)
        self._common_cause_tests(cause)

    def test_component_error_cause(self):
        exc = self._run_failing_graph(SourceWithFailingInit, SinkWithExceptionChaining)
        cause = exc[0]
        self.assertIs(type(cause), bt2._ComponentErrorCause)
        self._common_cause_tests(cause)

        self.assertIn("Source is failing", cause.message)
        self.assertEqual(cause.component_name, "src")
        self.assertEqual(cause.component_class_type, bt2.ComponentClassType.SOURCE)
        self.assertEqual(cause.component_class_name, "SourceWithFailingInit")
        self.assertIsNone(cause.plugin_name)

    def test_component_class_error_cause(self):
        q = bt2.QueryExecutor(SinkWithFailingQuery, "hello")

        with self.assertRaises(bt2._Error) as ctx:
            q.query()

        cause = ctx.exception[0]
        self.assertIs(type(cause), bt2._ComponentClassErrorCause)
        self._common_cause_tests(cause)

        self.assertIn("Query is failing", cause.message)

        self.assertEqual(cause.component_class_type, bt2.ComponentClassType.SINK)
        self.assertEqual(cause.component_class_name, "SinkWithFailingQuery")
        self.assertIsNone(cause.plugin_name)

    def test_message_iterator_error_cause(self):
        exc = self._run_failing_graph(SourceWithFailingIter, SinkWithExceptionChaining)
        cause = exc[0]
        self.assertIs(type(cause), bt2._MessageIteratorErrorCause)
        self._common_cause_tests(cause)

        self.assertIn("User message iterator is failing", cause.message)
        self.assertEqual(cause.component_name, "src")
        self.assertEqual(cause.component_output_port_name, "out")
        self.assertEqual(cause.component_class_type, bt2.ComponentClassType.SOURCE)
        self.assertEqual(cause.component_class_name, "SourceWithFailingIter")
        self.assertIsNone(cause.plugin_name)

    def test_str(self):
        # Test __str__.  We don't need to test the precise format used, but
        # just that it doesn't miserably crash and that it contains some
        # expected bits.
        exc = self._run_failing_graph(SourceWithFailingIter, SinkWithExceptionChaining)
        s = str(exc)
        self.assertIn("[src (out): 'source.SourceWithFailingIter']", s)
        self.assertIn("ValueError: oops", s)


if __name__ == "__main__":
    unittest.main()