feat(anta.tests): Add known EOS errors as failure (#959)

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."""

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.
 

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,18 +241,37 @@ 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.
         """
         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."""

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",

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: