"""
Internal global error types
"""
import sys
import threading
import traceback
from traceback import format_exception
# Add names you want to be imported by 'from errors import *' to this list.
# This must be list not a tuple as we modify it to include all of our
# the Exception classes we define below at the end of this file.
__all__ = ['format_error', 'context_aware', 'context', 'get_context',
'exception_context']
# Exception context information:
# ------------------------------
# Every function can have some context string associated with it.
# The context string can be changed by calling context(str) and cleared by
# calling context() with no parameters.
# get_context() joins the current context strings of all functions in the
# provided traceback. The result is a brief description of what the test was
# doing in the provided traceback (which should be the traceback of a caught
# exception).
#
# For example: assume a() calls b() and b() calls c().
#
# @error.context_aware
# def a():
# error.context("hello")
# b()
# error.context("world")
# error.get_context() ----> 'world'
#
# @error.context_aware
# def b():
# error.context("foo")
# c()
#
# @error.context_aware
# def c():
# error.context("bar")
# error.get_context() ----> 'hello --> foo --> bar'
#
# The current context is automatically inserted into exceptions raised in
# context_aware functions, so usually test code doesn't need to call
# error.get_context().
ctx = threading.local()
def _new_context(s=""):
if not hasattr(ctx, "contexts"):
ctx.contexts = []
ctx.contexts.append(s)
def _pop_context():
ctx.contexts.pop()
[docs]def context(s="", log=None):
"""
Set the context for the currently executing function and optionally log it.
:param s: A string. If not provided, the context for the current function
will be cleared.
:param log: A logging function to pass the context message to. If None, no
function will be called.
"""
ctx.contexts[-1] = s
if s and log:
log("Context: %s" % get_context())
def base_context(s="", log=None):
"""
Set the base context for the currently executing function and optionally
log it. The base context is just another context level that is hidden by
default. Functions that require a single context level should not use
base_context().
:param s: A string. If not provided, the base context for the current
function will be cleared.
:param log: A logging function to pass the context message to. If None, no
function will be called.
"""
ctx.contexts[-1] = ""
ctx.contexts[-2] = s
if s and log:
log("Context: %s" % get_context())
[docs]def get_context():
"""Return the current context (or None if none is defined)."""
if hasattr(ctx, "contexts"):
return " --> ".join([s for s in ctx.contexts if s])
[docs]def exception_context(e):
"""Return the context of a given exception (or None if none is defined)."""
if hasattr(e, "_context"):
return e._context
def set_exception_context(e, s):
"""Set the context of a given exception."""
e._context = s
def join_contexts(s1, s2):
"""Join two context strings."""
if s1:
if s2:
return "%s --> %s" % (s1, s2)
else:
return s1
else:
return s2
[docs]def context_aware(fn):
"""A decorator that must be applied to functions that call context()."""
def new_fn(*args, **kwargs):
_new_context()
_new_context("(%s)" % fn.__name__)
try:
try:
return fn(*args, **kwargs)
except Exception, e:
if not exception_context(e):
set_exception_context(e, get_context())
raise
finally:
_pop_context()
_pop_context()
new_fn.__name__ = fn.__name__
new_fn.__doc__ = fn.__doc__
new_fn.__dict__.update(fn.__dict__)
return new_fn
def _context_message(e):
s = exception_context(e)
if s:
return " [context: %s]" % s
else:
return ""
class JobContinue(SystemExit):
"""Allow us to bail out requesting continuance."""
pass
class JobComplete(SystemExit):
"""Allow us to bail out indicating continuation not required."""
pass
[docs]class AutotestError(Exception):
"""The parent of all errors deliberately thrown within the client code."""
def __str__(self):
return Exception.__str__(self) + _context_message(self)
[docs]class JobError(AutotestError):
"""Indicates an error which terminates and fails the whole job (ABORT)."""
pass
[docs]class UnhandledJobError(JobError):
"""Indicates an unhandled error in a job."""
def __init__(self, unhandled_exception):
if isinstance(unhandled_exception, JobError):
JobError.__init__(self, *unhandled_exception.args)
elif isinstance(unhandled_exception, str):
JobError.__init__(self, unhandled_exception)
else:
msg = "Unhandled %s: %s"
msg %= (unhandled_exception.__class__.__name__,
unhandled_exception)
if not isinstance(unhandled_exception, AutotestError):
msg += _context_message(unhandled_exception)
msg += "\n" + traceback.format_exc()
JobError.__init__(self, msg)
[docs]class TestBaseException(AutotestError):
"""The parent of all test exceptions."""
# Children are required to override this. Never instantiate directly.
exit_status = "NEVER_RAISE_THIS"
[docs]class TestError(TestBaseException):
"""Indicates that something went wrong with the test harness itself."""
exit_status = "ERROR"
[docs]class TestNAError(TestBaseException):
"""Indictates that the test is Not Applicable. Should be thrown
when various conditions are such that the test is inappropriate."""
exit_status = "TEST_NA"
[docs]class TestFail(TestBaseException):
"""Indicates that the test failed, but the job will not continue."""
exit_status = "FAIL"
[docs]class TestBug(TestBaseException):
"""Indicates that the test failed, but the fail was expected."""
exit_status = "BUG"
[docs]class TestWarn(TestBaseException):
"""Indicates that bad things (may) have happened, but not an explicit
failure."""
exit_status = "WARN"
[docs]class UnhandledTestError(TestError):
"""Indicates an unhandled error in a test."""
def __init__(self, unhandled_exception):
if isinstance(unhandled_exception, TestError):
TestError.__init__(self, *unhandled_exception.args)
elif isinstance(unhandled_exception, str):
TestError.__init__(self, unhandled_exception)
else:
msg = "Unhandled %s: %s"
msg %= (unhandled_exception.__class__.__name__,
unhandled_exception)
if not isinstance(unhandled_exception, AutotestError):
msg += _context_message(unhandled_exception)
msg += "\n" + traceback.format_exc()
TestError.__init__(self, msg)
[docs]class UnhandledTestFail(TestFail):
"""Indicates an unhandled fail in a test."""
def __init__(self, unhandled_exception):
if isinstance(unhandled_exception, TestFail):
TestFail.__init__(self, *unhandled_exception.args)
elif isinstance(unhandled_exception, str):
TestFail.__init__(self, unhandled_exception)
else:
msg = "Unhandled %s: %s"
msg %= (unhandled_exception.__class__.__name__,
unhandled_exception)
if not isinstance(unhandled_exception, AutotestError):
msg += _context_message(unhandled_exception)
msg += "\n" + traceback.format_exc()
TestFail.__init__(self, msg)
[docs]class CmdError(TestError):
"""
Indicates that a command failed, is fatal to the test unless caught.
"""
def __init__(self, command, result_obj, additional_text=None):
TestError.__init__(self, command, result_obj, additional_text)
self.command = command
self.result_obj = result_obj
self.additional_text = additional_text
def __str__(self):
if self.result_obj.exit_status is None:
msg = "Command <%s> failed and is not responding to signals"
msg %= self.command
else:
msg = "Command <%s> failed, rc=%d"
msg %= (self.command, self.result_obj.exit_status)
if self.additional_text:
msg += ", " + self.additional_text
msg += _context_message(self)
msg += '\n' + repr(self.result_obj)
return msg
[docs]class PackageError(TestError):
"""Indicates an error trying to perform a package operation."""
pass
[docs]class BarrierError(JobError):
"""Indicates an error happened during a barrier operation."""
pass
[docs]class BarrierAbortError(BarrierError):
"""Indicate that the barrier was explicitly aborted by a member."""
pass
[docs]class NetCommunicationError(JobError):
"""Indicate that network communication was broken."""
pass
[docs]class DataSyncError(NetCommunicationError):
"""Indicates problem during synchronization data over network."""
pass
[docs]class HarnessError(JobError):
"""Indicates problem with the harness."""
pass
[docs]class InstallError(JobError):
"""Indicates an installation error which Terminates and fails the job."""
pass
[docs]class AutotestRunError(AutotestError):
"""Indicates a problem running server side control files."""
pass
[docs]class AutotestTimeoutError(AutotestError):
"""This exception is raised when an autotest test exceeds the timeout
parameter passed to run_timed_test and is killed.
"""
pass
[docs]class HostRunErrorMixIn(Exception):
"""
Indicates a problem in the host run() function raised from client code.
Should always be constructed with a tuple of two args (error description
(str), run result object). This is a common class mixed in to create the
client and server side versions of it.
"""
def __init__(self, description, result_obj):
self.description = description
self.result_obj = result_obj
Exception.__init__(self, description, result_obj)
def __str__(self):
return self.description + '\n' + repr(self.result_obj)
[docs]class HostInstallTimeoutError(JobError):
"""
Indicates the machine failed to be installed after the predetermined
timeout.
"""
pass
[docs]class HostInstallProfileError(JobError):
"""
Indicates the machine failed to have a profile assigned.
"""
pass
[docs]class AutotestHostRunError(HostRunErrorMixIn, AutotestError):
pass
# server-specific errors
[docs]class AutoservError(Exception):
pass
[docs]class AutoservSSHTimeout(AutoservError):
"""SSH experienced a connection timeout"""
pass
[docs]class AutoservRunError(HostRunErrorMixIn, AutoservError):
pass
[docs]class AutoservSshPermissionDeniedError(AutoservRunError):
"""Indicates that a SSH permission denied error was encountered."""
pass
[docs]class AutoservVirtError(AutoservError):
"""Vitualization related error"""
pass
[docs]class AutoservUnsupportedError(AutoservError):
"""Error raised when you try to use an unsupported optional feature"""
pass
[docs]class AutoservHostError(AutoservError):
"""Error reaching a host"""
pass
[docs]class AutoservHostIsShuttingDownError(AutoservHostError):
"""Host is shutting down"""
pass
[docs]class AutoservNotMountedHostError(AutoservHostError):
"""Found unmounted partitions that should be mounted"""
pass
[docs]class AutoservSshPingHostError(AutoservHostError):
"""SSH ping failed"""
pass
[docs]class AutoservDiskFullHostError(AutoservHostError):
"""Not enough free disk space on host"""
def __init__(self, path, want_gb, free_space_gb):
AutoservHostError.__init__(self,
'Not enough free space on %s - %.3fGB free, want %.3fGB' %
(path, free_space_gb, want_gb))
self.path = path
self.want_gb = want_gb
self.free_space_gb = free_space_gb
[docs]class AutoservHardwareHostError(AutoservHostError):
"""Found hardware problems with the host"""
pass
[docs]class AutoservRebootError(AutoservError):
"""Error occurred while rebooting a machine"""
pass
[docs]class AutoservShutdownError(AutoservRebootError):
"""Error occurred during shutdown of machine"""
pass
[docs]class AutoservSubcommandError(AutoservError):
"""Indicates an error while executing a (forked) subcommand"""
def __init__(self, func, exit_code):
AutoservError.__init__(self, func, exit_code)
self.func = func
self.exit_code = exit_code
def __str__(self):
return ("Subcommand %s failed with exit code %d" %
(self.func, self.exit_code))
[docs]class AutoservHardwareRepairRequestedError(AutoservError):
"""
Exception class raised from Host.repair_full() (or overrides) when software
repair fails but it successfully managed to request a hardware repair (by
notifying the staff, sending mail, etc)
"""
pass
[docs]class AutoservHardwareRepairRequiredError(AutoservError):
"""
Exception class raised during repairs to indicate that a hardware repair
is going to be necessary.
"""
pass
[docs]class AutoservInstallError(AutoservError):
"""Error occurred while installing autotest on a host"""
pass
# packaging system errors
[docs]class PackagingError(AutotestError):
'Abstract error class for all packaging related errors.'
[docs]class PackageUploadError(PackagingError):
'Raised when there is an error uploading the package'
[docs]class PackageFetchError(PackagingError):
'Raised when there is an error fetching the package'
[docs]class PackageRemoveError(PackagingError):
'Raised when there is an error removing the package'
[docs]class PackageInstallError(PackagingError):
'Raised when there is an error installing the package'
[docs]class RepoDiskFullError(PackagingError):
'Raised when the destination for packages is full'
[docs]class RepoWriteError(PackagingError):
"Raised when packager cannot write to a repo's desitnation"
[docs]class RepoUnknownError(PackagingError):
"Raised when packager cannot write to a repo's desitnation"
[docs]class RepoError(PackagingError):
"Raised when a repo isn't working in some way"
# This MUST remain at the end of the file.
# Limit 'from error import *' to only import the exception instances.
for _name, _thing in locals().items():
try:
if issubclass(_thing, Exception):
__all__.append(_name)
except TypeError:
pass # _thing not a class
__all__ = tuple(__all__)