# Copyright 2016-2022 Swiss National Supercomputing Centre (CSCS/ETH Zurich)
# ReFrame Project Developers. See the top-level LICENSE file for details.
#
# SPDX-License-Identifier: BSD-3-Clause
#
# Base regression exceptions
#
import contextlib
import inspect
import os
import reframe
import reframe.utility as utility
[docs]class ReframeBaseError(BaseException):
'''Base exception for any ReFrame error.
This exception base class offers a specialized :func:`__str__` method that
concatenates the messages of a chain of exceptions by inspecting their
:py:data:`__cause__` field. For example, the following piece of code will
print ``error message 2: error message 1``:
.. code-block:: python
from reframe.core.exceptions import *
def foo():
raise ReframeError('error message 1)
def bar():
try:
foo()
except ReframeError as e:
raise ReframeError('error message 2') from e
if __name__ == '__main__':
try:
bar()
except Exception as e:
print(e)
'''
def __init__(self, *args):
self._message = str(args[0]) if args else None
@property
def message(self):
return self._message
def __str__(self):
ret = self._message or ''
if self.__cause__ is not None:
ret += ': ' + str(self.__cause__)
return ret
[docs]class ReframeError(ReframeBaseError, Exception):
'''Base exception for soft errors.
Soft errors may be treated by simply printing the exception's message and
trying to continue execution if possible.
'''
[docs]class ReframeFatalError(ReframeBaseError):
'''A fatal framework error.
Execution must be aborted.
'''
[docs]class ReframeSyntaxError(ReframeError):
'''Raised when the syntax of regression tests is incorrect.'''
[docs]class RegressionTestLoadError(ReframeError):
'''Raised when the regression test cannot be loaded.'''
[docs]class NameConflictError(RegressionTestLoadError):
'''Raised when there is a name clash in the test suite.'''
[docs]class TaskExit(ReframeError):
'''Raised when a regression task must exit the pipeline prematurely.'''
[docs]class TaskDependencyError(ReframeError):
'''Raised inside a regression task by the runtime when one of its
dependencies has failed.
'''
[docs]class FailureLimitError(ReframeError):
'''Raised when the limit of test failures has been reached.'''
[docs]class AbortTaskError(ReframeError):
'''Raised by the runtime inside a regression task to denote that it has
been aborted due to an external reason (e.g., keyboard interrupt, fatal
error in other places etc.)
'''
[docs]class ConfigError(ReframeError):
'''Raised when a configuration error occurs.'''
[docs]class LoggingError(ReframeError):
'''Raised when an error related to logging has occurred.'''
[docs]class EnvironError(ReframeError):
'''Raised when an error related to an environment occurs.'''
[docs]class SanityError(ReframeError):
'''Raised to denote an error in sanity checking.'''
[docs]class PipelineError(ReframeError):
'''Raised when a condition prevents the regression test pipeline to
continue and the error may not be described by another more specific
exception.
'''
[docs]class ForceExitError(ReframeError):
'''Raised when ReFrame execution must be forcefully ended,
e.g., after a SIGTERM was received.
'''
[docs]class StatisticsError(ReframeError):
'''Raised to denote an error in dealing with statistics.'''
[docs]class BuildSystemError(ReframeError):
'''Raised when a build system is not configured properly.'''
[docs]class ContainerError(ReframeError):
'''Raised when a container platform is not configured properly.'''
[docs]class CommandLineError(ReframeError):
'''Raised when an error in command-line arguments occurs.'''
[docs]class BuildError(ReframeError):
'''Raised when a build fails.'''
def __init__(self, stdout, stderr, prefix=None):
super().__init__()
num_lines = 10
prefix = prefix or '.'
lines = [
f'stdout: {stdout!r}, stderr: {stderr!r}',
f'--- {stderr} (first {num_lines} lines) ---'
]
with contextlib.suppress(OSError):
with open(os.path.join(prefix, stderr)) as fp:
for i, line in enumerate(fp):
if i < num_lines:
# Remove trailing '\n'
lines.append(line[:-1])
lines += [f'--- {stderr} --- ']
self._message = '\n'.join(lines)
[docs]class SpawnedProcessError(ReframeError):
'''Raised when a spawned OS command has failed.'''
def __init__(self, args, stdout, stderr, exitcode):
super().__init__()
if isinstance(args, str):
self._command = args
else:
self._command = ' '.join(args)
self._stdout = stdout
self._stderr = stderr
self._exitcode = exitcode
# Format message
lines = [
f"command '{self.command}' failed with exit code {self.exitcode}:"
]
lines.append('--- stdout ---')
if stdout:
lines.append(stdout)
lines.append('--- stdout ---')
lines.append('--- stderr ---')
if stderr:
lines.append(stderr)
lines.append('--- stderr ---')
self._message = '\n'.join(lines)
@property
def command(self):
'''The command that the spawned process tried to execute.'''
return self._command
@property
def stdout(self):
'''The standard output of the process as a string.'''
return self._stdout
@property
def stderr(self):
'''The standard error of the process as a string.'''
return self._stderr
@property
def exitcode(self):
'''The exit code of the process.'''
return self._exitcode
[docs]class SpawnedProcessTimeout(SpawnedProcessError):
'''Raised when a spawned OS command has timed out.'''
def __init__(self, args, stdout, stderr, timeout):
super().__init__(args, stdout, stderr, None)
self._timeout = timeout
# Format message
lines = [f"command '{self.command}' timed out after {self.timeout}s:"]
lines.append('--- stdout ---')
if self._stdout:
lines.append(self._stdout)
lines.append('--- stdout ---')
lines.append('--- stderr ---')
if self._stderr:
lines.append(self._stderr)
lines.append('--- stderr ---')
self._message = '\n'.join(lines)
@property
def timeout(self):
'''The timeout of the process.'''
return self._timeout
[docs]class JobSchedulerError(ReframeError):
'''Raised when a job scheduler encounters an error condition.'''
[docs]class JobError(ReframeError):
'''Raised for job related errors.'''
def __init__(self, msg=None, jobid=None):
message = '[jobid=%s]' % jobid
if msg:
message += ' ' + msg
super().__init__(message)
self._jobid = jobid
@property
def jobid(self):
'''The job ID of the job that encountered the error.'''
return self._jobid
[docs]class JobBlockedError(JobError):
'''Raised by job schedulers when a job is blocked indefinitely.'''
[docs]class JobNotStartedError(JobError):
'''Raised when trying an operation on a unstarted job.'''
[docs]class DependencyError(ReframeError):
'''Raised when a dependency problem is encountered.'''
[docs]class SkipTestError(ReframeError):
'''Raised when a test needs to be skipped.'''
[docs]def user_frame(exc_type, exc_value, tb):
'''Return a user frame from the exception's traceback.
As user frame is considered the first frame that is outside from
:mod:`reframe` module.
:returns: A frame object or :class:`None` if no user frame was found.
'''
if not inspect.istraceback(tb):
return None
for finfo in reversed(inspect.getinnerframes(tb)):
relpath = os.path.relpath(finfo.filename, reframe.INSTALL_PREFIX)
if relpath.split(os.sep)[0] != 'reframe':
return finfo
return None
[docs]def is_exit_request(exc_type, exc_value, tb):
'''Check if the error is a request to exit.'''
return isinstance(exc_value, (KeyboardInterrupt,
ForceExitError,
FailureLimitError))
[docs]def is_user_error(exc_type, exc_value, tb):
'''Check if error is a user programming error.
A user error is any of :py:class:`AttributeError`, :py:class:`NameError`,
:py:class:`ModuleNotFoundError`, :py:class:`TypeError` or
:py:class:`ValueError` and the exception isthrown from user context.
'''
frame = user_frame(exc_type, exc_value, tb)
if frame is None:
return False
return isinstance(exc_value,
(AttributeError, ModuleNotFoundError, NameError,
TypeError, ValueError))
[docs]def is_severe(exc_type, exc_value, tb):
'''Check if exception is a severe one.'''
soft_errors = (ReframeError,
OSError,
KeyboardInterrupt,
TimeoutError)
if isinstance(exc_value, soft_errors):
return False
# User errors are treated as soft
return not is_user_error(exc_type, exc_value, tb)
[docs]def what(exc_type, exc_value, tb):
'''A short description of the error.'''
if exc_type is None:
return ''
reason = utility.decamelize(exc_type.__name__, ' ')
# We need frame information for user type and value errors
if isinstance(exc_value, KeyboardInterrupt):
reason = 'cancelled by user'
elif isinstance(exc_value, AbortTaskError):
reason = f'aborted due to {type(exc_value.__cause__).__name__}'
elif is_user_error(exc_type, exc_value, tb):
frame = user_frame(exc_type, exc_value, tb)
relpath = os.path.relpath(frame.filename)
source = ''.join(frame.code_context or '<n/a>')
reason += f': {relpath}:{frame.lineno}: {exc_value}\n{source}'
else:
if str(exc_value):
reason += f': {exc_value}'
return reason