diff --git a/nodescraper/base/oobanddataplugin.py b/nodescraper/base/oobanddataplugin.py index da1a0e46..a17283ee 100644 --- a/nodescraper/base/oobanddataplugin.py +++ b/nodescraper/base/oobanddataplugin.py @@ -67,7 +67,7 @@ def run( # type: ignore[override] system_interaction_level: Union[ SystemInteractionLevel, str ] = SystemInteractionLevel.INTERACTIVE, - preserve_connection: bool = False, + preserve_connection: bool = True, data: Optional[Any] = None, collection_args: Optional[Any] = None, analysis_args: Optional[Any] = None, diff --git a/nodescraper/connection/inband/inbandlocal.py b/nodescraper/connection/inband/inbandlocal.py index 5fa4cf63..f017b81a 100644 --- a/nodescraper/connection/inband/inbandlocal.py +++ b/nodescraper/connection/inband/inbandlocal.py @@ -1,4 +1,4 @@ -############################################################################### +############################################################################## # # MIT License # diff --git a/nodescraper/connection/inband/inbandmanager.py b/nodescraper/connection/inband/inbandmanager.py index ab924fe6..04aead9a 100644 --- a/nodescraper/connection/inband/inbandmanager.py +++ b/nodescraper/connection/inband/inbandmanager.py @@ -37,7 +37,6 @@ from nodescraper.interfaces.connectionmanager import ConnectionManager from nodescraper.interfaces.taskresulthook import TaskResultHook from nodescraper.models import SystemInfo, TaskResult -from nodescraper.utils import get_exception_traceback from .inband import InBandConnection from .inbandlocal import LocalShell @@ -92,47 +91,46 @@ def connect( self.connection = LocalShell() return self.result - if not self.connection_args or not isinstance(self.connection_args, SSHConnectionParams): - if not self.connection_args: - message = "No SSH credentials provided" - else: - message = "Invalide SSH creddentials provided" + if not self.connection_args: + return self._fail_connect( + EventCategory.RUNTIME, + "Cannot connect to the system: no SSH credentials were provided.", + "No SSH connection arguments provided to InBandConnectionManager", + {"connection_args": None}, + ) - self._log_event( - category=EventCategory.RUNTIME, - description=message, - priority=EventPriority.CRITICAL, - console_log=True, + if not isinstance(self.connection_args, SSHConnectionParams): + return self._fail_connect( + EventCategory.RUNTIME, + "Cannot connect to the system: the SSH credentials are not valid.", + "SSH connection arguments are not SSHConnectionParams " + f"(got {type(self.connection_args).__name__})", + {"connection_args_type": type(self.connection_args).__name__}, ) - self.result.status = ExecutionStatus.EXECUTION_FAILURE - return self.result + hostname = self.connection_args.hostname + self.logger.info("Initializing SSH connection to system '%s'", hostname) + shell = RemoteShell(self.connection_args) try: - self.logger.info( - "Initializing SSH connection to system '%s'", - self.connection_args.hostname, - ) - self.connection = RemoteShell(self.connection_args) - self.connection.connect_ssh() + shell.connect_ssh() except SSHConnectionError as exception: - self._log_event( - category=EventCategory.SSH, - description=f"{str(exception)}", - priority=EventPriority.CRITICAL, - console_log=True, + shell.client.close() + port = self.connection_args.port + return self._fail_connect( + EventCategory.SSH, + "Could not connect to the system over SSH.", + f"SSH connection to '{hostname}:{port}' failed: {exception}", + { + "hostname": str(hostname), + "port": port, + "username": self.connection_args.username, + "exception": str(exception), + }, ) - self.connection = None # Nullify the connection since its not usable - self.result.status = ExecutionStatus.EXECUTION_FAILURE - except Exception as exception: - self._log_event( - category=EventCategory.SSH, - description=f"Exception during SSH: {str(exception)}", - data=get_exception_traceback(exception), - priority=EventPriority.CRITICAL, - console_log=True, - ) - self.connection = None # Nullify the connection since its not usable - self.result.status = ExecutionStatus.EXECUTION_FAILURE + + self.connection = shell + self.result.message = f"SSH connection to '{hostname}' established successfully" + self.result.status = ExecutionStatus.OK return self.result def disconnect(self): diff --git a/nodescraper/connection/inband/inbandremote.py b/nodescraper/connection/inband/inbandremote.py index c79d5f03..eebc728d 100644 --- a/nodescraper/connection/inband/inbandremote.py +++ b/nodescraper/connection/inband/inbandremote.py @@ -23,6 +23,7 @@ # SOFTWARE. # ############################################################################### +import logging import os import socket from typing import Type, Union @@ -98,8 +99,10 @@ def connect_ssh(self): raise SSHConnectionError(f"Unable to establish SSH connection: {str(e)}") from e except EOFError as e: raise SSHConnectionError("EOFError during SSH connection") from e + except OSError as e: + raise SSHConnectionError(f"SSH network error: {e}") from e except Exception as e: - raise e + raise SSHConnectionError(f"Unexpected error during SSH connection: {str(e)}") from e def read_file( self, @@ -176,6 +179,11 @@ def run_command( stderr_str = str(exc) or "SSH channel failed" stdout_str = "" exit_code = 124 + except Exception as exc: + logging.getLogger(__name__).exception("Unexpected error running SSH command") + stderr_str = f"{type(exc).__name__}: {exc}" + stdout_str = "" + exit_code = 124 return CommandArtifact( command=cmd_str, diff --git a/nodescraper/connection/oob_ssh/oob_ssh_connection_manager.py b/nodescraper/connection/oob_ssh/oob_ssh_connection_manager.py index 3d07d1bb..cbd5d737 100644 --- a/nodescraper/connection/oob_ssh/oob_ssh_connection_manager.py +++ b/nodescraper/connection/oob_ssh/oob_ssh_connection_manager.py @@ -32,7 +32,6 @@ from nodescraper.interfaces.connectionmanager import ConnectionManager from nodescraper.interfaces.taskresulthook import TaskResultHook from nodescraper.models import SystemInfo, TaskResult -from nodescraper.utils import get_exception_traceback from ..inband.inband import InBandConnection from ..inband.inbandremote import RemoteShell, SSHConnectionError @@ -66,62 +65,83 @@ def __init__( def connect(self) -> TaskResult: if not self.connection_args: - self._log_event( - category=EventCategory.RUNTIME, - description="No Redfish connection parameters provided for OOB SSH", - priority=EventPriority.CRITICAL, - console_log=True, + return self._fail_connect( + EventCategory.RUNTIME, + "Cannot connect to the BMC over SSH: no connection details were provided.", + "No Redfish connection parameters provided for OOB SSH", + {"connection_args": None}, ) - self.result.status = ExecutionStatus.EXECUTION_FAILURE - return self.result raw = self.connection_args if isinstance(raw, dict): - params = RedfishConnectionParams.model_validate(raw) + try: + params = RedfishConnectionParams.model_validate(raw) + except ValueError as exception: + return self._fail_connect( + EventCategory.RUNTIME, + "Cannot connect to the BMC over SSH: the connection details are not valid.", + f"Invalid Redfish connection parameters for OOB SSH: {exception}", + {"exception": str(exception)}, + ) elif isinstance(raw, RedfishConnectionParams): params = raw else: - self._log_event( - category=EventCategory.RUNTIME, - description="Redfish connection_args must be dict or RedfishConnectionParams", - priority=EventPriority.CRITICAL, - console_log=True, + return self._fail_connect( + EventCategory.RUNTIME, + "Cannot connect to the BMC over SSH: the connection details are not valid.", + "Redfish connection_args must be dict or RedfishConnectionParams " + f"(got {type(raw).__name__})", + {"connection_args_type": type(raw).__name__}, ) - self.result.status = ExecutionStatus.EXECUTION_FAILURE - return self.result if params.is_multi_target and params.host is None: - self.result.status = ExecutionStatus.NOT_RAN - self.result.message = ( + message = ( "OOB SSH plugins use a single BMC host. " "A targets list applies only to Redfish plugins." ) + self._log_event( + category=EventCategory.RUNTIME, + description=message, + priority=EventPriority.WARNING, + console_log=True, + ) + self.result.status = ExecutionStatus.NOT_RAN + self.result.message = message return self.result try: ssh_params = redfish_params_to_ssh(params) - self.logger.info("Initializing OOB SSH to BMC host %s", ssh_params.hostname) - self.connection = RemoteShell(ssh_params) - self.connection.connect_ssh() - except SSHConnectionError as exception: - self._log_event( - category=EventCategory.SSH, - description=str(exception), - priority=EventPriority.CRITICAL, - console_log=True, + except ValueError as exception: + return self._fail_connect( + EventCategory.RUNTIME, + "Cannot connect to the BMC over SSH: the connection details are not valid.", + f"Could not build OOB SSH parameters from the Redfish parameters: {exception}", + { + "host": str(params.host), + "exception": str(exception), + }, ) - self.result.status = ExecutionStatus.EXECUTION_FAILURE - self.connection = None - except Exception as exception: - self._log_event( - category=EventCategory.SSH, - description=f"Exception during OOB SSH: {exception!s}", - data=get_exception_traceback(exception), - priority=EventPriority.CRITICAL, - console_log=True, + + hostname = ssh_params.hostname + self.logger.info("Initializing OOB SSH to BMC host %s", hostname) + shell = RemoteShell(ssh_params) + try: + shell.connect_ssh() + except SSHConnectionError as exception: + shell.client.close() + return self._fail_connect( + EventCategory.SSH, + "Could not connect to the BMC over SSH.", + f"OOB SSH connection to BMC '{hostname}:{ssh_params.port}' failed: {exception}", + { + "hostname": str(hostname), + "port": ssh_params.port, + "username": ssh_params.username, + "exception": str(exception), + }, ) - self.result.status = ExecutionStatus.EXECUTION_FAILURE - self.connection = None + + self.connection = shell return self.result def disconnect(self) -> None: diff --git a/nodescraper/connection/redfish/redfish_connection.py b/nodescraper/connection/redfish/redfish_connection.py index 8999534e..1cfc1667 100644 --- a/nodescraper/connection/redfish/redfish_connection.py +++ b/nodescraper/connection/redfish/redfish_connection.py @@ -188,10 +188,7 @@ def _transport_error_message(self, exc: Exception) -> str: def _execute_request(self, request_fn: Callable[[], _T]) -> _T: try: return request_fn() - except ( - requests.exceptions.Timeout, - requests.exceptions.ConnectionError, - ) as exc: + except requests.exceptions.RequestException as exc: raise RedfishConnectionError(self._transport_error_message(exc)) from exc except socket.gaierror as exc: raise RedfishConnectionError(self._transport_error_message(exc)) from exc @@ -251,7 +248,12 @@ def get(self, path: RedfishPath) -> dict[str, Any]: f"GET {path_str} failed: {resp.status_code} {resp.reason}", response=resp, ) - return resp.json() + try: + return resp.json() + except ValueError as exc: + raise RedfishConnectionError( + f"GET {path_str} returned a body that is not valid JSON", response=resp + ) from exc def get_response(self, path: Union[str, "RedfishPath"]) -> RedfishHttpResponse: """GET a Redfish path and return the raw Response. path may be a string or RedfishPath.""" diff --git a/nodescraper/connection/redfish/redfish_manager.py b/nodescraper/connection/redfish/redfish_manager.py index 3ecb88bb..0b43dee7 100644 --- a/nodescraper/connection/redfish/redfish_manager.py +++ b/nodescraper/connection/redfish/redfish_manager.py @@ -122,67 +122,86 @@ def __init__( def connect(self) -> TaskResult: """Connect to Redfish (single-target or multi-target).""" if not self.connection_args: - self._log_event( - category=EventCategory.RUNTIME, - description="No Redfish connection parameters provided", - priority=EventPriority.CRITICAL, - console_log=True, + return self._fail_connect( + EventCategory.RUNTIME, + "Cannot connect to Redfish: no connection details were provided.", + "No Redfish connection parameters provided", + {"connection_args": None}, ) - self.result.status = ExecutionStatus.EXECUTION_FAILURE - return self.result raw = self.connection_args if isinstance(raw, dict): - params = RedfishConnectionParams.model_validate(raw) + try: + params = RedfishConnectionParams.model_validate(raw) + except ValueError as exc: + return self._fail_connect( + EventCategory.RUNTIME, + "Cannot connect to Redfish: the connection details are not valid.", + f"Invalid Redfish connection parameters: {exc}", + {"exception": str(exc)}, + ) elif isinstance(raw, RedfishConnectionParams): params = raw else: - self._log_event( - category=EventCategory.RUNTIME, - description="Redfish connection_args must be dict or RedfishConnectionParams", - priority=EventPriority.CRITICAL, - console_log=True, + return self._fail_connect( + EventCategory.RUNTIME, + "Cannot connect to Redfish: the connection details are not valid.", + "Redfish connection_args must be dict or RedfishConnectionParams " + f"(got {type(raw).__name__})", + {"connection_args_type": type(raw).__name__}, ) - self.result.status = ExecutionStatus.EXECUTION_FAILURE - return self.result if params.is_multi_target: - self.target_connections: dict[str, RedfishConnection] = {} - failed_targets: dict[str, str] = {} - for target in params.targets: # type: ignore[union-attr] - key, conn, error = self._connect_target(target) - if conn is None: - failed_targets[key] = error or f"Redfish connection failed for target {key!r}" - continue - if key in self.target_connections: - self._log_event( - category=EventCategory.RUNTIME, - description=f"Duplicate Redfish target key {key!r}; keeping the first connection", - priority=EventPriority.WARNING, - console_log=True, - ) - conn.close() - continue - self.target_connections[key] = conn - if not self.target_connections: - self.result.status = ExecutionStatus.EXECUTION_FAILURE - lines = "\n".join(f"[{k}] {v}" for k, v in failed_targets.items()) - self.result.message = f"Redfish connection failed for every target\n{lines}" - self.result.events.clear() - return self.result - if failed_targets: - lines = "\n".join(f"[{k}] {v}" for k, v in failed_targets.items()) - self.result.status = ExecutionStatus.WARNING - self.result.message = f"Some Redfish targets failed to connect\n{lines}" - self.result.events.clear() - self.connection = MultiTargetRedfishConnection( # type: ignore[assignment] - self.target_connections, - max_workers=params.max_workers, - failed_targets=failed_targets, + return self._connect_multi(params) + return self._connect_single(params) + + def _connect_multi(self, params: RedfishConnectionParams) -> TaskResult: + """Connect each Redfish target. One failure does not drop the others. + + Args: + params: Parameters whose targets list is set. + + Returns: + TaskResult: Warning when some targets connect, execution failure when none do. + """ + self.target_connections: dict[str, RedfishConnection] = {} + failed_targets: dict[str, str] = {} + for target in params.targets or []: + key, conn, error = self._connect_target(target) + if conn is None: + failed_targets[key] = error or f"Redfish connection failed for target {key!r}" + continue + if key in self.target_connections: + self._log_event( + category=EventCategory.RUNTIME, + description=f"Duplicate Redfish target key {key!r}; keeping the first connection", + priority=EventPriority.WARNING, + console_log=True, + ) + conn.close() + continue + self.target_connections[key] = conn + + lines = "\n".join(f"[{k}] {v}" for k, v in failed_targets.items()) + if not self.target_connections: + summary = "Could not connect to any of the Redfish targets." + self._fail_connect( + EventCategory.RUNTIME, + summary, + f"Redfish connection failed for every target ({len(failed_targets)} tried)\n{lines}", + {"failed_targets": dict(failed_targets)}, ) return self.result - return self._connect_single(params) + if failed_targets: + self.result.status = ExecutionStatus.WARNING + self.result.message = f"Some Redfish targets failed to connect\n{lines}" + self.connection = MultiTargetRedfishConnection( # type: ignore[assignment] + self.target_connections, + max_workers=params.max_workers, + failed_targets=failed_targets, + ) + return self.result def _connect_target( self, target: RedfishConnectionParams @@ -198,8 +217,9 @@ def _connect_target( key = target.target_key or str(target.host) password = target.password.get_secret_value() if target.password else None base_url = _build_base_url(str(target.host), target.port, target.use_https) + self.logger.info("Connecting to Redfish at %s (target=%r)", base_url, key) + conn: Optional[RedfishConnection] = None try: - self.logger.info("Connecting to Redfish at %s (target=%r)", base_url, key) conn = RedfishConnection( base_url=base_url, username=target.username or "", @@ -211,8 +231,9 @@ def _connect_target( ) conn._ensure_session() conn.get_service_root() - return key, conn, None - except (RedfishConnectionError, Exception) as exc: # noqa: BLE001 + except Exception as exc: # noqa: BLE001 - one bad target must not drop the others + if conn is not None: + conn.close() description = f"Redfish connection failed for target {key!r}: {exc}" self._log_event( category=EventCategory.RUNTIME, @@ -221,42 +242,43 @@ def _connect_target( console_log=True, ) return key, None, description + return key, conn, None def _connect_single(self, params: RedfishConnectionParams) -> TaskResult: - """Connect in single-target mode (original connect logic).""" + """Connect in single-target mode.""" password = params.password.get_secret_value() if params.password else None base_url = _build_base_url(str(params.host), params.port, params.use_https) + self.logger.info("Connecting to Redfish at %s", base_url) + conn = RedfishConnection( + base_url=base_url, + username=params.username or "", + password=password, + timeout=params.timeout_seconds, + use_session_auth=params.use_session_auth, + verify_ssl=params.verify_ssl, + api_root=params.api_root, + ) try: - self.logger.info("Connecting to Redfish at %s", base_url) - self.connection = RedfishConnection( - base_url=base_url, - username=params.username or "", - password=password, - timeout=params.timeout_seconds, - use_session_auth=params.use_session_auth, - verify_ssl=params.verify_ssl, - api_root=params.api_root, - ) - self.connection._ensure_session() - self.connection.get_service_root() + conn._ensure_session() + conn.get_service_root() except RedfishConnectionError as exc: - self._log_event( - category=EventCategory.RUNTIME, - description=str(exc), - priority=EventPriority.CRITICAL, - console_log=True, - ) - self.result.status = ExecutionStatus.EXECUTION_FAILURE - self.connection = None - except Exception as exc: - self._log_event( - category=EventCategory.RUNTIME, - description=f"Redfish connection failed: {exc}", - priority=EventPriority.CRITICAL, - console_log=True, + conn.close() + return self._fail_connect( + EventCategory.RUNTIME, + "Could not connect to Redfish.", + f"Redfish connection to {base_url} failed: {exc}", + { + "base_url": base_url, + "host": str(params.host), + "port": params.port, + "use_https": params.use_https, + "verify_ssl": params.verify_ssl, + "timeout_seconds": params.timeout_seconds, + "exception": str(exc), + }, ) - self.result.status = ExecutionStatus.EXECUTION_FAILURE - self.connection = None + + self.connection = conn return self.result def disconnect(self) -> None: diff --git a/nodescraper/connection/redfish/ssh_proxy_manager.py b/nodescraper/connection/redfish/ssh_proxy_manager.py index c13c0393..334dfa2c 100644 --- a/nodescraper/connection/redfish/ssh_proxy_manager.py +++ b/nodescraper/connection/redfish/ssh_proxy_manager.py @@ -84,36 +84,41 @@ def connect(self) -> TaskResult: return self._connect_single(params) def _validated_params(self) -> Optional[RedfishSshProxyConnectionParams]: - """Return connection params or record a failure when they are missing. + """Return connection params or record a failure when they are missing or invalid. Returns: Optional[RedfishSshProxyConnectionParams]: Parsed params, or None when invalid. """ if not self.connection_args: - self._log_event( - category=EventCategory.RUNTIME, - description="No Redfish SSH-proxy connection parameters provided", - priority=EventPriority.CRITICAL, - console_log=True, + self._fail_connect( + EventCategory.RUNTIME, + "Cannot connect through the BMC: no connection details were provided.", + "No Redfish SSH-proxy connection parameters provided", + {"connection_args": None}, ) - self.result.status = ExecutionStatus.EXECUTION_FAILURE return None raw = self.connection_args if isinstance(raw, dict): - return RedfishSshProxyConnectionParams.model_validate(raw) + try: + return RedfishSshProxyConnectionParams.model_validate(raw) + except Exception as exc: + self._fail_connect( + EventCategory.RUNTIME, + "Cannot connect through the BMC: the connection details are not valid.", + f"Invalid Redfish SSH-proxy connection parameters: {exc}", + {"exception": str(exc)}, + ) + return None if isinstance(raw, RedfishSshProxyConnectionParams): return raw - self._log_event( - category=EventCategory.RUNTIME, - description=( - "Redfish SSH-proxy connection_args must be dict or " - "RedfishSshProxyConnectionParams" - ), - priority=EventPriority.CRITICAL, - console_log=True, + self._fail_connect( + EventCategory.RUNTIME, + "Cannot connect through the BMC: the connection details are not valid.", + "Redfish SSH-proxy connection_args must be dict or RedfishSshProxyConnectionParams " + f"(got {type(raw).__name__})", + {"connection_args_type": type(raw).__name__}, ) - self.result.status = ExecutionStatus.EXECUTION_FAILURE return None def _connect_single(self, params: RedfishSshProxyConnectionParams) -> TaskResult: @@ -125,36 +130,45 @@ def _connect_single(self, params: RedfishSshProxyConnectionParams) -> TaskResult Returns: TaskResult: Connection result. """ + debug_data = { + "ssh_host": str(params.ssh.hostname) if params.ssh else None, + "ssh_port": params.ssh.port if params.ssh else None, + "ssh_username": params.ssh.username if params.ssh else None, + "redfish_host": str(params.host), + "redfish_port": params.port, + "use_https": params.use_https, + "timeout_seconds": params.timeout_seconds, + } try: self.connection = self._open_proxy(params) except SSHConnectionError as exc: - self._log_event( - category=EventCategory.SSH, - description=str(exc), - priority=EventPriority.CRITICAL, - console_log=True, + return self._fail_connect( + EventCategory.SSH, + "Could not connect to the BMC over SSH.", + f"SSH connection to BMC '{debug_data['ssh_host']}' failed: {exc}", + {**debug_data, "exception": str(exc)}, ) - self.result.status = ExecutionStatus.EXECUTION_FAILURE - self.connection = None except RedfishConnectionError as exc: - self._log_event( - category=EventCategory.RUNTIME, - description=str(exc), - priority=EventPriority.CRITICAL, - console_log=True, + return self._fail_connect( + EventCategory.RUNTIME, + "Connected to the BMC, but could not reach Redfish through it.", + f"Redfish check through BMC SSH proxy failed: {exc}", + {**debug_data, "exception": str(exc)}, + ) + except ValueError as exc: + return self._fail_connect( + EventCategory.RUNTIME, + "Cannot connect through the BMC: the connection details are not valid.", + f"Invalid Redfish SSH-proxy connection parameters: {exc}", + {**debug_data, "exception": str(exc)}, ) - self.result.status = ExecutionStatus.EXECUTION_FAILURE - self.connection = None except Exception as exc: - self._log_event( - category=EventCategory.RUNTIME, - description=f"Redfish SSH-proxy connection failed: {exc}", - data=get_exception_traceback(exc), - priority=EventPriority.CRITICAL, - console_log=True, + return self._fail_connect( + EventCategory.RUNTIME, + "Connecting through the BMC failed unexpectedly.", + f"Unexpected error during SSH connection: {exc}", + {**debug_data, "exception": str(exc), **get_exception_traceback(exc)}, ) - self.result.status = ExecutionStatus.EXECUTION_FAILURE - self.connection = None return self.result def _connect_multi(self, params: RedfishSshProxyConnectionParams) -> TaskResult: @@ -185,19 +199,21 @@ def _connect_multi(self, params: RedfishSshProxyConnectionParams) -> TaskResult: self._close_proxy(conn) continue self.target_connections[key] = conn + + lines = "\n".join(f"[{k}] {v}" for k, v in failed_targets.items()) if not self.target_connections: - self.result.status = ExecutionStatus.EXECUTION_FAILURE - lines = "\n".join(f"[{k}] {v}" for k, v in failed_targets.items()) - self.result.message = f"SSH-proxy Redfish connection failed for every target\n{lines}" - self.result.events.clear() + summary = "Could not connect to any of the BMC targets." + self._fail_connect( + EventCategory.RUNTIME, + summary, + f"SSH-proxy Redfish connection failed for every target ({len(failed_targets)} tried)\n{lines}", + {"failed_targets": dict(failed_targets)}, + ) return self.result + if failed_targets: - lines = "\n".join(f"[{k}] {v}" for k, v in failed_targets.items()) self.result.status = ExecutionStatus.WARNING self.result.message = f"Some SSH-proxy Redfish targets failed to connect\n{lines}" - # Clear events so TaskResult.finalize() does not re-append them as - # "(N warnings: ...)" on the same line — the info is already in result.message. - self.result.events.clear() self.connection = MultiTargetRedfishConnection( # type: ignore[assignment] self.target_connections, max_workers=params.max_workers, @@ -219,10 +235,17 @@ def _connect_target( key = _proxy_target_key(target) try: return key, self._open_proxy(target), None - except Exception as exc: # noqa: BLE001 - description = f"SSH-proxy Redfish connection failed for target {key!r}: {exc}" + except Exception as exc: + if isinstance(exc, (SSHConnectionError, RedfishConnectionError)): + description = f"SSH-proxy Redfish connection failed for target {key!r}: {exc}" + else: + description = f"Unknown SSH-proxy Redfish error for target {key!r}: {exc}" self._log_event( - category=EventCategory.RUNTIME, + category=( + EventCategory.SSH + if isinstance(exc, SSHConnectionError) + else EventCategory.RUNTIME + ), description=description, priority=EventPriority.WARNING, console_log=True, diff --git a/nodescraper/interfaces/connectionmanager.py b/nodescraper/interfaces/connectionmanager.py index 16deee89..60937f8d 100644 --- a/nodescraper/interfaces/connectionmanager.py +++ b/nodescraper/interfaces/connectionmanager.py @@ -67,8 +67,25 @@ def wrapper( "Exception connecting with %s", connection_manager.__class__.__name__ ) connection_manager.result.status = ExecutionStatus.EXECUTION_FAILURE + connection_manager.result.message = f"Connection failed: {str(exception)}" result = connection_manager.result + connection_manager.result = result + + if result.status == ExecutionStatus.UNSET and connection_manager.connection is None: + # Force a fail if conncetion manager failed to give us a connection + result.status = ExecutionStatus.EXECUTION_FAILURE + + if result.status >= ExecutionStatus.ERROR and not result.events: + result.message = result.message or "Connection could not be established" + # Push an event if the connection manager didn't + connection_manager._log_event( + category=EventCategory.RUNTIME, + description=result.message, + priority=EventPriority.CRITICAL, + console_log=True, + ) + result.finalize() connection_manager._run_hooks(result) @@ -163,7 +180,51 @@ def __exit__( def connect(self) -> TaskResult: """initialize connection""" + def _fail_connect( + self, + category: EventCategory, + message: str, + event_description: str | None = None, + event_data: Optional[dict[Any, Any]] = None, + ) -> TaskResult: + """Record a failed connection attempt: one CRITICAL event, nullifying the connection, and marking the result as failure. + + Call this for every expected failure inside connect() and return its result. Exceptions + that connect() does not expect are not caught there; connect_decorator reports them. + + Args: + category (EventCategory): Event category describing where the failure happened. + message (str): Short, plain-language summary for the end user (becomes the task result message), + e.g. "Could not connect to the BMC over SSH". No exception text or jargon. + event_description (str): Event description with a little more debugging detail than the message, e.g. the + host and the underlying error string. When None or empty the message is used. + event_data (Optional[dict], optional): Full debug information for the event, e.g. host, port, exception type, + exception string and traceback. Never include secrets. This data must be JSON-serializable. + Defaults to None. + + Returns: + TaskResult: The failed connection result. + """ + self._log_event( + category=category, + description=event_description or message, + data=event_data, + priority=EventPriority.CRITICAL, + console_log=True, + ) + self.connection = None + self.result.message = message + self.result.status = ExecutionStatus.EXECUTION_FAILURE + return self.result + + def safe_disconnect(self) -> None: + """Disconnect without raising; a failure is logged.""" + try: + self.disconnect() + except Exception: + self.logger.exception("Failed to disconnect %s", self.__class__.__name__) + def disconnect(self): - """disconnect connection (Optional)""" + """disconnect connection (Optional). The result of the connection attempt is kept.""" + self.logger.info(f"Connection {self.__class__.__name__} disconnected.") self.connection = None - self.result.status = ExecutionStatus.UNSET diff --git a/nodescraper/interfaces/dataplugin.py b/nodescraper/interfaces/dataplugin.py index 6604d3fe..44f9d3dc 100644 --- a/nodescraper/interfaces/dataplugin.py +++ b/nodescraper/interfaces/dataplugin.py @@ -46,6 +46,7 @@ SystemInfo, TaskResult, ) +from nodescraper.taskresulthooks.filesystemloghook import hooks_for_fixed_directory from nodescraper.utils import resolve_log_dir_name from .connectionmanager import TConnectArg, TConnectionManager @@ -155,6 +156,23 @@ def get_collector_classes(cls) -> tuple[Type[DataCollector], ...]: return tuple(collector) return (collector,) + def _connection_manager_hooks(self) -> list[TaskResultHook]: + """Hooks for a connection manager created by this plugin. + + Its result is saved in ``///connection`` so it is kept + apart from the collector's own result and events files. + """ + if not self.log_path: + return self.task_result_hooks + parts = [resolve_log_dir_name(self.__class__.__name__)] + collectors = self.get_collector_classes() + if collectors: + parts.append(resolve_log_dir_name(collectors[0].__name__)) + parts.append("connection") + return hooks_for_fixed_directory( + self.task_result_hooks, str(Path(self.log_path).joinpath(*parts)) + ) + @classmethod def _collector_args_class( cls, collector_cls: Type[DataCollector] @@ -318,15 +336,22 @@ def collect( system_interaction_level: Optional[ Union[SystemInteractionLevel, str] ] = SystemInteractionLevel.INTERACTIVE, - preserve_connection: bool = False, + preserve_connection: bool = True, collection_args: Optional[TCollectArg] = None, ) -> TaskResult: """Run data collector task + A connection manager created by this plugin is always closed (and its result saved) when + collection finishes. A connection manager passed in by the caller is left open by default + so the caller, who owns it, can reuse it across plugins and close it when done; pass + ``preserve_connection=False`` to have the plugin close it after this collection. + Args: max_event_priority_level (Union[EventPriority, str], optional): priority limit for events. Defaults to EventPriority.CRITICAL. system_interaction_level (Union[SystemInteractionLevel, str], optional): system interaction level. Defaults to SystemInteractionLevel.INTERACTIVE. - preserve_connection (bool, optional): whether we should close the connection after data collection. Defaults to False. + preserve_connection (bool, optional): keep a connection manager passed in by the caller open after + collection. Set False to close it. Has no effect on a manager the plugin created itself, + which is always closed. Defaults to True. collection_args (Optional[TCollectArg], optional): args for data collection (validated model). Defaults to None. Returns: @@ -346,27 +371,37 @@ def collect( try: if not self.connection_manager: if not self.CONNECTION_TYPE: - self.collection_result = TaskResult( + return TaskResult( task=primary_collector.__name__, parent=self.__class__.__name__, status=ExecutionStatus.NOT_RAN, message=f"No connection manager type provided for {self.__class__.__name__}", ) - return self.collection_result self.logger.info("No connection manager provide, initializing connection manager") + self._owns_connection_manager = True self.connection_manager = self.CONNECTION_TYPE( system_info=self.system_info.model_copy(), logger=self.logger, parent=self.__class__.__name__, - task_result_hooks=self.task_result_hooks, + task_result_hooks=self._connection_manager_hooks(), event_reporter=self.event_reporter, session_id=self.session_id, ) + # Connect if never attempted or previously connected then disconnected; + # If the connection instance has previously been known to be in a failed Status + # then it should not attempt to reconnect automatically. if ( not self.connection_manager.connection - and self.connection_manager.result.status == ExecutionStatus.UNSET + and self.connection_manager.result.status + in ( + ExecutionStatus.UNSET, + ExecutionStatus.OK, + ExecutionStatus.WARNING, + ) ): + # Set the parent to us + self.connection_manager.parent = self.__class__.__name__ self.connection_manager.connect() # Proceed as long as a connection was established. @@ -388,6 +423,7 @@ def collect( collector_args = self._resolve_collector_args(collector_cls, collection_args) collector_log_path = None if self.log_path: + # Place the log path of the connection in the collector's directory _log_path = ( Path(self.log_path) / resolve_log_dir_name(self.__class__.__name__) @@ -436,8 +472,10 @@ def collect( message=f"Unhandled exception running data collector: {str(e)}", ) finally: - if not preserve_connection and self.connection_manager: - self.connection_manager.disconnect() + if self.connection_manager and ( + self._owns_connection_manager or not preserve_connection + ): + self.connection_manager.safe_disconnect() return self.collection_result @@ -498,6 +536,15 @@ def analyze( self.analysis_result = analyzer_task.analyze_data(self.data, analysis_args) return self.analysis_result + def _get_connection_result(self) -> Optional[TaskResult]: + """Snapshot of the connection result, or None if no connection was initiated.""" + if self.connection_manager is None: + return None + result = self.connection_manager.result + if result is None or result.status == ExecutionStatus.UNSET: + return None + return result.model_copy(deep=True) + def run( self, collection: Annotated[ @@ -513,7 +560,7 @@ def run( Union[SystemInteractionLevel, str], "System interaction level (e.g. PASSIVE, INTERACTIVE, DISRUPTIVE).", ] = SystemInteractionLevel.INTERACTIVE, - preserve_connection: bool = False, + preserve_connection: bool = True, data: Annotated[ Optional[Union[str, dict, TDataModel]], Field( @@ -533,7 +580,8 @@ def run( analysis (bool, optional): Enable data analysis. Defaults to True. max_event_priority_level (Union[EventPriority, str], optional): Max priority level to assign to events. Defaults to EventPriority.CRITICAL. system_interaction_level (Union[SystemInteractionLevel, str], optional): System interaction level. Defaults to SystemInteractionLevel.INTERACTIVE. - preserve_connection (bool, optional): Whether to close the connection when data collection is complete. Defaults to False. + preserve_connection (bool, optional): Keep a connection manager passed in by the caller open after collection. + Set False to close it. A manager the plugin created itself is always closed. Defaults to True. data (Optional[Union[str, dict, TDataModel]], optional): Input data. Defaults to None. collection_args (Optional[Union[TCollectArg , dict]], optional): Arguments for data collection. Defaults to None. analysis_args (Optional[Union[TAnalyzeArg , dict]], optional): Arguments for data analysis. Defaults to None. @@ -546,8 +594,8 @@ def run( self.collect( max_event_priority_level=max_event_priority_level, system_interaction_level=system_interaction_level, - collection_args=collection_args, preserve_connection=preserve_connection, + collection_args=collection_args, ) if analysis: @@ -591,6 +639,7 @@ def run( system_data=self.data, collection_result=self.collection_result, analysis_result=self.analysis_result, + connection_result=self._get_connection_result(), ), ) diff --git a/nodescraper/interfaces/plugin.py b/nodescraper/interfaces/plugin.py index 84ad5bf2..22658ef4 100644 --- a/nodescraper/interfaces/plugin.py +++ b/nodescraper/interfaces/plugin.py @@ -95,18 +95,26 @@ def __init__( self.session_id = session_id self.connection_manager = connection_manager + # True only when this plugin built its connection manager itself; a manager passed in + # belongs to the caller, who is responsible for closing it. + self._owns_connection_manager = False if connection_args and self.CONNECTION_TYPE and not self.connection_manager: + self._owns_connection_manager = True self.connection_manager = self.CONNECTION_TYPE( system_info=self.system_info.model_copy(), logger=logger, connection_args=connection_args, parent=self.__class__.__name__, - task_result_hooks=self.task_result_hooks, + task_result_hooks=self._connection_manager_hooks(), event_reporter=event_reporter, session_id=self.session_id, ) + def _connection_manager_hooks(self) -> list: + """Result hooks for a connection manager this plugin creates itself.""" + return self.task_result_hooks + @classmethod def is_valid(cls) -> bool: """Check if plugin class is valid and can be instantiated diff --git a/nodescraper/models/datapluginresult.py b/nodescraper/models/datapluginresult.py index 2ae73742..2f990d63 100644 --- a/nodescraper/models/datapluginresult.py +++ b/nodescraper/models/datapluginresult.py @@ -37,3 +37,4 @@ class DataPluginResult(BaseModel): system_data: Optional[DataModel] = None collection_result: Optional[TaskResult] = None analysis_result: Optional[TaskResult] = None + connection_result: Optional[TaskResult] = None diff --git a/nodescraper/models/taskresult.py b/nodescraper/models/taskresult.py index 79c5a860..33fac415 100644 --- a/nodescraper/models/taskresult.py +++ b/nodescraper/models/taskresult.py @@ -162,13 +162,22 @@ def summary_str(self) -> str: """Message string for display.""" return self.message or "" - def log_result(self, log_path: str) -> None: - """Write result, artifacts, and events to log_path. Events are merged into a single events.json.""" + def log_result(self, log_path: str, file_suffix: str = "") -> None: + """Write result, artifacts, and events to log_path. + + Args: + log_path (str): Directory to write to. + file_suffix (str, optional): Appended to the ``result`` and ``events`` file names, e.g. + ``"(1)"`` writes ``result(1).json`` and ``events(1).json``. Events are merged into an + existing events file of the same name. Defaults to "". + """ from nodescraper.connection.inband import BaseFileArtifact os.makedirs(log_path, exist_ok=True) - with open(os.path.join(log_path, "result.json"), "w", encoding="utf-8") as log_file: + with open( + os.path.join(log_path, f"result{file_suffix}.json"), "w", encoding="utf-8" + ) as log_file: log_file.write(self.model_dump_json(exclude={"artifacts", "events"}, indent=2)) artifact_map: dict[str, list[dict[str, Any]]] = {} @@ -208,7 +217,7 @@ def log_result(self, log_path: str) -> None: html_file.write(render_command_artifacts_html(html_entries, title)) if self.events: - event_log = os.path.join(log_path, "events.json") + event_log = os.path.join(log_path, f"events{file_suffix}.json") new_events = [e.model_dump(mode="json", exclude_none=True) for e in self.events] existing_events = [] if os.path.isfile(event_log): diff --git a/nodescraper/pluginexecutor.py b/nodescraper/pluginexecutor.py index 4571ab03..33e862c8 100644 --- a/nodescraper/pluginexecutor.py +++ b/nodescraper/pluginexecutor.py @@ -28,6 +28,7 @@ import copy import inspect import logging +import os import uuid from collections import deque from collections.abc import Callable, Sequence @@ -46,12 +47,16 @@ ) from nodescraper.interfaces import ConnectionManager, DataPlugin, PluginInterface from nodescraper.interfaces.taskresulthook import TaskResultHook -from nodescraper.models import PluginConfig, SystemInfo +from nodescraper.models import PluginConfig, SystemInfo, TaskResult from nodescraper.models.pluginresult import PluginResult +from nodescraper.models.taskresult import STATUS_LOG_LEVEL_MAP from nodescraper.pluginregistry import PluginRegistry from nodescraper.taskresulthooks import FileSystemLogHook +from nodescraper.taskresulthooks.filesystemloghook import hooks_for_fixed_directory from nodescraper.typeutils import TypeUtils +OS_DISCOVERY_DIR = "os_discovery" + class PluginExecutor: """Class to manage execution of data collectors and error detectors""" @@ -221,19 +226,33 @@ def merge_configs(plugin_configs: list[PluginConfig]) -> PluginConfig: return merged_config def discover_os_info(self) -> None: - """If the connection library has an InBandConnectionManager, use it to discover OS info and update system_info - self.system_info will be updated with the discovered OS info. + """Use a dedicated InBandConnectionManager to discover OS info and update system_info. + + The discovery connection is separate from the connection library and its result is + saved under ``/os_discovery`` so it does not mix with the results of the + connections used by plugins. self.system_info will be updated with the discovered OS info. """ - inband_connection = self.connection_library.get(InBandConnectionManager) - if inband_connection is None: - # Init use and discard after + if self.log_path: + hooks = hooks_for_fixed_directory( + self.connection_result_hooks, os.path.join(self.log_path, OS_DISCOVERY_DIR) + ) + else: + hooks = list(self.connection_result_hooks) + try: inband_connection = InBandConnectionManager( system_info=self.system_info, logger=self.logger, connection_args=self.connection_configs.get(InBandConnectionManager.__name__), - task_result_hooks=self.connection_result_hooks, + task_result_hooks=hooks, + parent=OS_DISCOVERY_DIR, session_id=self.session_id, ) + except Exception as e: + # Raised for missing or invalid connection args or an invalid session_id. + self.logger.error( + "Error occurred while initializing InBandConnectionManager: %s", str(e) + ) + return try: result = inband_connection.connect() if inband_connection else None if (not inband_connection) or (not result) or (result.status != ExecutionStatus.OK): @@ -244,12 +263,10 @@ def discover_os_info(self) -> None: discover_and_write_os_family(inband_connection, self.system_info, self.logger) except Exception as e: self.logger.error( - "Error occurred during OS discovery with InBandConnectionManager: %s", - str(e), + "Error occurred during OS discovery with InBandConnectionManager: %s", str(e) ) finally: - if inband_connection is not None: - inband_connection.disconnect() + inband_connection.safe_disconnect() def _get_connection_manager_for_plugin( self, @@ -293,8 +310,9 @@ def _get_connection_manager_for_plugin( mgr_impl = connection_manager_class else: self.logger.error( - "Unable to find registered connection manager class for %s that is required by", - connection_manager_class.__name__, + "Unable to find registered connection manager class for %s that is required by %s", + getattr(connection_manager_class, "__name__", connection_manager_class), + plugin_name, ) return None connection_args = None @@ -309,7 +327,13 @@ def _get_connection_manager_for_plugin( } if connection_args is not None: init_kwargs["connection_args"] = connection_args - self.connection_library[mgr_impl] = mgr_impl(**init_kwargs) + try: + self.connection_library[mgr_impl] = mgr_impl(**init_kwargs) + except Exception as e: + self.logger.error( + "Failed to initialize %s for %s: %s", mgr_impl.__name__, plugin_name, e + ) + return None return self.connection_library[mgr_impl] @@ -350,22 +374,24 @@ def _run_single_plugin( "session_id": self.session_id, } + conn_mgr: Optional[ConnectionManager] = None if plugin_class.CONNECTION_TYPE: conn_mgr = self._get_connection_manager_for_plugin(plugin_class, plugin_name) + connection_config = self.connection_configs.get(plugin_class.CONNECTION_TYPE.__name__) + init_payload["connection_args"] = connection_config if conn_mgr is None: return False init_payload["connection_manager"] = conn_mgr - try: plugin_inst = plugin_class(**init_payload) plugin_run_args = copy.deepcopy(plugin_args) - run_args = TypeUtils.get_func_arg_types(plugin_class.run, plugin_class) try: global_run_args = self.apply_global_args_to_plugin( plugin_inst, plugin_class, self.plugin_config.global_args ) + run_payload = self.merge_plugin_run_args(plugin_run_args, global_run_args) except ValueError as ve: self.logger.error( @@ -375,10 +401,6 @@ def _run_single_plugin( ) return False - for arg in run_args.keys(): - if arg == "preserve_connection" and issubclass(plugin_class, DataPlugin): - run_payload[arg] = True - plugin_result = plugin_inst.run(**run_payload) plugin_results.append(plugin_result) for hook in self.plugin_run_result_hooks: @@ -386,6 +408,8 @@ def _run_single_plugin( except Exception as e: self.logger.exception("Unexpected exception when running plugin %s: %s", plugin_name, e) + if conn_mgr is not None: + conn_mgr.safe_disconnect() return True @@ -413,41 +437,57 @@ def run_queue(self) -> list[PluginResult]: except Exception as e: self.logger.exception("Unexpected exception running plugin queue: %s", str(e)) finally: - # Run post-action plugins before tearing down connections or collating - # results so that post-actions still have access to live connections - # and their results are included in the collator output. - self._run_post_actions(plugin_results) - - if self.plugin_config.result_collators: - self.logger.info("Running result collators") - for ( - collator, - collator_args, - ) in self.plugin_config.result_collators.items(): - collator_class = self.plugin_registry.result_collators.get(collator) - if collator_class is None: - self.logger.warning( - "No result collator found in registry for name: %s", - collator, - ) - continue + try: + # Run post-action plugins before tearing down connections or collating + # results so that post-actions still have access to live connections + # and their results are included in the collator output. + self._run_post_actions(plugin_results) + self._run_result_collators(plugin_results) + finally: + self._close_connections() - self.logger.info("Running %s result collator", collator) - collator_inst = collator_class(logger=self.logger, log_path=self.log_path) - collator_inst.collate_results( - plugin_results, - [ - connection_manager.result - for connection_manager in self.connection_library.values() - ], - **collator_args, - ) + return plugin_results - self.logger.info("Closing connections") - for connection_manager in self.connection_library.values(): - connection_manager.disconnect() + def _run_result_collators(self, plugin_results: list[PluginResult]) -> None: + """Run the configured result collators over the plugin and connection results.""" + if not self.plugin_config.result_collators: + return - return plugin_results + self.logger.info("Running result collators") + connection_results = [ + connection_manager.result for connection_manager in self.connection_library.values() + ] + for collator, collator_args in self.plugin_config.result_collators.items(): + collator_class = self.plugin_registry.result_collators.get(collator) + if collator_class is None: + self.logger.warning("No result collator found in registry for name: %s", collator) + continue + + self.logger.info("Running %s result collator", collator) + collator_inst = collator_class(logger=self.logger, log_path=self.log_path) + collator_inst.collate_results(plugin_results, connection_results, **collator_args) + + def _log_connection_result(self, result: TaskResult) -> None: + self.logger.log( + STATUS_LOG_LEVEL_MAP.get(result.status, logging.INFO), + "Connection result for %s: %s%s", + result.task, + result.status.name, + f" - {result.message}" if result.message else "", + ) + + def _close_connections(self) -> None: + """Close every connection and log the connection results. + + Each manager's own result was already saved by connect(). A failing disconnect cannot + prevent the remaining connections from being closed. + """ + self.logger.info("Closing connections") + for connection_manager in self.connection_library.values(): + connection_manager.safe_disconnect() + result = connection_manager.result + if result.status != ExecutionStatus.UNSET or result.events: + self._log_connection_result(result) def _run_post_actions(self, plugin_results: list[PluginResult]) -> None: """Evaluate post-action conditions and run qualifying plugins. @@ -548,24 +588,17 @@ def apply_global_args_to_plugin( args_cls = plugin_class._collector_args_class(collector_cls) if args_cls is None: continue - plugin_fields = set(args_cls.model_fields.keys()) - filtered = { - k: v - for k, v in global_args["collection_args"].items() - if k in plugin_fields - } + filtered = TypeUtils.filter_to_model_fields( + args_cls, global_args["collection_args"] + ) if filtered: per_collector_args[collector_cls.__name__] = filtered if per_collector_args: run_args["collection_args"] = per_collector_args elif collector_args is not None and not isinstance(collector_args, dict): - args_cls = ( - collector_args if isinstance(collector_args, type) else type(collector_args) + filtered = TypeUtils.filter_to_model_fields( + collector_args, global_args["collection_args"] ) - plugin_fields = set(args_cls.model_fields.keys()) - filtered = { - k: v for k, v in global_args["collection_args"].items() if k in plugin_fields - } if filtered: run_args["collection_args"] = filtered @@ -574,10 +607,9 @@ def apply_global_args_to_plugin( and hasattr(plugin_class, "ANALYZER_ARGS") and plugin_class.ANALYZER_ARGS is not None ): - analyzer_args = plugin_class.ANALYZER_ARGS - args_cls = analyzer_args if isinstance(analyzer_args, type) else type(analyzer_args) - plugin_fields = set(args_cls.model_fields.keys()) - filtered = {k: v for k, v in global_args["analysis_args"].items() if k in plugin_fields} + filtered = TypeUtils.filter_to_model_fields( + plugin_class.ANALYZER_ARGS, global_args["analysis_args"] + ) if filtered: run_args["analysis_args"] = filtered diff --git a/nodescraper/plugins/serviceability/mi4xx/serviceability_plugin_mi4xx.py b/nodescraper/plugins/serviceability/mi4xx/serviceability_plugin_mi4xx.py index 4341f5d4..464b7b88 100644 --- a/nodescraper/plugins/serviceability/mi4xx/serviceability_plugin_mi4xx.py +++ b/nodescraper/plugins/serviceability/mi4xx/serviceability_plugin_mi4xx.py @@ -175,7 +175,7 @@ def run( # type: ignore[override] Union[SystemInteractionLevel, str], "System interaction level (e.g. PASSIVE, INTERACTIVE, DISRUPTIVE).", ] = SystemInteractionLevel.INTERACTIVE, - preserve_connection: bool = False, + preserve_connection: bool = True, data: Annotated[ Optional[Union[str, dict, ServiceabilityDataModel]], Field( diff --git a/nodescraper/taskresulthooks/filesystemloghook.py b/nodescraper/taskresulthooks/filesystemloghook.py index 7e90ab54..40656f70 100644 --- a/nodescraper/taskresulthooks/filesystemloghook.py +++ b/nodescraper/taskresulthooks/filesystemloghook.py @@ -31,6 +31,19 @@ from nodescraper.utils import resolve_log_dir_name +def _next_log_suffix(log_path: str) -> str: + """First suffix ("", "(1)", "(2)", ...) for which neither result nor events file exists in log_path.""" + count = 0 + while True: + suffix = f"({count})" if count else "" + if not any( + os.path.exists(os.path.join(log_path, f"{name}{suffix}.json")) + for name in ("result", "events") + ): + return suffix + count += 1 + + class FileSystemLogHook(TaskResultHook): def __init__(self, log_base_path=None, include_task_path: bool = True, **kwargs) -> None: """Create a FileSystemLogHook Instance @@ -47,7 +60,12 @@ def __init__(self, log_base_path=None, include_task_path: bool = True, **kwargs) self.include_task_path = include_task_path def process_result(self, task_result: TaskResult, data: Optional[DataModel] = None, **kwargs): - """Log task result to the filesystem (single events.json per directory).""" + """Log task result to the filesystem. + + The first result written to a directory uses ``result.json`` / ``events.json``. Later results + sent to the same directory get an integer appended (``result(1).json`` / ``events(1).json``) + so earlier logs are never overwritten. + """ log_path = self.log_base_path if self.include_task_path: if task_result.parent: @@ -55,7 +73,7 @@ def process_result(self, task_result: TaskResult, data: Optional[DataModel] = No if task_result.task: log_path = os.path.join(log_path, resolve_log_dir_name(task_result.task)) - task_result.log_result(log_path) + task_result.log_result(log_path, _next_log_suffix(log_path)) if data: data.log_model(log_path) diff --git a/nodescraper/typeutils.py b/nodescraper/typeutils.py index 65ccf605..440a8a2b 100644 --- a/nodescraper/typeutils.py +++ b/nodescraper/typeutils.py @@ -195,3 +195,22 @@ def get_model_types(cls, model: type[BaseModel]) -> dict[str, TypeData]: ) return type_map + + @staticmethod + def filter_to_model_fields( + model: Union[BaseModel, Type[BaseModel]], data: dict[str, Any] + ) -> dict[str, Any]: + """Keep only the entries of ``data`` whose keys are fields of a pydantic model + + Use this to apply shared (global) arguments to a model that forbids extra keys, so keys + meant for other models are dropped instead of raising a validation error. + + Args: + model (Union[BaseModel, Type[BaseModel]]): model class or instance whose fields are allowed + data (dict[str, Any]): arguments to filter; not modified + + Returns: + dict[str, Any]: new dict with only the keys that are fields of the model + """ + model_cls = model if isinstance(model, type) else type(model) + return {key: value for key, value in data.items() if key in model_cls.model_fields} diff --git a/test/functional/test_os_plugin.py b/test/functional/test_os_plugin.py index 461edc56..12d6103a 100644 --- a/test/functional/test_os_plugin.py +++ b/test/functional/test_os_plugin.py @@ -184,3 +184,78 @@ def test_os_plugin_with_executor_programmatic(plugin_executor, tmp_path): assert hasattr(executor, "system_info") # OS family should be discovered and set assert executor.system_info.os_family is not None + + +def _run_os_plugin(plugin_executor, tmp_path, system_info=None, connections=None): + from nodescraper.models.pluginconfig import PluginConfig + + config = PluginConfig( + global_args={}, + plugins={"OsPlugin": {"analysis": False, "collection": True}}, + name="Connection result run", + desc="Check connection result is saved with the plugin result", + ) + executor = plugin_executor( + plugin_configs=[config], + log_path=str(tmp_path / "logs"), + system_info=system_info, + connections=connections, + ) + results = executor.run_queue() + os_results = [r for r in results if r.source == "OsPlugin"] + assert len(os_results) == 1, "OsPlugin did not produce a result" + return executor, os_results[0] + + +def test_connection_result_saved_on_successful_connection(plugin_executor, tmp_path): + """A successful connection is recorded as OK in result_data and survives disconnect.""" + from nodescraper.enums import ExecutionStatus + + executor, os_result = _run_os_plugin(plugin_executor, tmp_path) + + connection_result = os_result.result_data.connection_result + assert connection_result is not None, "connection_result was not saved" + assert connection_result.status == ExecutionStatus.OK + assert connection_result.task == "InBandConnectionManager" + + # run_queue disconnects every manager on exit; the result must not be reset to UNSET. + for manager in executor.connection_library.values(): + assert manager.connection is None + assert manager.result.status == ExecutionStatus.OK + + +def test_connection_result_saved_on_failed_connection(plugin_executor, tmp_path): + """A remote connection without credentials fails, reports an event and is saved.""" + from nodescraper.enums import ExecutionStatus, SystemLocation + from nodescraper.models.systeminfo import SystemInfo + + system_info = SystemInfo(name="unreachable_host", location=SystemLocation.REMOTE) + executor, os_result = _run_os_plugin(plugin_executor, tmp_path, system_info=system_info) + + connection_result = os_result.result_data.connection_result + assert connection_result is not None, "connection_result was not saved" + assert connection_result.status in ( + ExecutionStatus.ERROR, + ExecutionStatus.EXECUTION_FAILURE, + ) + assert len(connection_result.events) >= 1 + assert os_result.result_data.collection_result.status == ExecutionStatus.NOT_RAN + assert os_result.status != ExecutionStatus.OK + + for manager in executor.connection_library.values(): + assert manager.result.status in (ExecutionStatus.ERROR, ExecutionStatus.EXECUTION_FAILURE) + + +def test_connection_result_written_to_log_path(plugin_executor, tmp_path): + """The connection result is logged to disk by the connection result hook.""" + import json + + _run_os_plugin(plugin_executor, tmp_path) + + result_files = [ + p + for p in (tmp_path / "logs").rglob("result.json") + if json.loads(p.read_text()).get("task") == "InBandConnectionManager" + ] + assert result_files, "No connection result.json found under the log path" + assert json.loads(result_files[0].read_text())["status"] == "OK" diff --git a/test/unit/connection/test_connectionmanager.py b/test/unit/connection/test_connectionmanager.py index 1c7d6a8a..def18b4b 100644 --- a/test/unit/connection/test_connectionmanager.py +++ b/test/unit/connection/test_connectionmanager.py @@ -79,8 +79,8 @@ def test_disconnect_clears_connection(self): self.assertIsNone(manager.connection) - def test_disconnect_resets_status(self): - """Test that disconnect resets the result status to UNSET""" + def test_disconnect_preserves_status(self): + """Test that disconnect keeps the result of the connection attempt""" system_info = SystemInfo( name="test_system", location=SystemLocation.LOCAL, @@ -91,7 +91,48 @@ def test_disconnect_resets_status(self): manager.disconnect() - self.assertEqual(manager.result.status, ExecutionStatus.UNSET) + self.assertEqual(manager.result.status, ExecutionStatus.OK) + + def test_connect_success_sets_ok_and_survives_disconnect(self): + manager = DummyConnectionManager( + system_info=SystemInfo(name="test_system", location=SystemLocation.LOCAL) + ) + + result = manager.connect() + manager.disconnect() + + self.assertEqual(result.status, ExecutionStatus.OK) + self.assertEqual(manager.result.status, ExecutionStatus.OK) + + def test_connect_exception_sets_failure_with_event(self): + class FailingManager(DummyConnectionManager): + def connect(self) -> TaskResult: + raise RuntimeError("boom") + + manager = FailingManager( + system_info=SystemInfo(name="test_system", location=SystemLocation.LOCAL) + ) + + result = manager.connect() + manager.disconnect() + + self.assertEqual(result.status, ExecutionStatus.EXECUTION_FAILURE) + self.assertEqual(manager.result.status, ExecutionStatus.EXECUTION_FAILURE) + self.assertGreaterEqual(len(result.events), 1) + + def test_connect_without_connection_reports_failure_event(self): + class NoConnectionManager(DummyConnectionManager): + def connect(self) -> TaskResult: + return self.result + + manager = NoConnectionManager( + system_info=SystemInfo(name="test_system", location=SystemLocation.LOCAL) + ) + + result = manager.connect() + + self.assertEqual(result.status, ExecutionStatus.EXECUTION_FAILURE) + self.assertEqual(len(result.events), 1) def test_context_manager_calls_disconnect(self): """Test that using connection manager as context manager calls disconnect on exit""" diff --git a/test/unit/framework/test_dataplugin.py b/test/unit/framework/test_dataplugin.py index 108d3a09..48b41a3c 100644 --- a/test/unit/framework/test_dataplugin.py +++ b/test/unit/framework/test_dataplugin.py @@ -32,11 +32,17 @@ from framework.common.shared_utils import MockConnectionManager from nodescraper import utils -from nodescraper.enums import EventPriority, ExecutionStatus, SystemInteractionLevel +from nodescraper.enums import ( + EventCategory, + EventPriority, + ExecutionStatus, + SystemInteractionLevel, +) from nodescraper.interfaces.dataanalyzertask import DataAnalyzer from nodescraper.interfaces.datacollectortask import DataCollector from nodescraper.interfaces.dataplugin import DataPlugin -from nodescraper.models import CollectorArgs, DataModel, TaskResult +from nodescraper.interfaces.task import SystemCompatibilityError +from nodescraper.models import CollectorArgs, DataModel, DataPluginResult, TaskResult class StandardDataModel(DataModel): @@ -336,7 +342,7 @@ def test_run_with_parameters(self, plugin_with_conn): analysis=True, max_event_priority_level=EventPriority.ERROR, system_interaction_level=SystemInteractionLevel.PASSIVE, - preserve_connection=True, + preserve_connection=False, data=data, collection_args=collection_args, analysis_args=analysis_args, @@ -345,17 +351,16 @@ def test_run_with_parameters(self, plugin_with_conn): mock_collect.assert_called_once_with( max_event_priority_level=EventPriority.ERROR, system_interaction_level=SystemInteractionLevel.PASSIVE, + preserve_connection=False, collection_args=collection_args, - preserve_connection=True, ) mock_analyze.assert_called_once_with( max_event_priority_level=EventPriority.ERROR, analysis_args=analysis_args, data=data ) - def test_collect_preserve_connection(self, plugin_with_conn): - """Test the behavior of preserve_connection parameter in collect method.""" - # Test with preserve_connection=True + def test_collect_leaves_externally_owned_connection_open(self, plugin_with_conn): + """A connection manager passed in by the caller is closed by its owner, not the plugin.""" with patch.object(BaseDataCollector, "collect_data") as mock_collect: with patch.object(MockConnectionManager, "disconnect") as mock_disconnect: mock_collect.return_value = ( @@ -363,17 +368,31 @@ def test_collect_preserve_connection(self, plugin_with_conn): StandardDataModel(), ) - # Call collect with preserve_connection=True - result = plugin_with_conn.collect(preserve_connection=True) + result = plugin_with_conn.collect() - # Verify collect_data was called and result is OK mock_collect.assert_called_once() assert result.status == ExecutionStatus.OK - - # Verify disconnect was NOT called when preserve_connection=True mock_disconnect.assert_not_called() - # Test with preserve_connection=False (default) + def test_collect_closes_passed_in_connection_when_preserve_connection_false( + self, plugin_with_conn + ): + with patch.object(MockConnectionManager, "disconnect") as mock_disconnect: + result = plugin_with_conn.collect(preserve_connection=False) + + assert result.status == ExecutionStatus.OK + mock_disconnect.assert_called_once() + + def test_run_forwards_preserve_connection_to_collect(self, plugin_with_conn): + with patch.object(MockConnectionManager, "disconnect") as mock_disconnect: + plugin_with_conn.run(analysis=False, preserve_connection=False) + assert mock_disconnect.call_count == 1 + + plugin_with_conn.run(analysis=False) + assert mock_disconnect.call_count == 1 + + def test_collect_closes_connection_it_created(self, plugin): + """A connection manager created by the plugin itself is closed after collection.""" with patch.object(BaseDataCollector, "collect_data") as mock_collect: with patch.object(MockConnectionManager, "disconnect") as mock_disconnect: mock_collect.return_value = ( @@ -381,14 +400,10 @@ def test_collect_preserve_connection(self, plugin_with_conn): StandardDataModel(), ) - # Call collect with preserve_connection=False (default) - result = plugin_with_conn.collect(preserve_connection=False) + result = plugin.collect() - # Verify collect_data was called and result is OK mock_collect.assert_called_once() assert result.status == ExecutionStatus.OK - - # Verify disconnect WAS called when preserve_connection=False mock_disconnect.assert_called_once() def test_run_with_data_file_no_collection(self, plugin_with_conn, tmp_path): @@ -861,7 +876,7 @@ def test_find_datamodel_path_honors_registered_log_dir_name( log_path=str(run_path), ) - assert plugin.collect(preserve_connection=True).status == ExecutionStatus.OK + assert plugin.collect().status == ExecutionStatus.OK # collect() wrote the collector output under the registered override names collector_dir = run_path / "override_LOGDIR_plugin" / "override_LOGDIR_collector" @@ -882,8 +897,548 @@ def test_load_run_data_honors_registered_log_dir_name( connection_manager=plugin_with_conn.connection_manager, log_path=str(run_path), ) - plugin.collect(preserve_connection=True) + plugin.collect() loaded = OverrideLogDirPlugin.load_run_data(str(run_path)) assert loaded is not None assert loaded["value"] == "collected" + + +class TestDataPluginConnectionResult: + def test_run_includes_connection_result_when_connection_initiated(self, plugin_with_conn): + plugin_with_conn.connection_manager.result.status = ExecutionStatus.EXECUTION_FAILURE + plugin_with_conn.connection_manager.connection = None + + result = plugin_with_conn.run(analysis=False) + + assert result.result_data.connection_result is not None + assert result.result_data.connection_result.status == ExecutionStatus.EXECUTION_FAILURE + assert result.result_data.collection_result.status == ExecutionStatus.NOT_RAN + + def test_run_has_no_connection_result_without_connection_attempt(self, plugin): + with patch.object(CoreDataPlugin, "CONNECTION_TYPE", None): + result = plugin.run(analysis=False) + + assert result.result_data.connection_result is None + + +class FailingConnectionManager(MockConnectionManager): + """Connection manager whose connect() fails without producing a connection.""" + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.connection = None + self.result = TaskResult(task=self.__class__.__name__, parent=self.parent) + + def connect(self): + self._log_event( + category=EventCategory.SSH, + description="Unable to reach host", + priority=EventPriority.CRITICAL, + ) + self.result.status = ExecutionStatus.EXECUTION_FAILURE + return self.result + + +class FailingConnectionPlugin(CoreDataPlugin): + CONNECTION_TYPE = FailingConnectionManager + + +class SilentFailingConnectionManager(FailingConnectionManager): + """Connection manager that returns without a connection, status or event.""" + + def connect(self): + return self.result + + +class SilentFailingConnectionPlugin(CoreDataPlugin): + CONNECTION_TYPE = SilentFailingConnectionManager + + +class TestDataPluginResultConnection: + """connection_result is saved in DataPluginResult whenever a connection is initiated.""" + + def test_default_is_none(self): + assert DataPluginResult().connection_result is None + + def test_successful_connection_saved_after_disconnect(self, plugin): + result = plugin.run() + + connection_result = result.result_data.connection_result + assert connection_result is not None + assert connection_result.status == ExecutionStatus.OK + assert result.status == ExecutionStatus.OK + + def test_connection_result_reported_for_externally_owned_connection(self, plugin_with_conn): + result = plugin_with_conn.run() + + assert result.result_data.connection_result.status == ExecutionStatus.OK + + def test_failed_connection_saved_with_event(self, system_info, logger): + plugin = FailingConnectionPlugin(system_info=system_info, logger=logger) + + result = plugin.run() + + connection_result = result.result_data.connection_result + assert connection_result is not None + assert connection_result.status == ExecutionStatus.EXECUTION_FAILURE + assert any(e.description == "Unable to reach host" for e in connection_result.events) + assert result.result_data.collection_result.status == ExecutionStatus.NOT_RAN + assert result.result_data.collection_result.message == ( + "Connection not available, data collection skipped" + ) + + def test_failed_connection_not_reset_by_disconnect(self, system_info, logger): + plugin = FailingConnectionPlugin(system_info=system_info, logger=logger) + + plugin.collect() + plugin.connection_manager.disconnect() + + assert plugin.connection_manager.result.status == ExecutionStatus.EXECUTION_FAILURE + + def test_failed_connection_is_not_retried(self, system_info, logger): + plugin = FailingConnectionPlugin(system_info=system_info, logger=logger) + + with patch.object( + FailingConnectionManager, "connect", autospec=True, side_effect=lambda s: s.result + ) as mock_connect: + plugin.collect() + plugin.connection_manager.result.status = ExecutionStatus.EXECUTION_FAILURE + plugin.collect() + plugin.collect() + + assert mock_connect.call_count == 1 + + def test_silent_connect_failure_gets_status_and_event(self, system_info, logger): + plugin = SilentFailingConnectionPlugin(system_info=system_info, logger=logger) + + result = plugin.run() + + connection_result = result.result_data.connection_result + assert connection_result.status == ExecutionStatus.EXECUTION_FAILURE + assert len(connection_result.events) >= 1 + + def test_reconnects_after_successful_disconnect(self, plugin_with_conn): + manager = plugin_with_conn.connection_manager + manager.connection = None + manager.result.status = ExecutionStatus.OK + + with patch.object(MockConnectionManager, "connect", autospec=True) as mock_connect: + mock_connect.side_effect = lambda s: setattr(s, "connection", MagicMock()) or s.result + plugin_with_conn.collect() + + mock_connect.assert_called_once() + + def test_connection_result_is_a_snapshot(self, plugin_with_conn): + result = plugin_with_conn.run(analysis=False) + + plugin_with_conn.connection_manager.result.status = ExecutionStatus.EXECUTION_FAILURE + + assert result.result_data.connection_result.status == ExecutionStatus.OK + + def test_connection_result_serializes(self, plugin_with_conn): + result = plugin_with_conn.run(analysis=False) + + dumped = result.result_data.model_dump(mode="json") + + assert dumped["connection_result"]["status"] == "OK" + + +class WarningConnectionManager(MockConnectionManager): + """Connection manager whose successful connect() reports one warning event.""" + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.connection = None + self.result = TaskResult(task=self.__class__.__name__, parent=self.parent) + + def connect(self): + self._log_event( + category=EventCategory.SSH, + description="Slow handshake", + priority=EventPriority.WARNING, + ) + self.connection = MagicMock() + self.result.status = ExecutionStatus.OK + return self.result + + +class WarningConnectionPlugin(CoreDataPlugin): + CONNECTION_TYPE = WarningConnectionManager + + +class TestDataPluginSavesConnectionResult: + """The plugin saves the connection result when it closes the connection itself.""" + + @staticmethod + def _conn_dir(log_path: Path) -> Path: + # manager created by the plugin: //connection + return ( + log_path + / utils.resolve_log_dir_name("WarningConnectionPlugin") + / utils.resolve_log_dir_name("BaseDataCollector") + / "connection" + ) + + def test_successful_connection_saved_when_plugin_owns_connection( + self, system_info, logger, tmp_path + ): + plugin = WarningConnectionPlugin( + system_info=system_info, logger=logger, log_path=str(tmp_path) + ) + + plugin.run(analysis=False) + + conn_dir = self._conn_dir(tmp_path) + assert json.loads((conn_dir / "result.json").read_text())["status"] == "OK" + # connect() already saved the warning; closing the connection must not save it again + events = json.loads((conn_dir / "events.json").read_text()) + assert [e["description"] for e in events] == ["Slow handshake"] + + def test_failed_connection_result_saved_when_plugin_owns_connection( + self, system_info, logger, tmp_path + ): + plugin = FailingConnectionPlugin( + system_info=system_info, logger=logger, log_path=str(tmp_path) + ) + + plugin.run(analysis=False) + + conn_dir = ( + tmp_path + / utils.resolve_log_dir_name("FailingConnectionPlugin") + / utils.resolve_log_dir_name("BaseDataCollector") + / "connection" + ) + assert json.loads((conn_dir / "result.json").read_text())["status"] == "EXECUTION_FAILURE" + events = json.loads((conn_dir / "events.json").read_text()) + assert [e["description"] for e in events] == ["Unable to reach host"] + + +class LifecycleConnectionManager(MockConnectionManager): + """Starts disconnected, connects on demand and counts disconnects per instance.""" + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.connection = None + self.result = self._init_result() + self.disconnect_calls = 0 + + def connect(self): + self.connection = MagicMock() + self.result.status = ExecutionStatus.OK + return self.result + + def disconnect(self): + self.disconnect_calls += 1 + self.connection = None + + +class RaisingConnectManager(LifecycleConnectionManager): + def connect(self): + raise RuntimeError("connect boom") + + +class RaisingInitManager(LifecycleConnectionManager): + def __init__(self, *args, **kwargs): + raise ValueError("bad connection args") + + +class RaisingDisconnectManager(LifecycleConnectionManager): + def disconnect(self): + super().disconnect() + raise OSError("close boom") + + +class InitRaisingCollector(BaseDataCollector): + def __init__(self, *args, **kwargs): + raise RuntimeError("collector init boom") + + +class CollectRaisingCollector(BaseDataCollector): + def collect_data(self, args=None): + raise RuntimeError("collect boom") + + +class IncompatibleCollector(BaseDataCollector): + def __init__(self, *args, **kwargs): + raise SystemCompatibilityError("unsupported system") + + +def _own_connection_plugin(manager_cls, collector_cls=BaseDataCollector): + class OwnConnectionPlugin(DataPlugin): + DATA_MODEL = StandardDataModel + CONNECTION_TYPE = manager_cls + COLLECTOR = collector_cls + ANALYZER = StandardAnalyzer + + return OwnConnectionPlugin + + +class TestDataPluginOwnConnection: + """The DataPlugin creates the connection manager itself and must clean it up.""" + + @staticmethod + def _conn_dir(log_path: Path, collector_cls=BaseDataCollector) -> Path: + return ( + log_path + / utils.resolve_log_dir_name("OwnConnectionPlugin") + / utils.resolve_log_dir_name(collector_cls.__name__) + / "connection" + ) + + @staticmethod + def _external_conn_dir(log_path: Path) -> Path: + # a manager supplied by the caller keeps the caller's hooks: / + return ( + log_path + / utils.resolve_log_dir_name("OwnConnectionPlugin") + / utils.resolve_log_dir_name("LifecycleConnectionManager") + ) + + @staticmethod + def _saved_result(conn_dir: Path) -> dict: + return json.loads((conn_dir / "result.json").read_text()) + + @staticmethod + def _assert_only_plugin_folder(log_path: Path): + # nothing may be written next to (or outside) the plugin's own folder + assert {p.name for p in log_path.iterdir() if p.is_dir()} == { + utils.resolve_log_dir_name("OwnConnectionPlugin") + } + + def _build(self, system_info, logger, tmp_path, manager_cls, collector_cls=BaseDataCollector): + plugin = _own_connection_plugin(manager_cls, collector_cls)( + system_info=system_info, logger=logger, log_path=str(tmp_path) + ) + assert plugin.connection_manager is None + return plugin + + def test_created_connection_is_closed_and_saved_in_plugin_folder( + self, system_info, logger, tmp_path + ): + plugin = self._build(system_info, logger, tmp_path, LifecycleConnectionManager) + + result = plugin.run() + + manager = plugin.connection_manager + assert isinstance(manager, LifecycleConnectionManager) + assert manager.disconnect_calls == 1 + assert manager.connection is None + conn_dir = self._conn_dir(tmp_path) + saved = self._saved_result(conn_dir) + assert saved["status"] == "OK" + assert saved["task"] == "LifecycleConnectionManager" + assert saved["parent"] == "OwnConnectionPlugin" + assert not (conn_dir / "events.json").exists() + self._assert_only_plugin_folder(tmp_path) + assert result.result_data.connection_result.status == ExecutionStatus.OK + + def test_connection_result_is_kept_apart_from_collector_files( + self, system_info, logger, tmp_path + ): + plugin = self._build(system_info, logger, tmp_path, LifecycleConnectionManager) + + plugin.collect() + + collector_dir = ( + tmp_path + / utils.resolve_log_dir_name("OwnConnectionPlugin") + / utils.resolve_log_dir_name("BaseDataCollector") + ) + conn_dir = collector_dir / "connection" + assert json.loads((conn_dir / "result.json").read_text())["task"] == ( + "LifecycleConnectionManager" + ) + # the connection's files are not mixed into the collector folder itself + assert not (collector_dir / "result.json").exists() + assert not (collector_dir / "events.json").exists() + # nothing is written under a per-manager folder next to the collector + assert not ( + tmp_path + / utils.resolve_log_dir_name("OwnConnectionPlugin") + / utils.resolve_log_dir_name("LifecycleConnectionManager") + ).exists() + + def test_externally_owned_connection_is_left_open_for_its_owner( + self, system_info, logger, tmp_path + ): + plugin = self._build(system_info, logger, tmp_path, LifecycleConnectionManager) + manager = LifecycleConnectionManager( + system_info=system_info, + logger=logger, + parent="OwnConnectionPlugin", + task_result_hooks=plugin.task_result_hooks, + ) + plugin.connection_manager = manager + + plugin.collect() + + # the plugin did not create the manager, so it must not close it + assert manager.disconnect_calls == 0 + assert manager.connection is not None + conn_dir = self._external_conn_dir(tmp_path) + assert self._saved_result(conn_dir)["status"] == "OK" + + # the owner closes its own connection + manager.disconnect() + assert manager.disconnect_calls == 1 + assert self._saved_result(conn_dir)["status"] == "OK" + self._assert_only_plugin_folder(tmp_path) + + def test_passed_in_connection_closed_and_saved_when_preserve_connection_false( + self, system_info, logger, tmp_path + ): + plugin = self._build(system_info, logger, tmp_path, LifecycleConnectionManager) + manager = LifecycleConnectionManager( + system_info=system_info, + logger=logger, + parent="OwnConnectionPlugin", + task_result_hooks=plugin.task_result_hooks, + ) + plugin.connection_manager = manager + + plugin.collect(preserve_connection=False) + + assert manager.disconnect_calls == 1 + assert manager.connection is None + saved = self._saved_result(self._external_conn_dir(tmp_path)) + assert saved["status"] == "OK" + self._assert_only_plugin_folder(tmp_path) + + def test_owned_connection_closed_even_when_preserve_connection_true( + self, system_info, logger, tmp_path + ): + plugin = self._build(system_info, logger, tmp_path, LifecycleConnectionManager) + + plugin.collect(preserve_connection=True) + + assert plugin.connection_manager.disconnect_calls == 1 + + def test_externally_owned_failed_connection_is_not_closed_by_plugin( + self, system_info, logger, tmp_path + ): + plugin = self._build(system_info, logger, tmp_path, RaisingConnectManager) + manager = RaisingConnectManager( + system_info=system_info, + logger=logger, + parent="OwnConnectionPlugin", + task_result_hooks=plugin.task_result_hooks, + ) + plugin.connection_manager = manager + + plugin.collect() + + assert manager.disconnect_calls == 0 + + def test_manager_built_from_connection_args_is_owned_and_closed( + self, system_info, logger, tmp_path + ): + plugin = _own_connection_plugin(LifecycleConnectionManager)( + system_info=system_info, + logger=logger, + log_path=str(tmp_path), + connection_args={"host": "example"}, + ) + manager = plugin.connection_manager + assert isinstance(manager, LifecycleConnectionManager) + + plugin.collect() + + assert manager.disconnect_calls == 1 + saved = self._saved_result(self._conn_dir(tmp_path)) + assert saved["status"] == "OK" + + @pytest.mark.parametrize( + "collector_cls, expected_status", + [ + (InitRaisingCollector, ExecutionStatus.EXECUTION_FAILURE), + (CollectRaisingCollector, ExecutionStatus.EXECUTION_FAILURE), + (IncompatibleCollector, ExecutionStatus.NOT_RAN), + ], + ) + def test_collector_exception_still_closes_and_saves_connection( + self, system_info, logger, tmp_path, collector_cls, expected_status + ): + plugin = self._build( + system_info, logger, tmp_path, LifecycleConnectionManager, collector_cls + ) + + result = plugin.collect() + + assert result.status == expected_status + manager = plugin.connection_manager + assert manager.disconnect_calls == 1 + assert manager.connection is None + # the connection itself was fine, so it is saved as OK in the plugin folder + saved = self._saved_result(self._conn_dir(tmp_path, collector_cls)) + assert saved["status"] == "OK" + self._assert_only_plugin_folder(tmp_path) + + def test_connect_exception_is_saved_as_failure_and_connection_closed( + self, system_info, logger, tmp_path + ): + plugin = self._build(system_info, logger, tmp_path, RaisingConnectManager) + + result = plugin.run(analysis=False) + + assert result.result_data.collection_result.status == ExecutionStatus.NOT_RAN + manager = plugin.connection_manager + assert manager.disconnect_calls == 1 + conn_dir = self._conn_dir(tmp_path) + assert self._saved_result(conn_dir)["status"] == "EXECUTION_FAILURE" + events = json.loads((conn_dir / "events.json").read_text()) + # reported once by connect(); closing the connection must not repeat it + assert len(events) == 1 + assert events[0]["priority"] == "CRITICAL" + assert events[0]["data"]["exception_type"] == "RuntimeError" + assert "connect boom" in events[0]["description"] + self._assert_only_plugin_folder(tmp_path) + + def test_connection_manager_creation_failure_does_not_raise( + self, system_info, logger, tmp_path + ): + plugin = self._build(system_info, logger, tmp_path, RaisingInitManager) + + result = plugin.collect() + + assert result.status == ExecutionStatus.EXECUTION_FAILURE + assert "bad connection args" in result.message + assert plugin.connection_manager is None + assert list(tmp_path.iterdir()) == [] + + def test_disconnect_exception_still_saves_result_and_does_not_raise( + self, system_info, logger, tmp_path + ): + plugin = self._build(system_info, logger, tmp_path, RaisingDisconnectManager) + + result = plugin.collect() + + assert result.status == ExecutionStatus.OK + assert plugin.connection_manager.disconnect_calls == 1 + saved = self._saved_result(self._conn_dir(tmp_path)) + assert saved["status"] == "OK" + + def test_without_log_path_connection_is_closed_and_nothing_written( + self, system_info, logger, tmp_path, monkeypatch + ): + monkeypatch.chdir(tmp_path) + plugin = _own_connection_plugin(LifecycleConnectionManager)( + system_info=system_info, logger=logger + ) + + result = plugin.collect() + + assert result.status == ExecutionStatus.OK + assert plugin.connection_manager.disconnect_calls == 1 + assert list(tmp_path.iterdir()) == [] + + def test_reconnects_and_saves_again_on_second_collect(self, system_info, logger, tmp_path): + plugin = self._build(system_info, logger, tmp_path, LifecycleConnectionManager) + + plugin.collect() + plugin.collect() + + assert plugin.connection_manager.disconnect_calls == 2 + saved = self._saved_result(self._conn_dir(tmp_path)) + assert saved["status"] == "OK" diff --git a/test/unit/framework/test_filesystemloghook.py b/test/unit/framework/test_filesystemloghook.py new file mode 100644 index 00000000..b31e6b08 --- /dev/null +++ b/test/unit/framework/test_filesystemloghook.py @@ -0,0 +1,135 @@ +############################################################################### +# +# MIT License +# +# Copyright (c) 2025 Advanced Micro Devices, Inc. +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in all +# copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +# SOFTWARE. +# +############################################################################### +import json + +from nodescraper.enums import EventCategory, EventPriority, ExecutionStatus +from nodescraper.models import DataModel, Event, TaskResult +from nodescraper.taskresulthooks import FileSystemLogHook + + +class _Model(DataModel): + value: str = "x" + + +def _result(task="MyTask", parent="MyParent", message="", events=()): + return TaskResult( + task=task, + parent=parent, + status=ExecutionStatus.OK, + message=message, + events=[ + Event( + reporter="test", + category=EventCategory.RUNTIME, + description=d, + priority=EventPriority.WARNING, + ) + for d in events + ], + ) + + +def _names(path): + return sorted(p.name for p in path.iterdir()) + + +def _read(path): + return json.loads(path.read_text()) + + +def test_first_result_uses_plain_file_names(tmp_path): + hook = FileSystemLogHook(log_base_path=str(tmp_path), include_task_path=False) + + hook.process_result(_result(events=["one"])) + + assert _names(tmp_path) == ["events.json", "result.json"] + + +def test_later_results_in_same_directory_get_an_integer_and_keep_earlier_logs(tmp_path): + hook = FileSystemLogHook(log_base_path=str(tmp_path), include_task_path=False) + + hook.process_result(_result(message="first", events=["one"])) + hook.process_result(_result(message="second", events=["two"])) + hook.process_result(_result(message="third", events=["three"])) + + assert _names(tmp_path) == [ + "events(1).json", + "events(2).json", + "events.json", + "result(1).json", + "result(2).json", + "result.json", + ] + assert _read(tmp_path / "result.json")["message"] == "first" + assert _read(tmp_path / "result(1).json")["message"] == "second" + assert _read(tmp_path / "result(2).json")["message"] == "third" + # events are not merged across results: each result keeps its own events file + assert [e["description"] for e in _read(tmp_path / "events.json")] == ["one"] + assert [e["description"] for e in _read(tmp_path / "events(1).json")] == ["two"] + assert [e["description"] for e in _read(tmp_path / "events(2).json")] == ["three"] + + +def test_result_and_events_share_the_same_index_when_earlier_result_had_no_events(tmp_path): + hook = FileSystemLogHook(log_base_path=str(tmp_path), include_task_path=False) + + hook.process_result(_result(message="first")) + hook.process_result(_result(message="second", events=["only"])) + + assert _names(tmp_path) == ["events(1).json", "result(1).json", "result.json"] + + +def test_results_for_different_tasks_do_not_get_an_integer(tmp_path): + hook = FileSystemLogHook(log_base_path=str(tmp_path)) + + hook.process_result(_result(task="TaskA")) + hook.process_result(_result(task="TaskB")) + + assert (tmp_path / "my_parent" / "task_a" / "result.json").is_file() + assert (tmp_path / "my_parent" / "task_b" / "result.json").is_file() + assert not list(tmp_path.rglob("result(1).json")) + + +def test_a_new_hook_instance_does_not_overwrite_logs_already_on_disk(tmp_path): + FileSystemLogHook(log_base_path=str(tmp_path), include_task_path=False).process_result( + _result(message="old") + ) + + FileSystemLogHook(log_base_path=str(tmp_path), include_task_path=False).process_result( + _result(message="new") + ) + + assert _read(tmp_path / "result.json")["message"] == "old" + assert _read(tmp_path / "result(1).json")["message"] == "new" + + +def test_data_model_is_logged_next_to_each_result_without_overwriting(tmp_path): + hook = FileSystemLogHook(log_base_path=str(tmp_path), include_task_path=False) + + hook.process_result(_result(), data=_Model(value="a")) + hook.process_result(_result(), data=_Model(value="b")) + + assert _read(tmp_path / "_model.json")["value"] == "a" + assert _read(tmp_path / "_model(1).json")["value"] == "b" diff --git a/test/unit/framework/test_plugin_executor.py b/test/unit/framework/test_plugin_executor.py index f5d46ba1..a4c4c77d 100644 --- a/test/unit/framework/test_plugin_executor.py +++ b/test/unit/framework/test_plugin_executor.py @@ -23,22 +23,35 @@ # SOFTWARE. # ############################################################################### +import abc +import json import logging from typing import Optional +from unittest.mock import MagicMock, Mock, patch import pytest from framework.common.shared_utils import DummyDataModel, MockConnectionManager from pydantic import BaseModel -from nodescraper.enums import ExecutionStatus +from nodescraper.base.oobsshdataplugin import OOBSSHDataPlugin +from nodescraper.connection.inband.inbandmanager import InBandConnectionManager +from nodescraper.connection.oob_ssh import OobSshConnectionManager +from nodescraper.enums import ExecutionStatus, OSFamily, SystemLocation from nodescraper.enums.eventpriority import EventPriority from nodescraper.enums.systeminteraction import SystemInteractionLevel -from nodescraper.interfaces import PluginInterface -from nodescraper.models import PluginConfig, PluginResult +from nodescraper.interfaces import ( + DataPlugin, + PluginInterface, + PluginResultCollator, + TaskResultHook, +) +from nodescraper.interfaces.datacollectortask import DataCollector +from nodescraper.models import PluginConfig, PluginResult, SystemInfo, TaskResult from nodescraper.models.postactioncondition import PostActionCondition from nodescraper.models.postactionpluginconfig import PostActionPluginConfig from nodescraper.pluginexecutor import PluginExecutor from nodescraper.pluginregistry import PluginRegistry +from nodescraper.utils import resolve_log_dir_name class DummyArgs(BaseModel): @@ -54,7 +67,6 @@ class TestPluginA(PluginInterface[MockConnectionManager, None]): ANALYZER_ARGS = DummyArgs(foo="initial") collection = False analysis = False - preserve_connection = False data = DummyDataModel(some_version="1") max_event_priority_level = EventPriority.INFO system_interaction_level = SystemInteractionLevel.PASSIVE @@ -274,7 +286,6 @@ def test_apply_global_args_to_plugin(): global_args = { "collection": True, "analysis": True, - "preserve_connection": True, "data": {"some_version": "1"}, "max_event_priority_level": 4, "system_interaction_level": "INTERACTIVE", @@ -287,7 +298,6 @@ def test_apply_global_args_to_plugin(): assert run_payload["collection"] is True assert run_payload["analysis"] is True - assert run_payload["preserve_connection"] is True assert run_payload["data"]["some_version"] == "1" assert run_payload["max_event_priority_level"] == 4 assert run_payload["system_interaction_level"] == "INTERACTIVE" @@ -601,7 +611,7 @@ def test_closing_connections_logged_after_post_actions(plugin_registry, caplog): def test_discover_os_info_detects_linux(system_info): """discover_os_info() should detect Linux OS when uname succeeds.""" - from unittest.mock import MagicMock, Mock + from unittest.mock import MagicMock from nodescraper.connection.inband import CommandArtifact from nodescraper.connection.inband.inbandmanager import InBandConnectionManager @@ -654,7 +664,7 @@ def test_discover_os_info_detects_linux(system_info): def test_discover_os_info_detects_arista_eos(system_info): """discover_os_info() should detect Arista EOS when uname fails but Arista command succeeds.""" import json - from unittest.mock import MagicMock, Mock + from unittest.mock import MagicMock from nodescraper.connection.inband import CommandArtifact from nodescraper.connection.inband.inbandmanager import InBandConnectionManager @@ -717,7 +727,7 @@ def test_discover_os_info_detects_arista_eos(system_info): def test_discover_os_info_skips_when_connection_fails(system_info, caplog): """discover_os_info() should skip detection and log when InBandConnectionManager fails to connect.""" - from unittest.mock import MagicMock, Mock + from unittest.mock import MagicMock from nodescraper.connection.inband.inbandmanager import InBandConnectionManager from nodescraper.enums import OSFamily @@ -757,3 +767,667 @@ def test_discover_os_info_skips_when_connection_fails(system_info, caplog): # Verify OS detection was skipped assert system_info.os_family == OSFamily.LINUX # Default from fixture assert any("Skipping OS discovery" in record.message for record in caplog.records) + + +class RaisingPlugin(PluginInterface[MockConnectionManager, None]): + __test__ = False # Tells pytest to ignore this class + + CONNECTION_TYPE = MockConnectionManager + + def run(self, **kwargs): + raise RuntimeError("plugin boom") + + +class FailingInitConnectionManager(MockConnectionManager): + def __init__(self, *args, **kwargs): + raise ValueError("bad connection args") + + +class FailingInitPlugin(PluginInterface[FailingInitConnectionManager, None]): + __test__ = False # Tells pytest to ignore this class + + CONNECTION_TYPE = FailingInitConnectionManager + + def run(self, **kwargs): + return PluginResult(source="FailingInitPlugin", status=ExecutionStatus.OK) + + +class RaisingCollator(PluginResultCollator): + def collate_results(self, plugin_results, connection_results, **kwargs): + raise RuntimeError("collator boom") + + +def _isolated_registry(plugins: dict, managers: dict) -> PluginRegistry: + registry = PluginRegistry( + load_internal_plugins=False, + load_entry_point_plugins=False, + load_entry_point_connection_managers=False, + ) + registry.plugins = plugins + registry.connection_managers = managers + return registry + + +def _read_events(path) -> list[dict]: + with open(path, encoding="utf-8") as f: + return json.load(f) + + +def test_connection_manager_instantiation_failure_is_logged_and_queue_continues(tmp_path, caplog): + """A connection manager that cannot be built is logged and skipped, not fatal for the queue.""" + registry = _isolated_registry( + {"FailingInitPlugin": FailingInitPlugin, "TestPluginB": TestPluginB}, + { + "FailingInitConnectionManager": FailingInitConnectionManager, + "MockConnectionManager": MockConnectionManager, + }, + ) + executor = PluginExecutor( + plugin_configs=[PluginConfig(plugins={"FailingInitPlugin": {}, "TestPluginB": {}})], + plugin_registry=registry, + system_info=SystemInfo(os_family=OSFamily.LINUX), + log_path=str(tmp_path), + ) + + with caplog.at_level(logging.ERROR): + results = executor.run_queue() + + assert [r.source for r in results] == ["testB"] + assert any( + "FailingInitConnectionManager" in r.message and "bad connection args" in r.message + for r in caplog.records + ) + assert not ( + tmp_path / "connection" / resolve_log_dir_name("FailingInitConnectionManager") + ).exists() + + +def test_connections_closed_when_cleanup_step_raises(tmp_path): + """Connections are closed even if a collator raises.""" + registry = _isolated_registry( + {"RaisingPlugin": RaisingPlugin}, + {"MockConnectionManager": MockConnectionManager}, + ) + registry.result_collators = {"RaisingCollator": RaisingCollator} + executor = PluginExecutor( + plugin_configs=[ + PluginConfig(plugins={"RaisingPlugin": {}}, result_collators={"RaisingCollator": {}}) + ], + plugin_registry=registry, + system_info=SystemInfo(os_family=OSFamily.LINUX), + log_path=str(tmp_path), + ) + + with patch.object(MockConnectionManager, "disconnect") as disconnect: + with pytest.raises(RuntimeError, match="collator boom"): + executor.run_queue() + + assert disconnect.call_count == 2 + + +def test_discover_os_info_result_saved_separately_from_connection_results(tmp_path): + """The OS discovery connection result goes to os_discovery/, not to connection/.""" + executor = PluginExecutor( + plugin_configs=[PluginConfig(plugins={})], + plugin_registry=_isolated_registry({}, {}), + system_info=SystemInfo(location=SystemLocation.REMOTE), + log_path=str(tmp_path), + ) + + executor.discover_os_info() + + events = _read_events(tmp_path / "os_discovery" / "events.json") + assert events[0]["priority"] == "CRITICAL" + assert events[0]["data"]["parent"] == "os_discovery" + assert not (tmp_path / "connection").exists() + + +# --------------------------------------------------------------------------- +# Connection cleanup when exceptions are raised at different points of a run +# --------------------------------------------------------------------------- + + +class TrackedManager(MockConnectionManager): + """Starts disconnected, connects on demand and records every disconnect.""" + + calls: list = [] + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.connection = None + self.result = self._init_result() + + def connect(self): + TrackedManager.calls.append(("connect", type(self).__name__)) + self.connection = MagicMock() + self.result.status = ExecutionStatus.OK + return self.result + + def disconnect(self): + TrackedManager.calls.append(("disconnect", type(self).__name__)) + self.connection = None + + +class ManagerA(TrackedManager): + pass + + +class ManagerB(TrackedManager): + pass + + +class AbstractManager(TrackedManager): + @abc.abstractmethod + def extra(self): ... + + +def _make_plugin(manager_cls, name, outcome="ok"): + """Plugin that connects, then returns OK / ERROR or raises.""" + + class _Plugin(PluginInterface[manager_cls, None]): + CONNECTION_TYPE = manager_cls + + def run(self, **kwargs): + TrackedManager.calls.append(("run", name)) + self.connection_manager.connect() + if outcome == "raise": + raise RuntimeError(f"{name} boom") + status = ExecutionStatus.ERROR if outcome == "error" else ExecutionStatus.OK + return PluginResult(source=name, status=status) + + _Plugin.__name__ = name + return _Plugin + + +class InitFailsPlugin(PluginInterface[TrackedManager, None]): + __test__ = False + + CONNECTION_TYPE = TrackedManager + + def __init__(self, *args, **kwargs): + raise RuntimeError("init boom") + + def run(self, **kwargs): + return PluginResult(source="init_fails", status=ExecutionStatus.OK) + + +@pytest.fixture +def tracked(): + TrackedManager.calls = [] + yield TrackedManager.calls + TrackedManager.calls = [] + + +def _executor(tmp_path, plugins, managers, config_plugins, **kwargs): + config_kwargs = kwargs.pop("config_kwargs", {}) + registry = _isolated_registry(plugins, managers) + registry.result_collators = kwargs.pop("collators", {}) + return PluginExecutor( + plugin_configs=[PluginConfig(plugins=config_plugins, **config_kwargs)], + plugin_registry=registry, + system_info=kwargs.pop("system_info", SystemInfo(os_family=OSFamily.LINUX)), + log_path=str(tmp_path), + **kwargs, + ) + + +def _conn_dir(tmp_path, manager_cls): + return tmp_path / "connection" / resolve_log_dir_name(manager_cls.__name__) + + +def _disconnects(calls, name="TrackedManager"): + return [c for c in calls if c == ("disconnect", name)] + + +def _failing_and_ok_plugins(): + return ( + {"Failing": InitFailsPlugin, "Ok": _make_plugin(TrackedManager, "Ok")}, + {"TrackedManager": TrackedManager}, + {"Failing": {}, "Ok": {}}, + ) + + +@pytest.mark.parametrize("spot", ["plugin_init", "global_args", "plugin_run", "result_hook"]) +def test_exception_at_each_plugin_spot_closes_connection(tmp_path, tracked, spot): + """Whichever step of running a plugin raises, its connection is closed right away + and the next plugin still runs.""" + plugins, managers, config = _failing_and_ok_plugins() + if spot != "plugin_init": + plugins["Failing"] = _make_plugin( + TrackedManager, "Failing", "raise" if spot == "plugin_run" else "ok" + ) + + def hook(result): + if result.source == "Failing": + raise RuntimeError("hook boom") + + executor = _executor( + tmp_path, + plugins, + managers, + config, + plugin_run_result_hooks=[hook] if spot == "result_hook" else None, + ) + + if spot == "global_args": + real = PluginExecutor.apply_global_args_to_plugin + + def flaky(self, plugin_inst, plugin_class, global_args): + if plugin_class.__name__ == "Failing": + raise TypeError("bad global args") + return real(self, plugin_inst, plugin_class, global_args) + + with patch.object(PluginExecutor, "apply_global_args_to_plugin", flaky): + results = executor.run_queue() + else: + results = executor.run_queue() + + assert "Ok" in [r.source for r in results] + # closed straight after the failure (before the next plugin ran) and again during cleanup + assert len(_disconnects(tracked)) == 2 + assert tracked.index(("disconnect", "TrackedManager")) < tracked.index(("run", "Ok")) + + +def test_invalid_global_args_skips_plugin_without_connection_failure(tmp_path, tracked): + """A ValueError from global args skips the plugin: no connection failure, closed at cleanup.""" + executor = _executor( + tmp_path, + {"Ok": _make_plugin(TrackedManager, "Ok")}, + {"TrackedManager": TrackedManager}, + {"Ok": {}}, + ) + + with patch.object( + PluginExecutor, "apply_global_args_to_plugin", side_effect=ValueError("bad value") + ): + results = executor.run_queue() + + assert results == [] + assert _disconnects(tracked) == [("disconnect", "TrackedManager")] + assert not (tmp_path / "connection").exists() + + +def test_unresolvable_connection_manager_is_logged_and_queue_continues(tmp_path, tracked, caplog): + """An abstract / unregistered CONNECTION_TYPE is logged as an error and the plugin skipped.""" + abstract_plugin = _make_plugin(AbstractManager, "Abstract") + executor = _executor( + tmp_path, + {"Abstract": abstract_plugin, "Ok": _make_plugin(TrackedManager, "Ok")}, + {"TrackedManager": TrackedManager}, + {"Abstract": {}, "Ok": {}}, + ) + + with caplog.at_level(logging.ERROR): + results = executor.run_queue() + + assert [r.source for r in results] == ["Ok"] + assert any("AbstractManager" in r.message and "Abstract" in r.message for r in caplog.records) + assert not _conn_dir(tmp_path, AbstractManager).exists() + + +def test_oob_plugin_without_redfish_config_is_logged_and_queue_continues(tmp_path, tracked, caplog): + """An OOB plugin without a RedfishConnectionManager config logs an error and is skipped.""" + + class NoConfigOobPlugin(OOBSSHDataPlugin): + DATA_MODEL = DummyDataModel + + executor = _executor( + tmp_path, + {"NoConfigOobPlugin": NoConfigOobPlugin, "Ok": _make_plugin(TrackedManager, "Ok")}, + {"TrackedManager": TrackedManager}, + {"NoConfigOobPlugin": {}, "Ok": {}}, + ) + + with caplog.at_level(logging.ERROR): + results = executor.run_queue() + + assert [r.source for r in results] == ["Ok"] + assert any("RedfishConnectionManager" in r.message for r in caplog.records) + assert not _conn_dir(tmp_path, OobSshConnectionManager).exists() + + +def test_post_action_plugin_exception_closes_connection(tmp_path, tracked): + """A post-action plugin raising is handled like a primary plugin failure.""" + executor = _executor( + tmp_path, + { + "Primary": _make_plugin(TrackedManager, "Primary", "error"), + "PostAction": _make_plugin(TrackedManager, "PostAction", "raise"), + }, + {"TrackedManager": TrackedManager}, + {"Primary": {}}, + config_kwargs={ + "post_action_plugins": [ + PostActionPluginConfig( + plugin="PostAction", conditions=[PostActionCondition(status="ERROR")] + ) + ] + }, + ) + + results = executor.run_queue() + + assert [r.source for r in results] == ["Primary"] + assert len(_disconnects(tracked)) == 2 + + +def test_queue_level_exception_still_saves_and_closes_connections(tmp_path, tracked): + """An exception escaping _run_single_plugin aborts the queue but connections are cleaned up.""" + plugins = { + "First": _make_plugin(TrackedManager, "First"), + "Second": _make_plugin(TrackedManager, "Second"), + } + executor = _executor( + tmp_path, plugins, {"TrackedManager": TrackedManager}, {"First": {}, "Second": {}} + ) + real = PluginExecutor._run_single_plugin + + def flaky(self, plugin_name, *args, **kwargs): + if plugin_name == "Second": + raise RuntimeError("queue boom") + return real(self, plugin_name, *args, **kwargs) + + with patch.object(PluginExecutor, "_run_single_plugin", flaky): + results = executor.run_queue() + + assert [r.source for r in results] == ["First"] + assert ("run", "Second") not in tracked + assert _disconnects(tracked) == [("disconnect", "TrackedManager")] + with open(_conn_dir(tmp_path, TrackedManager) / "result.json", encoding="utf-8") as f: + assert json.load(f)["status"] == "OK" + + +def test_post_action_evaluation_exception_still_closes_connections(tmp_path, tracked): + """If post-action handling itself raises, connections are still saved and closed.""" + executor = _executor( + tmp_path, + {"First": _make_plugin(TrackedManager, "First")}, + {"TrackedManager": TrackedManager}, + {"First": {}}, + ) + + with patch.object(PluginExecutor, "_run_post_actions", side_effect=RuntimeError("post boom")): + with pytest.raises(RuntimeError, match="post boom"): + executor.run_queue() + + assert _disconnects(tracked) == [("disconnect", "TrackedManager")] + assert (_conn_dir(tmp_path, TrackedManager) / "result.json").is_file() + + +def test_cleanup_failure_for_one_connection_does_not_block_the_others(tmp_path, tracked): + """A manager whose disconnect raises must not stop the other managers being closed.""" + executor = _executor( + tmp_path, + { + "PluginA": _make_plugin(ManagerA, "PluginA"), + "PluginB": _make_plugin(ManagerB, "PluginB"), + }, + {"ManagerA": ManagerA, "ManagerB": ManagerB}, + {"PluginA": {}, "PluginB": {}}, + ) + + with patch.object(ManagerA, "disconnect", side_effect=OSError("cleanup boom")): + results = executor.run_queue() + + assert {r.source for r in results} == {"PluginA", "PluginB"} + assert _disconnects(tracked, "ManagerB") == [("disconnect", "ManagerB")] + with open(_conn_dir(tmp_path, ManagerB) / "result.json", encoding="utf-8") as f: + assert json.load(f)["status"] == "OK" + + +def test_failing_connection_hook_does_not_stop_cleanup(tmp_path, tracked): + """A hook that raises while saving a recorded failure must not abort cleanup.""" + + class RaisingOnFailureHook(TaskResultHook): + def process_result(self, task_result, data=None, **kwargs): + if task_result.status == ExecutionStatus.EXECUTION_FAILURE: + raise OSError("disk full") + + executor = _executor( + tmp_path, + { + "Failing": _make_plugin(ManagerA, "Failing", "raise"), + "Ok": _make_plugin(ManagerB, "Ok"), + }, + {"ManagerA": ManagerA, "ManagerB": ManagerB}, + {"Failing": {}, "Ok": {}}, + ) + executor.connection_result_hooks.append(RaisingOnFailureHook()) + + results = executor.run_queue() + + assert [r.source for r in results] == ["Ok"] + assert len(_disconnects(tracked, "ManagerA")) == 2 + assert len(_disconnects(tracked, "ManagerB")) == 1 + + +def test_failure_close_error_does_not_mask_plugin_failure(tmp_path, tracked): + """If disconnect raises while closing after a plugin failure, the queue carries on.""" + executor = _executor( + tmp_path, + {"Failing": _make_plugin(ManagerA, "Failing", "raise"), "Ok": _make_plugin(ManagerB, "Ok")}, + {"ManagerA": ManagerA, "ManagerB": ManagerB}, + {"Failing": {}, "Ok": {}}, + ) + + with patch.object(ManagerA, "disconnect", side_effect=OSError("close boom")): + results = executor.run_queue() + + assert [r.source for r in results] == ["Ok"] + + +# --------------------------------------------------------------------------- +# OS discovery +# --------------------------------------------------------------------------- + + +def _discovery_executor(tmp_path, **system_kwargs): + return PluginExecutor( + plugin_configs=[PluginConfig(plugins={})], + plugin_registry=_isolated_registry({}, {}), + system_info=SystemInfo(**system_kwargs), + log_path=str(tmp_path), + ) + + +def test_discover_os_info_success_saves_result_in_os_discovery_dir(tmp_path): + executor = _discovery_executor(tmp_path) + + with ( + patch("nodescraper.pluginexecutor.discover_and_write_os_family") as detect, + patch.object(InBandConnectionManager, "disconnect", autospec=True) as disconnect, + ): + executor.discover_os_info() + + detect.assert_called_once() + disconnect.assert_called_once() + with open(tmp_path / "os_discovery" / "result.json", encoding="utf-8") as f: + saved = json.load(f) + assert saved["status"] == "OK" + assert saved["parent"] == "os_discovery" + assert not (tmp_path / "os_discovery" / "events.json").exists() + assert not (tmp_path / "connection").exists() + + +@pytest.mark.parametrize( + "spot, expected_priority", + [ + ("connect_raises", "CRITICAL"), + ("no_credentials", "CRITICAL"), + # the connection itself succeeded; the detection error is only logged + ("detection_raises", None), + ], +) +def test_discover_os_info_failures_are_saved_in_os_discovery_dir_and_closed( + tmp_path, spot, expected_priority +): + system_kwargs = {"location": SystemLocation.REMOTE} if spot == "no_credentials" else {} + executor = _discovery_executor(tmp_path, **system_kwargs) + original_family = executor.system_info.os_family + + patches = [patch.object(InBandConnectionManager, "disconnect", autospec=True)] + if spot == "connect_raises": + patches.append( + patch( + "nodescraper.connection.inband.inbandmanager.LocalShell", + side_effect=OSError("no shell"), + ) + ) + elif spot == "detection_raises": + patches.append( + patch( + "nodescraper.pluginexecutor.discover_and_write_os_family", + side_effect=RuntimeError("detect boom"), + ) + ) + with patches[0] as disconnect: + for extra in patches[1:]: + extra.start() + try: + executor.discover_os_info() + finally: + for extra in patches[1:]: + extra.stop() + + disconnect.assert_called_once() + assert executor.system_info.os_family == original_family + if expected_priority is None: + assert not (tmp_path / "os_discovery" / "events.json").exists() + else: + events = _read_events(tmp_path / "os_discovery" / "events.json") + assert events[-1]["priority"] == expected_priority + assert events[-1]["data"]["parent"] == "os_discovery" + assert not (tmp_path / "connection").exists() + + +def test_discover_os_info_disconnect_error_does_not_propagate(tmp_path): + executor = _discovery_executor(tmp_path) + + with ( + patch("nodescraper.pluginexecutor.discover_and_write_os_family"), + patch.object(InBandConnectionManager, "disconnect", side_effect=OSError("close boom")), + ): + executor.discover_os_info() + + +def test_discover_os_info_constructor_failure_does_not_raise(tmp_path): + executor = _discovery_executor(tmp_path) + + with patch( + "nodescraper.pluginexecutor.InBandConnectionManager", + side_effect=ValueError("bad connection args"), + ): + executor.discover_os_info() + + assert not (tmp_path / "connection").exists() + + +def test_discovery_and_plugin_results_go_to_separate_folders(tmp_path, tracked): + """OS discovery output and plugin connection output never share a folder.""" + executor = _executor( + tmp_path, + {"Failing": _make_plugin(TrackedManager, "Failing", "raise")}, + {"TrackedManager": TrackedManager}, + {"Failing": {}}, + system_info=SystemInfo(os_family=OSFamily.UNKNOWN), + ) + + with patch("nodescraper.pluginexecutor.discover_and_write_os_family"): + executor.run_queue() + + discovery = json.loads((tmp_path / "os_discovery" / "result.json").read_text()) + assert discovery["task"] == "InBandConnectionManager" + assert discovery["status"] == "OK" + assert not (tmp_path / "os_discovery" / "events.json").exists() + + assert not (tmp_path / "connection" / "os_discovery").exists() + assert not (tmp_path / "connection" / resolve_log_dir_name("InBandConnectionManager")).exists() + + +def test_run_queue_continues_when_discovery_connection_cannot_be_built(tmp_path, tracked): + executor = _executor( + tmp_path, + {"Ok": _make_plugin(TrackedManager, "Ok")}, + {"TrackedManager": TrackedManager}, + {"Ok": {}}, + system_info=SystemInfo(os_family=OSFamily.UNKNOWN), + ) + + with patch( + "nodescraper.pluginexecutor.InBandConnectionManager", side_effect=ValueError("bad args") + ): + results = executor.run_queue() + + assert [r.source for r in results] == ["Ok"] + + +class _OkCollector(DataCollector): + DATA_MODEL = DummyDataModel + + def collect_data(self, args=None): + return TaskResult(status=ExecutionStatus.OK), DummyDataModel() + + +def _make_data_plugin(name): + class _DataPlugin(DataPlugin): + DATA_MODEL = DummyDataModel + CONNECTION_TYPE = TrackedManager + COLLECTOR = _OkCollector + + _DataPlugin.__name__ = name + return _DataPlugin + + +def test_data_plugins_share_one_connection_by_default(tmp_path, tracked): + """No argument is needed for plugins to share the executor's connection.""" + executor = _executor( + tmp_path, + {"First": _make_data_plugin("First"), "Second": _make_data_plugin("Second")}, + {"TrackedManager": TrackedManager}, + {"First": {"analysis": False}, "Second": {"analysis": False}}, + ) + + results = executor.run_queue() + + assert [r.status for r in results] == [ExecutionStatus.OK, ExecutionStatus.OK] + assert tracked == [("connect", "TrackedManager"), ("disconnect", "TrackedManager")] + + +def test_plugin_config_preserve_connection_false_is_honored(tmp_path, tracked): + """A plugin configured with preserve_connection=False closes the shared connection and the + next plugin reconnects; the executor no longer overrides the value.""" + executor = _executor( + tmp_path, + {"First": _make_data_plugin("First"), "Second": _make_data_plugin("Second")}, + {"TrackedManager": TrackedManager}, + { + "First": {"analysis": False, "preserve_connection": False}, + "Second": {"analysis": False}, + }, + ) + + results = executor.run_queue() + + assert [r.status for r in results] == [ExecutionStatus.OK, ExecutionStatus.OK] + assert tracked == [ + ("connect", "TrackedManager"), + ("disconnect", "TrackedManager"), # First closed it + ("connect", "TrackedManager"), # Second reconnected + ("disconnect", "TrackedManager"), # executor cleanup + ] + + +def test_global_args_preserve_connection_false_is_honored(tmp_path, tracked): + executor = _executor( + tmp_path, + {"First": _make_data_plugin("First"), "Second": _make_data_plugin("Second")}, + {"TrackedManager": TrackedManager}, + {"First": {"analysis": False}, "Second": {"analysis": False}}, + config_kwargs={"global_args": {"preserve_connection": False}}, + ) + + executor.run_queue() + + # each plugin closes it after collecting, then the executor's cleanup closes it once more + assert len(_disconnects(tracked)) == 3 + assert len([c for c in tracked if c[0] == "connect"]) == 2 diff --git a/test/unit/framework/test_type_utils.py b/test/unit/framework/test_type_utils.py index ad613d83..bb63b171 100644 --- a/test/unit/framework/test_type_utils.py +++ b/test/unit/framework/test_type_utils.py @@ -130,3 +130,37 @@ def test_process_type_strips_annotated_inside_optional(): assert TypeUtils.process_type(Optional[Annotated[int, "meta"]]) == [ TypeClass(type_class=int, inner_type=None) ] + + +class _FilterModel(BaseModel): + host: str = "h" + port: int = 22 + + +def test_filter_to_model_fields_drops_keys_that_are_not_model_fields(): + data = {"host": "a", "port": 2, "other": "x"} + + assert TypeUtils.filter_to_model_fields(_FilterModel, data) == {"host": "a", "port": 2} + + +def test_filter_to_model_fields_accepts_a_model_instance(): + assert TypeUtils.filter_to_model_fields(_FilterModel(), {"port": 1, "nope": 2}) == {"port": 1} + + +def test_filter_to_model_fields_does_not_modify_its_input_and_handles_empty(): + data = {"host": "a", "other": "x"} + + TypeUtils.filter_to_model_fields(_FilterModel, data) + + assert data == {"host": "a", "other": "x"} + assert TypeUtils.filter_to_model_fields(_FilterModel, {}) == {} + + +def test_filter_to_model_fields_result_validates_against_a_model_that_forbids_extra(): + from pydantic import ConfigDict + + class Strict(BaseModel): + model_config = ConfigDict(extra="forbid") + host: str + + Strict.model_validate(TypeUtils.filter_to_model_fields(Strict, {"host": "a", "other": 1}))