# Copyright 2014 The ChromiumOS Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""Test the failures_lib module."""

import json

from chromite.lib import cros_build_lib
from chromite.lib import cros_test_lib
from chromite.lib import failure_message_lib
from chromite.lib import failures_lib


class StepFailureTests(cros_test_lib.TestCase):
    """Tests for StepFailure."""

    def testConvertToStageFailureMessage(self):
        """Test ConvertToStageFailureMessage."""
        failure = failures_lib.StepFailure("step failure message")
        stage_failure_msg = failure.ConvertToStageFailureMessage(
            1, "HWTest [sanity]"
        )
        self.assertEqual(stage_failure_msg.stage_name, "HWTest [sanity]")
        self.assertEqual(stage_failure_msg.stage_prefix_name, "HWTest")
        self.assertEqual(stage_failure_msg.exception_type, "StepFailure")
        self.assertEqual(stage_failure_msg.exception_category, "unknown")


class CompoundFailureTest(cros_test_lib.TestCase):
    """Test the CompoundFailure class."""

    def _CreateExceptInfos(self, cls, message="", traceback="", num=1):
        """A helper function to create a list of ExceptInfo objects."""
        exc_infos = []
        for _ in range(num):
            exc_infos.extend(
                failures_lib.CreateExceptInfo(cls(message), traceback)
            )

        return exc_infos

    def testHasEmptyList(self):
        """Tests the HasEmptyList method."""
        self.assertTrue(failures_lib.CompoundFailure().HasEmptyList())
        exc_infos = self._CreateExceptInfos(KeyError)
        self.assertFalse(
            failures_lib.CompoundFailure(exc_infos=exc_infos).HasEmptyList()
        )

    def testHasAndMatchesFailureType(self):
        """Tests the HasFailureType and the MatchesFailureType methods."""
        # Create a CompoundFailure instance with mixed types of exceptions.
        exc_infos = self._CreateExceptInfos(KeyError)
        exc_infos.extend(self._CreateExceptInfos(ValueError))
        exc = failures_lib.CompoundFailure(exc_infos=exc_infos)
        self.assertTrue(exc.HasFailureType(KeyError))
        self.assertTrue(exc.HasFailureType(ValueError))
        self.assertFalse(exc.MatchesFailureType(KeyError))
        self.assertFalse(exc.MatchesFailureType(ValueError))

        # Create a CompoundFailure instance with a single type of exceptions.
        exc_infos = self._CreateExceptInfos(KeyError, num=5)
        exc = failures_lib.CompoundFailure(exc_infos=exc_infos)
        self.assertTrue(exc.HasFailureType(KeyError))
        self.assertFalse(exc.HasFailureType(ValueError))
        self.assertTrue(exc.MatchesFailureType(KeyError))
        self.assertFalse(exc.MatchesFailureType(ValueError))

    def testHasFatalFailure(self):
        """Tests the HasFatalFailure method."""
        exc_infos = self._CreateExceptInfos(KeyError)
        exc_infos.extend(self._CreateExceptInfos(ValueError))
        exc = failures_lib.CompoundFailure(exc_infos=exc_infos)
        self.assertTrue(exc.HasFatalFailure())
        self.assertTrue(exc.HasFatalFailure(exempt_exception_list=[KeyError]))
        self.assertFalse(
            exc.HasFatalFailure(exempt_exception_list=[KeyError, ValueError])
        )

        exc = failures_lib.CompoundFailure()
        self.assertFalse(exc.HasFatalFailure())

    def testMessageContainsAllInfo(self):
        """Tests that by default, all information is included in the message."""
        exc_infos = self._CreateExceptInfos(
            KeyError, message="bar1", traceback="foo1"
        )
        exc_infos.extend(
            self._CreateExceptInfos(
                ValueError, message="bar2", traceback="foo2"
            )
        )
        exc = failures_lib.CompoundFailure(exc_infos=exc_infos)
        self.assertIn("bar1", str(exc))
        self.assertIn("bar2", str(exc))
        self.assertIn("KeyError", str(exc))
        self.assertIn("ValueError", str(exc))
        self.assertIn("foo1", str(exc))
        self.assertIn("foo2", str(exc))

    def testConvertToStageFailureMessage(self):
        """Test ConvertToStageFailureMessage."""
        exc_infos = self._CreateExceptInfos(
            KeyError, message="bar1", traceback="foo1"
        )
        exc_infos.extend(
            self._CreateExceptInfos(
                failures_lib.StepFailure, message="bar2", traceback="foo2"
            )
        )
        exc = failures_lib.CompoundFailure(
            message="compound failure", exc_infos=exc_infos
        )
        stage_failure_msg = exc.ConvertToStageFailureMessage(
            1, "HWTest [sanity]"
        )

        self.assertEqual(len(stage_failure_msg.inner_failures), 2)
        self.assertEqual(stage_failure_msg.stage_name, "HWTest [sanity]")
        self.assertEqual(stage_failure_msg.stage_prefix_name, "HWTest")
        self.assertEqual(stage_failure_msg.exception_type, "CompoundFailure")
        self.assertEqual(stage_failure_msg.exception_category, "unknown")


