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""" 7User-defined exceptions used across the framework. 8""" 9 10from enum import IntEnum, unique 11from typing import ClassVar 12 13 14@unique 15class ErrorSeverity(IntEnum): 16 """ 17 The severity of errors that occur during DTS execution. 18 All exceptions are caught and the most severe error is used as return code. 19 """ 20 21 NO_ERR = 0 22 GENERIC_ERR = 1 23 CONFIG_ERR = 2 24 REMOTE_CMD_EXEC_ERR = 3 25 SSH_ERR = 4 26 DPDK_BUILD_ERR = 10 27 TESTCASE_VERIFY_ERR = 20 28 BLOCKING_TESTSUITE_ERR = 25 29 30 31class DTSError(Exception): 32 """ 33 The base exception from which all DTS exceptions are derived. 34 Stores error severity. 35 """ 36 37 severity: ClassVar[ErrorSeverity] = ErrorSeverity.GENERIC_ERR 38 39 40class SSHTimeoutError(DTSError): 41 """ 42 Command execution timeout. 43 """ 44 45 severity: ClassVar[ErrorSeverity] = ErrorSeverity.SSH_ERR 46 _command: str 47 48 def __init__(self, command: str): 49 self._command = command 50 51 def __str__(self) -> str: 52 return f"TIMEOUT on {self._command}" 53 54 55class SSHConnectionError(DTSError): 56 """ 57 SSH connection error. 58 """ 59 60 severity: ClassVar[ErrorSeverity] = ErrorSeverity.SSH_ERR 61 _host: str 62 _errors: list[str] 63 64 def __init__(self, host: str, errors: list[str] | None = None): 65 self._host = host 66 self._errors = [] if errors is None else errors 67 68 def __str__(self) -> str: 69 message = f"Error trying to connect with {self._host}." 70 if self._errors: 71 message += f" Errors encountered while retrying: {', '.join(self._errors)}" 72 73 return message 74 75 76class SSHSessionDeadError(DTSError): 77 """ 78 SSH session is not alive. 79 It can no longer be used. 80 """ 81 82 severity: ClassVar[ErrorSeverity] = ErrorSeverity.SSH_ERR 83 _host: str 84 85 def __init__(self, host: str): 86 self._host = host 87 88 def __str__(self) -> str: 89 return f"SSH session with {self._host} has died" 90 91 92class ConfigurationError(DTSError): 93 """ 94 Raised when an invalid configuration is encountered. 95 """ 96 97 severity: ClassVar[ErrorSeverity] = ErrorSeverity.CONFIG_ERR 98 99 100class RemoteCommandExecutionError(DTSError): 101 """ 102 Raised when a command executed on a Node returns a non-zero exit status. 103 """ 104 105 severity: ClassVar[ErrorSeverity] = ErrorSeverity.REMOTE_CMD_EXEC_ERR 106 command: str 107 _command_return_code: int 108 109 def __init__(self, command: str, command_return_code: int): 110 self.command = command 111 self._command_return_code = command_return_code 112 113 def __str__(self) -> str: 114 return f"Command {self.command} returned a non-zero exit code: {self._command_return_code}" 115 116 117class RemoteDirectoryExistsError(DTSError): 118 """ 119 Raised when a remote directory to be created already exists. 120 """ 121 122 severity: ClassVar[ErrorSeverity] = ErrorSeverity.REMOTE_CMD_EXEC_ERR 123 124 125class DPDKBuildError(DTSError): 126 """ 127 Raised when DPDK build fails for any reason. 128 """ 129 130 severity: ClassVar[ErrorSeverity] = ErrorSeverity.DPDK_BUILD_ERR 131 132 133class TestCaseVerifyError(DTSError): 134 """ 135 Used in test cases to verify the expected behavior. 136 """ 137 138 severity: ClassVar[ErrorSeverity] = ErrorSeverity.TESTCASE_VERIFY_ERR 139 140 141class BlockingTestSuiteError(DTSError): 142 severity: ClassVar[ErrorSeverity] = ErrorSeverity.BLOCKING_TESTSUITE_ERR 143 _suite_name: str 144 145 def __init__(self, suite_name: str) -> None: 146 self._suite_name = suite_name 147 148 def __str__(self) -> str: 149 return f"Blocking suite {self._suite_name} failed." 150