diff --git a/nodescraper/connection/inband/inbandmanager.py b/nodescraper/connection/inband/inbandmanager.py index a934f5b3..5547e270 100644 --- a/nodescraper/connection/inband/inbandmanager.py +++ b/nodescraper/connection/inband/inbandmanager.py @@ -26,13 +26,12 @@ from __future__ import annotations from logging import Logger -from typing import Optional, Union +from typing import Any, Optional, Union from nodescraper.enums import ( EventCategory, EventPriority, ExecutionStatus, - OSFamily, SystemLocation, ) from nodescraper.interfaces.connectionmanager import ConnectionManager @@ -43,12 +42,10 @@ from .inband import InBandConnection from .inbandlocal import LocalShell from .inbandremote import RemoteShell, SSHConnectionError -from .osdetection import NetworkOsDetection, detect_network_os from .sshparams import SSHConnectionParams class InBandConnectionManager(ConnectionManager[InBandConnection, SSHConnectionParams]): - def __init__( self, system_info: SystemInfo, @@ -56,7 +53,7 @@ def __init__( max_event_priority_level: Union[EventPriority, str] = EventPriority.CRITICAL, parent: Optional[str] = None, task_result_hooks: Optional[list[TaskResultHook]] = None, - connection_args: Optional[SSHConnectionParams] = None, + connection_args: Optional[SSHConnectionParams | dict[str, Any]] = None, **kwargs, ): super().__init__( @@ -69,54 +66,6 @@ def __init__( **kwargs, ) - @staticmethod - def _apply_network_os_detection( - system_info: SystemInfo, - detection: NetworkOsDetection, - ) -> None: - """Apply network OS probe results to system info.""" - system_info.os_family = detection.os_family - system_info.platform = detection.platform - if system_info.metadata is None: - system_info.metadata = {} - system_info.metadata.update(detection.metadata) - - def _check_os_family(self): - """Check the OS family of the system under test (SUT) - - Raises: - RuntimeError: If the connection is not initialized - """ - if not self.connection: - raise RuntimeError("Connection not initialized") - - self.logger.info("Checking OS family") - res = self.connection.run_command("uname -s") - if "not recognized as an internal or external command" in res.stdout + res.stderr: - self.system_info.os_family = OSFamily.WINDOWS - elif res.exit_code == 0 and "VMkernel" in res.stdout: - self.system_info.os_family = OSFamily.ESXI - elif res.exit_code == 0: - self.system_info.os_family = OSFamily.LINUX - else: - detection = detect_network_os(self.connection) - if detection is not None: - self._apply_network_os_detection(self.system_info, detection) - else: - self._log_event( - category=EventCategory.UNKNOWN, - description="Unable to determine SUT OS", - priority=EventPriority.WARNING, - ) - if self.system_info.platform: - self.logger.info( - "OS Family: %s (%s)", - self.system_info.os_family.name, - self.system_info.platform, - ) - else: - self.logger.info("OS Family: %s", self.system_info.os_family.name) - def connect( self, ) -> TaskResult: @@ -128,7 +77,6 @@ def connect( if self.system_info.location == SystemLocation.LOCAL: self.logger.info("Using local shell") self.connection = LocalShell() - self._check_os_family() return self.result if not self.connection_args or not isinstance(self.connection_args, SSHConnectionParams): @@ -148,11 +96,11 @@ def connect( try: self.logger.info( - "Initializing SSH connection to system '%s'", self.connection_args.hostname + "Initializing SSH connection to system '%s'", + self.connection_args.hostname, ) self.connection = RemoteShell(self.connection_args) self.connection.connect_ssh() - self._check_os_family() except SSHConnectionError as exception: self._log_event( category=EventCategory.SSH, diff --git a/nodescraper/connection/inband/osdetection.py b/nodescraper/connection/inband/osdetection.py index 9353439d..84fbb66c 100644 --- a/nodescraper/connection/inband/osdetection.py +++ b/nodescraper/connection/inband/osdetection.py @@ -24,11 +24,14 @@ # ############################################################################### import json +import logging import re from dataclasses import dataclass from typing import Optional +from nodescraper.connection.inband import InBandConnectionManager from nodescraper.enums import OSFamily +from nodescraper.models import SystemInfo from .inband import InBandConnection @@ -150,3 +153,55 @@ def detect_network_os(connection: InBandConnection) -> Optional[NetworkOsDetecti return detection return None + + +def apply_network_os_detection( + system_info: SystemInfo, + detection: NetworkOsDetection, +) -> None: + """Apply network OS probe results to system info.""" + system_info.os_family = detection.os_family + system_info.platform = detection.platform + if system_info.metadata is None: + system_info.metadata = {} + system_info.metadata.update(detection.metadata) + + +def discover_and_write_os_family( + connection_manager: InBandConnectionManager, + system_info: SystemInfo, + logger: logging.Logger, +) -> None: + """Check + + Args: + connection_manager (InBandConnectionManager): _description_ + system_info (SystemInfo): _description_ + logger (logging.Logger): _description_ + """ + if connection_manager.connection is None: + logger.error("Connection is not initialized, OS family check cannot be performed.") + return + logger.info("Checking OS family") + res = connection_manager.connection.run_command("uname -s") + if "not recognized as an internal or external command" in res.stdout + res.stderr: + system_info.os_family = OSFamily.WINDOWS + elif res.exit_code == 0: + system_info.os_family = OSFamily.LINUX + else: + detection = detect_network_os(connection_manager.connection) + if detection is not None: + apply_network_os_detection(system_info, detection) + else: + logger.warning( + "Unable to determine OS family. uname failed and no supported network OS detected." + ) + + if system_info.platform: + logger.info( + "OS Family: %s (%s)", + system_info.os_family.name, + system_info.platform, + ) + else: + logger.info("OS Family: %s", system_info.os_family.name) diff --git a/nodescraper/interfaces/connectionmanager.py b/nodescraper/interfaces/connectionmanager.py index d413ffaf..0369b928 100644 --- a/nodescraper/interfaces/connectionmanager.py +++ b/nodescraper/interfaces/connectionmanager.py @@ -29,7 +29,7 @@ import logging import types from functools import wraps -from typing import Callable, Generic, Optional, TypeVar, Union +from typing import Any, Callable, Generic, Optional, TypeVar, Union from pydantic import BaseModel @@ -96,7 +96,7 @@ def __init__( max_event_priority_level: Union[EventPriority, str] = EventPriority.CRITICAL, parent: Optional[str] = None, task_result_hooks: Optional[list[TaskResultHook], None] = None, - connection_args: Optional[Union[TConnectArg, dict]] = None, + connection_args: Optional[Union[TConnectArg, dict[str, Any]]] = None, event_reporter: str = DEFAULT_EVENT_REPORTER, session_id: Optional[str] = None, **kwargs, diff --git a/nodescraper/interfaces/plugin.py b/nodescraper/interfaces/plugin.py index 9fe998d9..5e2b33a4 100644 --- a/nodescraper/interfaces/plugin.py +++ b/nodescraper/interfaces/plugin.py @@ -71,7 +71,7 @@ def __init__( if system_info is None: system_info = SystemInfo() - self.system_info = system_info + self.system_info: SystemInfo = system_info if not task_result_hooks: task_result_hooks = [] diff --git a/nodescraper/pluginexecutor.py b/nodescraper/pluginexecutor.py index 93421e8e..921f0fbe 100644 --- a/nodescraper/pluginexecutor.py +++ b/nodescraper/pluginexecutor.py @@ -31,13 +31,16 @@ import uuid from collections import deque from collections.abc import Callable, Sequence -from typing import Optional, Type, Union +from typing import Any, Optional, Type, Union from pydantic import BaseModel from nodescraper.base.oobsshdataplugin import OOBSSHDataPlugin +from nodescraper.connection.inband import InBandConnectionManager +from nodescraper.connection.inband.osdetection import discover_and_write_os_family from nodescraper.connection.oob_ssh import OobSshConnectionManager from nodescraper.constants import DEFAULT_LOGGER +from nodescraper.enums import ExecutionStatus, OSFamily from nodescraper.helpers.plugin_execution_target import ( format_in_band_target_summary, ) @@ -103,23 +106,7 @@ def __init__( if log_path: self.connection_result_hooks.append(FileSystemLogHook(log_base_path=log_path)) - if connections: - for connection, connection_args in connections.items(): - if connection not in self.plugin_registry.connection_managers: - self.logger.error( - "Unable to find registered connection manager class for %s", connection - ) - continue - - connection_manager = self.plugin_registry.connection_managers[connection] - - self.connection_library[connection_manager] = connection_manager( - system_info=self.system_info, - logger=self.logger, - connection_args=connection_args, - task_result_hooks=self.connection_result_hooks, - session_id=self.session_id, - ) + self._populate_connection_library(connections) self.logger.info("System Name: %s", self.system_info.name) if self.system_info.sku: @@ -131,6 +118,35 @@ def __init__( format_in_band_target_summary(self.system_info, self.connection_configs), ) + def _populate_connection_library( + self, connections: dict[str, dict[str, Any] | BaseModel] | None + ) -> None: + """Init the connection library with the provided connections. + + Args: + connections (dict[str, dict[str, Any]]): A dictionary mapping connection names to their arguments. + where the first level of the dict is always a name of the connection class and then its arguments. + """ + if connections is None: + return + for connection, connection_args in connections.items(): + if connection not in self.plugin_registry.connection_managers: + self.logger.error( + "Unable to find registered connection manager class for %s", + connection, + ) + continue + + connection_manager = self.plugin_registry.connection_managers[connection] + + self.connection_library[connection_manager] = connection_manager( + system_info=self.system_info, + logger=self.logger, + connection_args=connection_args, + task_result_hooks=self.connection_result_hooks, + session_id=self.session_id, + ) + @staticmethod def _deep_merge_plugin_args(existing: dict, incoming: dict) -> dict: """Merge incoming plugin args into existing; do not let empty dicts overwrite @@ -168,6 +184,37 @@ 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. + """ + inband_connection = self.connection_library.get(InBandConnectionManager) + if inband_connection is None: + # Init use and discard after + 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, + session_id=self.session_id, + ) + try: + result = inband_connection.connect() if inband_connection else None + if (not inband_connection) or (not result) or (result.status != ExecutionStatus.OK): + self.logger.info( + "InBandConnectionManager not available or failed to connect for OS discovery. Skipping OS discovery." + ) + return + 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), + ) + finally: + if inband_connection is not None: + inband_connection.disconnect() + def _get_connection_manager_for_plugin( self, plugin_class: type, @@ -313,6 +360,9 @@ def run_queue(self) -> list[PluginResult]: list[PluginResult]: List of results from running the plugins in the queue """ plugin_results = [] + # For Plugins discover OS Family + if self.system_info.os_family is None or self.system_info.os_family == OSFamily.UNKNOWN: + self.discover_os_info() plugin_queue = deque(self.plugin_config.plugins.items()) try: while len(plugin_queue) > 0: @@ -334,11 +384,15 @@ def run_queue(self) -> list[PluginResult]: if self.plugin_config.result_collators: self.logger.info("Running result collators") - for collator, collator_args in self.plugin_config.result_collators.items(): + 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 + "No result collator found in registry for name: %s", + collator, ) continue diff --git a/nodescraper/pluginregistry.py b/nodescraper/pluginregistry.py index cca5bbf2..537f470c 100644 --- a/nodescraper/pluginregistry.py +++ b/nodescraper/pluginregistry.py @@ -274,10 +274,8 @@ def _load_plugins_uncached() -> dict[str, type]: """Internal: Load plugins without caching logic.""" plugins = {} eps: Iterable = PluginRegistry.load_entry_points(ENTRY_POINT_PLUGINS) - for entry_point in eps: plugin_class = entry_point.load() # type: ignore[attr-defined, union-attr] - if not PluginRegistry._valid_sub_class_check( in_cls=plugin_class, base_class=PluginInterface ): diff --git a/pyproject.toml b/pyproject.toml index a60b7846..99c403c8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -91,3 +91,20 @@ explicit_package_bases = true [tool.setuptools_scm] version_scheme = "post-release" + +[dependency-groups] +dev = [ + "build", + "black", + "pylint", + "coverage", + "twine", + "ruff", + "pre-commit", + "pytest", + "pytest-cov", + "mypy", + "types-paramiko", + "types-requests", + "types-setuptools", +] diff --git a/test/functional/test_os_plugin.py b/test/functional/test_os_plugin.py new file mode 100644 index 00000000..461edc56 --- /dev/null +++ b/test/functional/test_os_plugin.py @@ -0,0 +1,186 @@ +############################################################################### +# +# 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. +# +############################################################################### +"""Functional tests for OsPlugin with --plugin-configs.""" + +from pathlib import Path +from typing import Any, Union + +import pytest + + +@pytest.fixture +def fixtures_dir(): + """Return path to fixtures directory.""" + return Path(__file__).parent / "fixtures" + + +@pytest.fixture +def os_config_file(fixtures_dir): + """Return path to OsPlugin config file.""" + return fixtures_dir / "os_plugin_config.json" + + +@pytest.fixture +def plugin_executor(plugin_reg: Union[Any, None] = None): + """Fixture that creates a PluginExecutor instance for programmatic testing.""" + from nodescraper.models.pluginconfig import PluginConfig + from nodescraper.pluginexecutor import PluginExecutor + from nodescraper.pluginregistry import PluginRegistry + + if plugin_reg is None: + plugin_reg = PluginRegistry() + + def _create_executor( + plugin_configs: list[PluginConfig], + connections=None, + log_path=None, + system_info=None, + ): + """Create a PluginExecutor with the given configuration. + + Args: + plugin_configs: List of PluginConfig objects + connections: Optional dict of connection configs + log_path: Optional path for logs + + Returns: + PluginExecutor instance + """ + return PluginExecutor( + plugin_configs=plugin_configs, + connections=connections, + system_info=system_info, + log_path=log_path, + plugin_registry=plugin_reg, + ) + + return _create_executor + + +def test_os_plugin_with_basic_config(run_cli_command, os_config_file, tmp_path): + """Test OsPlugin using basic config file.""" + assert os_config_file.exists(), f"Config file not found: {os_config_file}" + + log_path = str(tmp_path / "logs_os_basic") + result = run_cli_command( + ["--log-path", log_path, f"--plugin-configs={os_config_file}"], check=False + ) + + assert ( + result.returncode == 1 + ), f"Expected success (1), got {result.returncode}. Output: {result.stdout + result.stderr}" + + output = result.stdout + result.stderr + assert len(output) > 0 + assert "osplugin" + + +def test_os_plugin_with_os_family_discovery(run_cli_command, tmp_path): + """Test OsPlugin with OS family auto-discovery (not explicitly set).""" + log_path = str(tmp_path / "logs_os_family_discovery") + # Don't specify OS family - let it be discovered + result = run_cli_command( + [ + "--log-path", + log_path, + "run-plugins", + "OsPlugin", + ], + check=False, + ) + + # Should succeed - OS family discovery is required and should always work locally + assert result.returncode in [0, 1, 2, 3, 4] + output = result.stdout + result.stderr + assert len(output) > 0 + # Verify that collection happened (OS family was discovered) + assert "osplugin" in output.lower() or "os" in output.lower() + + +def test_os_plugin_with_executor_programmatic(plugin_executor, tmp_path): + """Test OsPlugin using PluginExecutor The executor no CLI then we will inspect the result.""" + import string + + from nodescraper.models.pluginconfig import PluginConfig + + # Exp os's are every letter and number combo + exp_os = [] + for letter in string.ascii_letters + string.digits: + exp_os.append(letter) + + config = PluginConfig( + global_args={}, + plugins={ + "OsPlugin": { + "analysis_args": {"exp_os": exp_os, "exact_match": False}, + "analysis": True, + "collection": True, + } + }, + name="Test Run", + desc="Test description", + ) + + # Create executor + executor = plugin_executor( + plugin_configs=[config], + log_path=str(tmp_path / "logs_executor"), + ) + + # Run plugins (correct method is run_queue) + results = executor.run_queue() + + # Verify execution succeeded + assert results is not None + assert len(results) > 0 + + # Check that OsPlugin ran + os_results = [r for r in results if r.source == "OsPlugin"] + assert len(os_results) > 0, "OsPlugin did not run" + + # Check the OsPlugin result + from nodescraper.enums import ExecutionStatus + + os_result = os_results[0] + assert os_result.result_data is not None, "OsPlugin result_data is None" + assert ( + os_result.status == ExecutionStatus.OK + ), f"OsPlugin status: {os_result.status}, message: {os_result.message}" + + # Verify collection and analysis results + assert os_result.result_data.collection_result is not None, "Collection result is None" + assert ( + os_result.result_data.collection_result.status == ExecutionStatus.OK + ), f"Collection status: {os_result.result_data.collection_result.status}" + assert os_result.result_data.analysis_result is not None, "Analysis result is None" + assert ( + os_result.result_data.analysis_result.status == ExecutionStatus.OK + ), f"Analysis status: {os_result.result_data.analysis_result.status}" + + # Verify data was collected (OS family was auto-discovered) + assert hasattr(executor, "system_info") + # OS family should be discovered and set + assert executor.system_info.os_family is not None diff --git a/test/unit/connection/test_osdetection.py b/test/unit/connection/test_osdetection.py index d9092f04..ca445317 100644 --- a/test/unit/connection/test_osdetection.py +++ b/test/unit/connection/test_osdetection.py @@ -31,6 +31,7 @@ ARISTA_VERSION_CMD, DELL_VERSION_CMD, detect_network_os, + discover_and_write_os_family, parse_arista_version_output, parse_dell_sonic_version_output, ) @@ -177,7 +178,7 @@ def test_detect_network_os_falls_back_to_dell(conn_mock): assert conn_mock.run_command.call_count == 2 -def test_check_os_family_detects_arista_eos(system_info, conn_mock): +def test_discover_and_write_os_family_detects_arista_eos(system_info, conn_mock, logger): manager = InBandConnectionManager(system_info=system_info) manager.connection = conn_mock conn_mock.run_command.side_effect = [ @@ -185,18 +186,15 @@ def test_check_os_family_detects_arista_eos(system_info, conn_mock): DUMMY_ARISTA_VERSION_CMD_OK, ] - manager._check_os_family() + discover_and_write_os_family(manager, system_info, logger) assert system_info.os_family == OSFamily.EOS assert system_info.platform == "Arista EOS" assert system_info.metadata["os_version"] == DUMMY_ARISTA_VERSION["version"] assert system_info.metadata["device_model"] == DUMMY_ARISTA_VERSION["modelName"] - assert not any( - event.description == "Unable to determine SUT OS" for event in manager.result.events - ) -def test_check_os_family_detects_dell_sonic(system_info, conn_mock): +def test_discover_and_write_os_family_detects_dell_sonic(system_info, conn_mock, logger): system_info.os_family = OSFamily.UNKNOWN manager = InBandConnectionManager(system_info=system_info) manager.connection = conn_mock @@ -206,18 +204,17 @@ def test_check_os_family_detects_dell_sonic(system_info, conn_mock): DUMMY_DELL_VERSION_CMD_OK, ] - manager._check_os_family() + discover_and_write_os_family(manager, system_info, logger) assert system_info.os_family == OSFamily.SONIC assert system_info.platform == "Dell SONiC" assert system_info.metadata["os_version"] == "4.1.0-Enterprise" assert system_info.metadata["device_model"] == "DellEMC-S5248F-ON" - assert not any( - event.description == "Unable to determine SUT OS" for event in manager.result.events - ) -def test_check_os_family_still_warns_when_unknown(system_info, conn_mock): +def test_discover_and_write_os_family_leaves_unknown_when_undetected( + system_info, conn_mock, logger +): system_info.os_family = OSFamily.UNKNOWN manager = InBandConnectionManager(system_info=system_info) manager.connection = conn_mock @@ -227,21 +224,17 @@ def test_check_os_family_still_warns_when_unknown(system_info, conn_mock): DUMMY_DELL_VERSION_CMD_NON_DELL, ] - manager._check_os_family() + discover_and_write_os_family(manager, system_info, logger) assert system_info.os_family == OSFamily.UNKNOWN - assert any( - event.description == "Unable to determine SUT OS" and event.category == "UNKNOWN" - for event in manager.result.events - ) -def test_check_os_family_linux_skips_network_probes(system_info, conn_mock): +def test_discover_and_write_os_family_linux_skips_network_probes(system_info, conn_mock, logger): manager = InBandConnectionManager(system_info=system_info) manager.connection = conn_mock conn_mock.run_command.return_value = DUMMY_UNAME_LINUX - manager._check_os_family() + discover_and_write_os_family(manager, system_info, logger) assert system_info.os_family == OSFamily.LINUX conn_mock.run_command.assert_called_once_with("uname -s") diff --git a/test/unit/framework/test_plugin_executor.py b/test/unit/framework/test_plugin_executor.py index c539386b..584c8ff6 100644 --- a/test/unit/framework/test_plugin_executor.py +++ b/test/unit/framework/test_plugin_executor.py @@ -589,3 +589,163 @@ def test_closing_connections_logged_after_post_actions(plugin_registry, caplog): assert ( post_action_idx < closing_idx ), "'Closing connections' must be logged after the post-action plugin runs" + + +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 nodescraper.connection.inband import CommandArtifact + from nodescraper.connection.inband.inbandmanager import InBandConnectionManager + from nodescraper.enums import OSFamily + from nodescraper.models.taskresult import TaskResult + + # Mock the connection to return Linux + mock_conn = MagicMock() + mock_conn.run_command.return_value = CommandArtifact( + command="uname -s", + stdout="Linux", + stderr="", + exit_code=0, + ) + + # Mock InBandConnectionManager instance + mock_manager = MagicMock(spec=InBandConnectionManager) + mock_manager.connection = mock_conn + mock_manager.connect.return_value = TaskResult(status=ExecutionStatus.OK) + mock_manager.disconnect.return_value = None + + # Create a mock class that returns our mock manager instance + mock_class = Mock(return_value=mock_manager) + mock_class.__name__ = "InBandConnectionManager" + + # Store original and patch + import nodescraper.pluginexecutor + + original = nodescraper.pluginexecutor.InBandConnectionManager + nodescraper.pluginexecutor.InBandConnectionManager = mock_class + + try: + # Create executor + executor = PluginExecutor( + plugin_configs=[PluginConfig(plugins={})], + system_info=system_info, + ) + + # Run OS discovery + executor.discover_os_info() + finally: + # Restore original + nodescraper.pluginexecutor.InBandConnectionManager = original + + # Verify Linux was detected + assert system_info.os_family == OSFamily.LINUX + mock_conn.run_command.assert_called_once_with("uname -s") + + +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 nodescraper.connection.inband import CommandArtifact + from nodescraper.connection.inband.inbandmanager import InBandConnectionManager + from nodescraper.enums import OSFamily + from nodescraper.models.taskresult import TaskResult + + arista_version = { + "mfgName": "Arista Networks", + "version": "4.32.1F", + "modelName": "DCS-7280CR3-32P4", + } + + # Mock the connection to fail uname but succeed with Arista command + mock_conn = MagicMock() + mock_conn.run_command.side_effect = [ + CommandArtifact(command="uname -s", stdout="", stderr="invalid", exit_code=1), + CommandArtifact( + command="show version | json | no-more", + stdout=json.dumps(arista_version), + stderr="", + exit_code=0, + ), + ] + + # Mock InBandConnectionManager instance + mock_manager = MagicMock(spec=InBandConnectionManager) + mock_manager.connection = mock_conn + mock_manager.connect.return_value = TaskResult(status=ExecutionStatus.OK) + mock_manager.disconnect.return_value = None + + # Create a mock class that returns our mock manager instance + mock_class = Mock(return_value=mock_manager) + mock_class.__name__ = "InBandConnectionManager" + + # Store original and patch + import nodescraper.pluginexecutor + + original = nodescraper.pluginexecutor.InBandConnectionManager + nodescraper.pluginexecutor.InBandConnectionManager = mock_class + + try: + # Create executor + executor = PluginExecutor( + plugin_configs=[PluginConfig(plugins={})], + system_info=system_info, + ) + + # Run OS discovery + executor.discover_os_info() + finally: + # Restore original + nodescraper.pluginexecutor.InBandConnectionManager = original + + # Verify Arista EOS was detected + assert system_info.os_family == OSFamily.EOS + assert system_info.platform == "Arista EOS" + assert system_info.metadata["os_version"] == "4.32.1F" + assert system_info.metadata["device_model"] == "DCS-7280CR3-32P4" + + +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 nodescraper.connection.inband.inbandmanager import InBandConnectionManager + from nodescraper.enums import OSFamily + from nodescraper.models.taskresult import TaskResult + + # Mock InBandConnectionManager instance to fail connection + mock_manager = MagicMock(spec=InBandConnectionManager) + mock_manager.connect.return_value = TaskResult( + status=ExecutionStatus.ERROR, message="Connection failed" + ) + mock_manager.disconnect.return_value = None + + # Create a mock class that returns our mock manager instance + mock_class = Mock(return_value=mock_manager) + mock_class.__name__ = "InBandConnectionManager" + + # Store original and patch + import nodescraper.pluginexecutor + + original = nodescraper.pluginexecutor.InBandConnectionManager + nodescraper.pluginexecutor.InBandConnectionManager = mock_class + + try: + # Create executor + executor = PluginExecutor( + plugin_configs=[PluginConfig(plugins={})], + system_info=system_info, + ) + + # Run OS discovery + with caplog.at_level(logging.INFO): + executor.discover_os_info() + finally: + # Restore original + nodescraper.pluginexecutor.InBandConnectionManager = original + + # 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)