xref: /dpdk/dts/framework/exception.py (revision d029f35384d0844e9aeb5dbc46fbe1b063d649f7) 
1# SPDX-License-Identifier: BSD-3-Clause
2# Copyright(c) 2010-2014 Intel Corporation
3# Copyright(c) 2022-2023 PANTHEON.tech s.r.o.
4# Copyright(c) 2022-2023 University of New Hampshire
5
6"""DTS exceptions.
7
8The exceptions all have different severities expressed as an integer.
9The highest severity of all raised exceptions is used as the exit code of DTS.
10"""
11
12from enum import IntEnum, unique
13from typing import ClassVar
14
15
16@unique
17class ErrorSeverity(IntEnum):
18    """The severity of errors that occur during DTS execution.
19
20    All exceptions are caught and the most severe error is used as return code.
21    """
22
23    #:
24    NO_ERR = 0
25    #:
26    GENERIC_ERR = 1
27    #:
28    CONFIG_ERR = 2
29    #:
30    REMOTE_CMD_EXEC_ERR = 3
31    #:
32    SSH_ERR = 4
33    #:
34    DPDK_BUILD_ERR = 10
35    #:
36    TESTCASE_VERIFY_ERR = 20
37    #:
38    BLOCKING_TESTSUITE_ERR = 25
39
40
41class DTSError(Exception):
42    """The base exception from which all DTS exceptions are subclassed.
43
44    Do not use this exception, only use subclassed exceptions.
45    """
46
47    #:
48    severity: ClassVar[ErrorSeverity] = ErrorSeverity.GENERIC_ERR
49
50
51class SSHTimeoutError(DTSError):
52    """The SSH execution of a command timed out."""
53
54    #:
55    severity: ClassVar[ErrorSeverity] = ErrorSeverity.SSH_ERR
56    _command: str
57
58    def __init__(self, command: str):
59        """Define the meaning of the first argument.
60
61        Args:
62            command: The executed command.
63        """
64        self._command = command
65
66    def __str__(self) -> str:
67        """Add some context to the string representation."""
68        return f"{self._command} execution timed out."
69
70
71class SSHConnectionError(DTSError):
72    """An unsuccessful SSH connection."""
73
74    #:
75    severity: ClassVar[ErrorSeverity] = ErrorSeverity.SSH_ERR
76    _host: str
77    _errors: list[str]
78
79    def __init__(self, host: str, errors: list[str] | None = None):
80        """Define the meaning of the first two arguments.
81
82        Args:
83            host: The hostname to which we're trying to connect.
84            errors: Any errors that occurred during the connection attempt.
85        """
86        self._host = host
87        self._errors = [] if errors is None else errors
88
89    def __str__(self) -> str:
90        """Include the errors in the string representation."""
91        message = f"Error trying to connect with {self._host}."
92        if self._errors:
93            message += f" Errors encountered while retrying: {', '.join(self._errors)}"
94
95        return message
96
97
98class SSHSessionDeadError(DTSError):
99    """The SSH session is no longer alive."""
100
101    #:
102    severity: ClassVar[ErrorSeverity] = ErrorSeverity.SSH_ERR
103    _host: str
104
105    def __init__(self, host: str):
106        """Define the meaning of the first argument.
107
108        Args:
109            host: The hostname of the disconnected node.
110        """
111        self._host = host
112
113    def __str__(self) -> str:
114        """Add some context to the string representation."""
115        return f"SSH session with {self._host} has died."
116
117
118class ConfigurationError(DTSError):
119    """An invalid configuration."""
120
121    #:
122    severity: ClassVar[ErrorSeverity] = ErrorSeverity.CONFIG_ERR
123
124
125class RemoteCommandExecutionError(DTSError):
126    """An unsuccessful execution of a remote command."""
127
128    #:
129    severity: ClassVar[ErrorSeverity] = ErrorSeverity.REMOTE_CMD_EXEC_ERR
130    #: The executed command.
131    command: str
132    _command_return_code: int
133
134    def __init__(self, command: str, command_return_code: int):
135        """Define the meaning of the first two arguments.
136
137        Args:
138            command: The executed command.
139            command_return_code: The return code of the executed command.
140        """
141        self.command = command
142        self._command_return_code = command_return_code
143
144    def __str__(self) -> str:
145        """Include both the command and return code in the string representation."""
146        return f"Command {self.command} returned a non-zero exit code: {self._command_return_code}"
147
148
149class RemoteDirectoryExistsError(DTSError):
150    """A directory that exists on a remote node."""
151
152    #:
153    severity: ClassVar[ErrorSeverity] = ErrorSeverity.REMOTE_CMD_EXEC_ERR
154
155
156class DPDKBuildError(DTSError):
157    """A DPDK build failure."""
158
159    #:
160    severity: ClassVar[ErrorSeverity] = ErrorSeverity.DPDK_BUILD_ERR
161
162
163class TestCaseVerifyError(DTSError):
164    """A test case failure."""
165
166    #:
167    severity: ClassVar[ErrorSeverity] = ErrorSeverity.TESTCASE_VERIFY_ERR
168
169
170class BlockingTestSuiteError(DTSError):
171    """A failure in a blocking test suite."""
172
173    #:
174    severity: ClassVar[ErrorSeverity] = ErrorSeverity.BLOCKING_TESTSUITE_ERR
175    _suite_name: str
176
177    def __init__(self, suite_name: str) -> None:
178        """Define the meaning of the first argument.
179
180        Args:
181            suite_name: The blocking test suite.
182        """
183        self._suite_name = suite_name
184
185    def __str__(self) -> str:
186        """Add some context to the string representation."""
187        return f"Blocking suite {self._suite_name} failed."
188