| # 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) |