class SetFailureTypeTest(cros_test_lib.TestCase):
    """Test that the SetFailureType decorator works."""

    ERROR_MESSAGE = "You failed!"

    class TacoNotTasty(failures_lib.CompoundFailure):
        """Raised when the taco is not tasty."""

    class NoGuacamole(TacoNotTasty):
        """Raised when no guacamole in the taco."""

    class SubparLunch(failures_lib.CompoundFailure):
        """Raised when the lunch is subpar."""

    class FooException(Exception):
        """A foo exception."""

    def _GetFunction(self, set_type, raise_type, *args, **kwargs):
        """Returns a function to test.

        Args:
            set_type: The exception type that the function is decorated with.
            raise_type: The exception type that the function raises.
            *args: args to pass to the instance of |raise_type|.

        Returns:
            The function to test.
        """

        @failures_lib.SetFailureType(set_type)
        def f():
            raise raise_type(*args, **kwargs)

        return f

    def testAssertionFailOnIllegalExceptionType(self):
        """Assertion should fail if the pre-set type is not allowed ."""
        self.assertRaises(
            AssertionError, self._GetFunction, ValueError, self.FooException
        )

    def testReraiseAsNewException(self):
        """Tests that the pre-set exception type is raised correctly."""
        try:
            self._GetFunction(
                self.TacoNotTasty, self.FooException, self.ERROR_MESSAGE
            )()
        except Exception as e:
            self.assertIsInstance(e, self.TacoNotTasty)
            self.assertTrue(e.msg, self.ERROR_MESSAGE)
            self.assertEqual(len(e.exc_infos), 1)
            self.assertEqual(e.exc_infos[0].str, self.ERROR_MESSAGE)
            self.assertEqual(e.exc_infos[0].type, self.FooException)
            self.assertIsInstance(e.exc_infos[0].traceback, str)

    def testReraiseACompoundFailure(self):
        """Tests that the list of ExceptInfo objects are copied over."""
        tb1 = "Stub traceback1"
        tb2 = "Stub traceback2"
        org_infos = failures_lib.CreateExceptInfo(
            ValueError("No taco."), tb1
        ) + failures_lib.CreateExceptInfo(OSError("No salsa"), tb2)
        try:
            self._GetFunction(
                self.SubparLunch, self.TacoNotTasty, exc_infos=org_infos
            )()
        except Exception as e:
            self.assertIsInstance(e, self.SubparLunch)
            # The orignal exceptions stored in exc_infos are preserved.
            self.assertEqual(e.exc_infos, org_infos)
            # All essential inforamtion should be included in the message of
            # the new excpetion.
            self.assertIn(tb1, str(e))
            self.assertIn(tb2, str(e))
            self.assertIn(str(ValueError), str(e))
            self.assertIn(str(OSError), str(e))
            self.assertIn(str("No taco"), str(e))
            self.assertIn(str("No salsa"), str(e))

            # Assert that summary does not contain the textual tracebacks.
            self.assertFalse(tb1 in e.ToSummaryString())
            self.assertFalse(tb2 in e.ToSummaryString())

    def testReraiseACompoundFailureWithEmptyList(self):
        """Tests that a CompoundFailure with empty list is handled correctly."""
        try:
            self._GetFunction(
                self.SubparLunch, self.TacoNotTasty, message="empty list"
            )()
        except Exception as e:
            self.assertIsInstance(e, self.SubparLunch)
            self.assertEqual(e.exc_infos[0].type, self.TacoNotTasty)

    def testReraiseOriginalException(self):
        """Tests that the original exception is re-raised."""
        # NoGuacamole is a subclass of TacoNotTasty, so the wrapper has no
        # effect on it.
        f = self._GetFunction(self.TacoNotTasty, self.NoGuacamole)
        self.assertRaises(self.NoGuacamole, f)

    def testPassArgsToWrappedFunctor(self):
        """Tests that we can pass arguments to the functor."""

        @failures_lib.SetFailureType(self.TacoNotTasty)
        def f(arg):
            return arg

        @failures_lib.SetFailureType(self.TacoNotTasty)
        def g(kwarg=""):
            return kwarg

        # Test passing arguments.
        self.assertEqual(f("foo"), "foo")
        # Test passing keyword arguments.
        self.assertEqual(g(kwarg="bar"), "bar")


