diff --git a/anta/constants.py b/anta/constants.py index 175a4adcc..4dcef3050 100644 --- a/anta/constants.py +++ b/anta/constants.py @@ -17,3 +17,12 @@ - [Summary Totals Per Category](#summary-totals-per-category) - [Test Results](#test-results)""" """Table of Contents for the Markdown report.""" + +KNOWN_EOS_ERRORS = [ + r"BGP inactive", + r"VRF '.*' is not active", + r".* does not support IP", + r"IS-IS (.*) is disabled because: .*", + r"No source interface .*", +] +"""List of known EOS errors that should set a test status to 'failure' with the error message.""" diff --git a/anta/device.py b/anta/device.py index d7d2b0de2..894569e7f 100644 --- a/anta/device.py +++ b/anta/device.py @@ -372,7 +372,7 @@ def _keys(self) -> tuple[Any, ...]: """ return (self._session.host, self._session.port) - async def _collect(self, command: AntaCommand, *, collection_id: str | None = None) -> None: # noqa: C901 function is too complex - because of many required except blocks + async def _collect(self, command: AntaCommand, *, collection_id: str | None = None) -> None: """Collect device command output from EOS using aio-eapi. Supports outformat `json` and `text` as output structure. @@ -409,15 +409,7 @@ async def _collect(self, command: AntaCommand, *, collection_id: str | None = No command.output = response[-1] except asynceapi.EapiCommandError as e: # This block catches exceptions related to EOS issuing an error. - command.errors = e.errors - if command.requires_privileges: - logger.error( - "Command '%s' requires privileged mode on %s. Verify user permissions and if the `enable` option is required.", command.command, self.name - ) - if command.supported: - logger.error("Command '%s' failed on %s: %s", command.command, self.name, e.errors[0] if len(e.errors) == 1 else e.errors) - else: - logger.debug("Command '%s' is not supported on '%s' (%s)", command.command, self.name, self.hw_model) + self._log_eapi_command_error(command, e) except TimeoutException as e: # This block catches Timeout exceptions. command.errors = [exc_to_str(e)] @@ -446,6 +438,18 @@ async def _collect(self, command: AntaCommand, *, collection_id: str | None = No anta_log_exception(e, f"An error occurred while issuing an eAPI request to {self.name}", logger) logger.debug("%s: %s", self.name, command) + def _log_eapi_command_error(self, command: AntaCommand, e: asynceapi.EapiCommandError) -> None: + """Appropriately log the eapi command error.""" + command.errors = e.errors + if command.requires_privileges: + logger.error("Command '%s' requires privileged mode on %s. Verify user permissions and if the `enable` option is required.", command.command, self.name) + if not command.supported: + logger.debug("Command '%s' is not supported on '%s' (%s)", command.command, self.name, self.hw_model) + elif command.returned_known_eos_error: + logger.debug("Command '%s' returned a known error '%s': %s", command.command, self.name, command.errors) + else: + logger.error("Command '%s' failed on %s: %s", command.command, self.name, e.errors[0] if len(e.errors) == 1 else e.errors) + async def refresh(self) -> None: """Update attributes of an AsyncEOSDevice instance. diff --git a/anta/models.py b/anta/models.py index 69f305e2f..c69f78e1f 100644 --- a/anta/models.py +++ b/anta/models.py @@ -16,6 +16,7 @@ from pydantic import BaseModel, ConfigDict, ValidationError, create_model from anta import GITHUB_SUGGESTION +from anta.constants import KNOWN_EOS_ERRORS from anta.custom_types import REGEXP_EOS_BLACKLIST_CMDS, Revision from anta.logger import anta_log_exception, exc_to_str from anta.result_manager.models import AntaTestStatus, TestResult @@ -240,10 +241,29 @@ def requires_privileges(self) -> bool: @property def supported(self) -> bool: - """Return True if the command is supported on the device hardware platform, False otherwise. + """Indicates if the command is supported on the device. + + Returns + ------- + bool + True if the command is supported on the device hardware platform, False otherwise. Raises ------ + RuntimeError + If the command has not been collected and has not returned an error. + AntaDevice.collect() must be called before this property. + """ + if not self.collected and not self.error: + msg = f"Command '{self.command}' has not been collected and has not returned an error. Call AntaDevice.collect()." + + raise RuntimeError(msg) + return all("not supported on this hardware platform" not in e for e in self.errors) + + @property + def returned_known_eos_error(self) -> bool: + """Return True if the command returned a known_eos_error on the device, False otherwise. + RuntimeError If the command has not been collected and has not returned an error. AntaDevice.collect() must be called before this property. @@ -251,7 +271,7 @@ def supported(self) -> bool: if not self.collected and not self.error: msg = f"Command '{self.command}' has not been collected and has not returned an error. Call AntaDevice.collect()." raise RuntimeError(msg) - return not any("not supported on this hardware platform" in e for e in self.errors) + return any(any(re.match(pattern, e) for e in self.errors) for pattern in KNOWN_EOS_ERRORS) class AntaTemplateRenderError(RuntimeError): @@ -630,14 +650,9 @@ async def wrapper( AntaTest.update_progress() return self.result - if cmds := self.failed_commands: - unsupported_commands = [f"'{c.command}' is not supported on {self.device.hw_model}" for c in cmds if not c.supported] - if unsupported_commands: - msg = f"Test {self.name} has been skipped because it is not supported on {self.device.hw_model}: {GITHUB_SUGGESTION}" - self.logger.warning(msg) - self.result.is_skipped("\n".join(unsupported_commands)) - else: - self.result.is_error(message="\n".join([f"{c.command} has failed: {', '.join(c.errors)}" for c in cmds])) + if self.failed_commands: + self._handle_failed_commands() + AntaTest.update_progress() return self.result @@ -657,6 +672,28 @@ async def wrapper( return wrapper + def _handle_failed_commands(self) -> None: + """Handle failed commands inside a test. + + There can be 3 types: + * unsupported on hardware commands which set the test status to 'skipped' + * known EOS error which set the test status to 'failure' + * unknown failure which set the test status to 'error' + """ + cmds = self.failed_commands + unsupported_commands = [f"'{c.command}' is not supported on {self.device.hw_model}" for c in cmds if not c.supported] + if unsupported_commands: + msg = f"Test {self.name} has been skipped because it is not supported on {self.device.hw_model}: {GITHUB_SUGGESTION}" + self.logger.warning(msg) + self.result.is_skipped("\n".join(unsupported_commands)) + return + returned_known_eos_error = [f"'{c.command}' failed on {self.device.name}: {', '.join(c.errors)}" for c in cmds if c.returned_known_eos_error] + if returned_known_eos_error: + self.result.is_failure("\n".join(returned_known_eos_error)) + return + + self.result.is_error(message="\n".join([f"{c.command} has failed: {', '.join(c.errors)}" for c in cmds])) + @classmethod def update_progress(cls: type[AntaTest]) -> None: """Update progress bar for all AntaTest objects if it exists.""" diff --git a/pyproject.toml b/pyproject.toml index 212418760..d6f2bd47a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -70,7 +70,6 @@ dev = [ "pytest-cov>=4.1.0", "pytest-dependency", "pytest-codspeed>=2.2.0", - "respx", "pytest-html>=3.2.0", "pytest-httpx>=0.30.0", "pytest-metadata>=3.0.0", diff --git a/tests/units/test_device.py b/tests/units/test_device.py index faf614481..17669df2a 100644 --- a/tests/units/test_device.py +++ b/tests/units/test_device.py @@ -6,13 +6,15 @@ from __future__ import annotations import asyncio +from contextlib import AbstractContextManager +from contextlib import nullcontext as does_not_raise from pathlib import Path from typing import TYPE_CHECKING, Any from unittest.mock import patch import pytest from asyncssh import SSHClientConnection, SSHClientConnectionOptions -from httpx import ConnectError, HTTPError +from httpx import ConnectError, HTTPError, TimeoutException from rich import print as rprint from anta.device import AntaDevice, AsyncEOSDevice @@ -24,13 +26,37 @@ from _pytest.mark.structures import ParameterSet INIT_PARAMS: list[ParameterSet] = [ - pytest.param({"host": "42.42.42.42", "username": "anta", "password": "anta"}, {"name": "42.42.42.42"}, id="no name, no port"), - pytest.param({"host": "42.42.42.42", "username": "anta", "password": "anta", "port": 666}, {"name": "42.42.42.42:666"}, id="no name, port"), + pytest.param({"host": "42.42.42.42", "username": "anta", "password": "anta"}, {"name": "42.42.42.42"}, does_not_raise(), id="no name, no port"), + pytest.param({"host": "42.42.42.42", "username": "anta", "password": "anta", "port": 666}, {"name": "42.42.42.42:666"}, does_not_raise(), id="no name, port"), pytest.param( - {"host": "42.42.42.42", "username": "anta", "password": "anta", "name": "test.anta.ninja", "disable_cache": True}, {"name": "test.anta.ninja"}, id="name" + {"host": "42.42.42.42", "username": "anta", "password": "anta", "name": "test.anta.ninja", "disable_cache": True}, + {"name": "test.anta.ninja"}, + does_not_raise(), + id="name", ), pytest.param( - {"host": "42.42.42.42", "username": "anta", "password": "anta", "name": "test.anta.ninja", "insecure": True}, {"name": "test.anta.ninja"}, id="insecure" + {"host": "42.42.42.42", "username": "anta", "password": "anta", "name": "test.anta.ninja", "insecure": True}, + {"name": "test.anta.ninja"}, + does_not_raise(), + id="insecure", + ), + pytest.param( + {"host": None, "username": "anta", "password": "anta", "name": "test.anta.ninja"}, + None, + pytest.raises(ValueError, match="'host' is required to create an AsyncEOSDevice"), + id="host is None", + ), + pytest.param( + {"host": "42.42.42.42", "username": None, "password": "anta", "name": "test.anta.ninja"}, + None, + pytest.raises(ValueError, match="'username' is required to instantiate device 'test.anta.ninja'"), + id="username is None", + ), + pytest.param( + {"host": "42.42.42.42", "username": "anta", "password": None, "name": "test.anta.ninja"}, + None, + pytest.raises(ValueError, match="'password' is required to instantiate device 'test.anta.ninja'"), + id="password is None", ), ] EQUALITY_PARAMS: list[ParameterSet] = [ @@ -48,7 +74,10 @@ id="not-equal-port", ), pytest.param( - {"host": "42.42.42.41", "username": "anta", "password": "anta"}, {"host": "42.42.42.42", "username": "anta", "password": "anta"}, False, id="not-equal-host" + {"host": "42.42.42.41", "username": "anta", "password": "anta"}, + {"host": "42.42.42.42", "username": "anta", "password": "anta"}, + False, + id="not-equal-host", ), ] ASYNCEAPI_COLLECT_PARAMS: list[ParameterSet] = [ @@ -287,7 +316,58 @@ }, }, {"output": None, "errors": ["Authorization denied for command 'show version'"]}, - id="asynceapi.EapiCommandError", + id="asynceapi.EapiCommandError - Authorization denied", + ), + pytest.param( + {}, + { + "command": "show version", + "patch_kwargs": { + "side_effect": EapiCommandError( + passed=[], + failed="show version", + errors=["not supported on this hardware platform"], + errmsg="Invalid command", + not_exec=[], + ) + }, + }, + {"output": None, "errors": ["not supported on this hardware platform"]}, + id="asynceapi.EapiCommandError - not supported", + ), + pytest.param( + {}, + { + "command": "show version", + "patch_kwargs": { + "side_effect": EapiCommandError( + passed=[], + failed="show version", + errors=["BGP inactive"], + errmsg="Invalid command", + not_exec=[], + ) + }, + }, + {"output": None, "errors": ["BGP inactive"]}, + id="asynceapi.EapiCommandError - known EOS error", + ), + pytest.param( + {}, + { + "command": "show version", + "patch_kwargs": { + "side_effect": EapiCommandError( + passed=[], + failed="show version", + errors=["Invalid input (privileged mode required)"], + errmsg="Invalid command", + not_exec=[], + ) + }, + }, + {"output": None, "errors": ["Invalid input (privileged mode required)"]}, + id="asynceapi.EapiCommandError - requires privileges", ), pytest.param( {}, @@ -301,6 +381,12 @@ {"output": None, "errors": ["ConnectError: Cannot open port"]}, id="httpx.ConnectError", ), + pytest.param( + {}, + {"command": "show version", "patch_kwargs": {"side_effect": TimeoutException("Test")}}, + {"output": None, "errors": ["TimeoutException: Test"]}, + id="httpx.TimeoutException", + ), ] ASYNCEAPI_COPY_PARAMS: list[ParameterSet] = [ pytest.param({}, {"sources": [Path("/mnt/flash"), Path("/var/log/agents")], "destination": Path(), "direction": "from"}, id="from"), @@ -531,22 +617,24 @@ def test_cache_statistics(self, device: AntaDevice, expected: dict[str, Any] | N class TestAsyncEOSDevice: """Test for anta.device.AsyncEOSDevice.""" - @pytest.mark.parametrize(("device", "expected"), INIT_PARAMS) - def test__init__(self, device: dict[str, Any], expected: dict[str, Any]) -> None: + @pytest.mark.parametrize(("device", "expected", "expected_raise"), INIT_PARAMS) + def test__init__(self, device: dict[str, Any], expected: dict[str, Any] | None, expected_raise: AbstractContextManager[Exception]) -> None: """Test the AsyncEOSDevice constructor.""" - dev = AsyncEOSDevice(**device) + with expected_raise: + dev = AsyncEOSDevice(**device) - assert dev.name == expected["name"] - if device.get("disable_cache") is True: - assert dev.cache is None - assert dev.cache_locks is None - else: # False or None - assert dev.cache is not None - assert dev.cache_locks is not None - hash(dev) + assert expected is not None + assert dev.name == expected["name"] + if device.get("disable_cache") is True: + assert dev.cache is None + assert dev.cache_locks is None + else: # False or None + assert dev.cache is not None + assert dev.cache_locks is not None + hash(dev) - with patch("anta.device.__DEBUG__", new=True): - rprint(dev) + with patch("anta.device.__DEBUG__", new=True): + rprint(dev) @pytest.mark.parametrize(("device1", "device2", "expected"), EQUALITY_PARAMS) def test__eq(self, device1: dict[str, Any], device2: dict[str, Any], expected: bool) -> None: diff --git a/tests/units/test_models.py b/tests/units/test_models.py index 8b7c50f10..d12d85941 100644 --- a/tests/units/test_models.py +++ b/tests/units/test_models.py @@ -64,6 +64,23 @@ def test(self) -> None: self.result.is_success() +class FakeTestWithKnownEOSError(AntaTest): + """ANTA test triggering a known EOS Error that should translate to failure of the test.""" + + categories: ClassVar[list[str]] = [] + commands: ClassVar[list[AntaCommand | AntaTemplate]] = [ + AntaCommand( + command="show bgp evpn route-type mac-ip aa:c1:ab:de:50:ad vni 10010", + errors=["BGP inactive"], + ) + ] + + @AntaTest.anta_test + def test(self) -> None: + """Test function.""" + self.result.is_success() + + class FakeTestWithInput(AntaTest): """ANTA test with inputs that always succeed.""" @@ -484,6 +501,18 @@ class FakeTestWithMissingTest(AntaTest): }, }, }, + { + "name": "known EOS error command", + "test": FakeTestWithKnownEOSError, + "inputs": None, + "expected": { + "__init__": {"result": "unset"}, + "test": { + "result": "failure", + "messages": ["BGP inactive"], + }, + }, + }, ] BLACKLIST_COMMANDS_PARAMS = ["reload", "reload --force", "write", "wr mem"] @@ -613,7 +642,7 @@ def test_result_overwrite(self, device: AntaDevice) -> None: assert test.result.custom_field == "a custom field" -class TestAntaComamnd: +class TestAntaCommand: """Test for anta.models.AntaCommand.""" # ruff: noqa: B018 @@ -672,6 +701,32 @@ def test_requires_privileges(self) -> None: ) assert command.requires_privileges is False command = AntaCommand(command="show aaa methods accounting") - with pytest.raises(RuntimeError) as exec_info: + with pytest.raises( + RuntimeError, match="Command 'show aaa methods accounting' has not been collected and has not returned an error. Call AntaDevice.collect()." + ): command.requires_privileges - assert exec_info.value.args[0] == "Command 'show aaa methods accounting' has not been collected and has not returned an error. Call AntaDevice.collect()." + + @pytest.mark.parametrize( + ("command_str", "error", "is_known"), + [ + ("show ip interface Ethernet1", "Ethernet1 does not support IP", True), + ("ping vrf MGMT 1.1.1.1 source Management0 size 100 df-bit repeat 2", "VRF 'MGMT' is not active", True), + ("ping vrf MGMT 1.1.1.1 source Management1 size 100 df-bit repeat 2", "No source interface Management1", True), + ("show bgp evpn route-type mac-ip aa:c1:ab:de:50:ad vni 10010", "BGP inactive", True), + ("show isis BLAH neighbors", "IS-IS (BLAH) is disabled because: IS-IS Network Entity Title (NET) configuration is not present", True), + ("show ip interface Ethernet1", None, False), + ], + ) + def test_returned_known_eos_error(self, command_str: str, error: str | None, is_known: bool) -> None: + """Test the returned_known_eos_error property.""" + # Adding fake output when no error is present to mimic that the command has been collected + command = AntaCommand(command=command_str, errors=[error] if error else [], output=None if error else "{}") + assert command.returned_known_eos_error is is_known + + def test_returned_known_eos_error_failure(self) -> None: + """Test the returned_known_eos_error property unset.""" + command = AntaCommand(command="show ip interface Ethernet1") + with pytest.raises( + RuntimeError, match="Command 'show ip interface Ethernet1' has not been collected and has not returned an error. Call AntaDevice.collect()." + ): + command.returned_known_eos_error