class ExceptInfoTest(cros_test_lib.TestCase):
    """Tests the namedtuple class ExceptInfo."""

    def testConvertToExceptInfo(self):
        """Tests converting an exception to an ExceptInfo object."""
        traceback = "Stub traceback"
        message = "Taco is not a valid option!"
        except_infos = failures_lib.CreateExceptInfo(
            ValueError(message), traceback
        )

        self.assertEqual(except_infos[0].type, ValueError)
        self.assertEqual(except_infos[0].str, message)
        self.assertEqual(except_infos[0].traceback, traceback)


class FailureTypeListTests(cros_test_lib.TestCase):
    """Tests for failure type lists."""

    def testFailureTypeList(self):
        """Verify current failure names are already added to the type lists."""
        self.assertTrue(
            failures_lib.BuildScriptFailure.__name__
            in failure_message_lib.BUILD_SCRIPT_FAILURE_TYPES
        )
        self.assertTrue(
            failures_lib.PackageBuildFailure.__name__
            in failure_message_lib.PACKAGE_BUILD_FAILURE_TYPES
        )


class GetStageFailureMessageFromExceptionTests(cros_test_lib.TestCase):
    """Tests for GetStageFailureMessageFromException"""

    def testGetStageFailureMessageFromExceptionOnStepFailure(self):
        """Test GetStageFailureMessageFromException on StepFailure."""
        exc = failures_lib.StepFailure("step failure message")
        msg = failures_lib.GetStageFailureMessageFromException(
            "CommitQueueSync", 1, exc
        )
        self.assertEqual(msg.build_stage_id, 1)
        self.assertEqual(msg.stage_name, "CommitQueueSync")
        self.assertEqual(msg.stage_prefix_name, "CommitQueueSync")
        self.assertEqual(msg.exception_type, "StepFailure")
        self.assertEqual(msg.exception_category, "unknown")

    def testGetStageFailureMessageFromExceptionOnException(self):
        """Test GetStageFailureMessageFromException on regular exception."""
        exc = ValueError("Invalid valure.")
        msg = failures_lib.GetStageFailureMessageFromException(
            "CommitQueueSync", 1, exc
        )
        self.assertEqual(msg.build_stage_id, 1)
        self.assertEqual(msg.stage_name, "CommitQueueSync")
        self.assertEqual(msg.stage_prefix_name, "CommitQueueSync")
        self.assertEqual(msg.exception_type, "ValueError")
        self.assertEqual(msg.exception_category, "unknown")


class BuildFailuresForFindit(cros_test_lib.TestCase):
    """Test cases for exporting build failures for Findit integration."""

    def testBuildFailuresJson(self):
        error = cros_build_lib.RunCommandError("run cmd error")
        failed_packages = ["sys-apps/mosys", "chromeos-base/cryptohome"]
        build_failure = failures_lib.PackageBuildFailure(
            error, "build_packages", failed_packages
        )
        self.assertSetEqual(set(failed_packages), build_failure.failed_packages)
        failure_json = build_failure.BuildCompileFailureOutputJson()
        values = json.loads(failure_json)
        failures = values["failures"]
        self.assertEqual(len(failures), 2)
        # Verify both output targets are not equal, this makes sure the loop
        # below is correct.
        self.assertNotEqual(
            failures[0]["output_targets"], failures[1]["output_targets"]
        )
        for value in failures:
            self.assertEqual(value["rule"], "emerge")
            self.assertIn(value["output_targets"], failed_packages)