Repository: aristanetworks/anta (python)

Issue:
VerifyEVPNType2Route test fails for locally originated MAC/IP addresses
The 'VerifyEVPNType2Route' test currently marks endpoints with locally originated MAC/IP addresses as "FAILURE.

This behavior is unexpected because locally originated entries:
1. Are valid endpoints that can exist in the database of a VTEP.
2. Can still be advertised correctly to other VTEPs.

Marking these as "FAILURE" results in misleading test outcomes, especially when specific endpoints are explicitly validated and when the 'leaf' tag is applied to include all leaf devices in test.

Output of entry that contain locally originated mac-ip:

```VerifyEVPNType2Route is FAILURE for Leaf-01 with Message: The following EVPN Type-2 routes do not have at least one valid and active path: ['RD: 1.2.12.2:91 mac-ip 507c.6f4a.d5c6']```

Pull request:
fix(anta.tests): VerifyEVPNType2Route test fails for locally originated MAC/IP addresses
# Description

Updated the test to at least one EVPN route has at least one active/valid path across all learned routes from all RDs combined

Fixes #950 

# Checklist:

- [X] My code follows the style guidelines of this project
- [X] I have performed a self-review of my code
- [X] I have run pre-commit for code linting and typing (`pre-commit run`)
- [x] I have made corresponding changes to the documentation
- [x] I have added tests that prove my fix is effective or that my feature works
- [X] New and existing unit tests pass locally with my changes (`tox -e testenv`)


Patch:
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index 5b7a0a1e..8716d34d 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -46,7 +46,7 @@ repos:
           - '<!--|  ~|  -->'
 
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.8.2
+    rev: v0.8.3
     hooks:
     - id: ruff
       name: Run Ruff linter
diff --git a/anta/constants.py b/anta/constants.py
index 175a4adc..4dcef305 100644
--- a/anta/constants.py
+++ b/anta/constants.py
@@ -17,3 +17,12 @@ MD_REPORT_TOC = """**Table of Contents:**
     - [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 d7d2b0de..894569e7 100644
--- a/anta/device.py
+++ b/anta/device.py
@@ -372,7 +372,7 @@ class AsyncEOSDevice(AntaDevice):
         """
         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 @@ class AsyncEOSDevice(AntaDevice):
             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 @@ class AsyncEOSDevice(AntaDevice):
             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 69f305e2..c69f78e1 100644
--- a/anta/models.py
+++ b/anta/models.py
@@ -16,6 +16,7 @@ from typing import TYPE_CHECKING, Any, Callable, ClassVar, Literal, TypeVar
 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 @@ class AntaCommand(BaseModel):
 
     @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 @@ class AntaCommand(BaseModel):
         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 @@ class AntaTest(ABC):
                     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 @@ class AntaTest(ABC):
 
         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 21241876..d6f2bd47 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",


Test patch:
diff --git a/anta/tests/routing/bgp.py b/anta/tests/routing/bgp.py
index 3f691ae3..f44729ec 100644
--- a/anta/tests/routing/bgp.py
+++ b/anta/tests/routing/bgp.py
@@ -877,16 +877,17 @@ class VerifyEVPNType2Route(AntaTest):
             if not evpn_routes:
                 no_evpn_routes.append((address, vni))
                 continue
-            # Verify that each EVPN route has at least one valid and active path
-            for route, route_data in evpn_routes.items():
-                has_active_path = False
-                for path in route_data["evpnRoutePaths"]:
-                    if path["routeType"]["valid"] is True and path["routeType"]["active"] is True:
-                        # At least one path is valid and active, no need to check the other paths
+
+            # Verify that at least one EVPN route has at least one active/valid path across all learned routes from all RDs combined
+            has_active_path = False
+            for route_data in evpn_routes.values():
+                for path in route_data.get("evpnRoutePaths", []):
+                    route_type = path.get("routeType", {})
+                    if route_type.get("active") and route_type.get("valid"):
                         has_active_path = True
                         break
-                if not has_active_path:
-                    bad_evpn_routes.append(route)
+            if not has_active_path:
+                bad_evpn_routes.extend(list(evpn_routes))
 
         if no_evpn_routes:
             self.result.is_failure(f"The following VXLAN endpoint do not have any EVPN Type-2 route: {no_evpn_routes}")
diff --git a/tests/units/anta_tests/routing/test_bgp.py b/tests/units/anta_tests/routing/test_bgp.py
index 5ce75b43..2ae5e8e1 100644
--- a/tests/units/anta_tests/routing/test_bgp.py
+++ b/tests/units/anta_tests/routing/test_bgp.py
@@ -2815,8 +2815,8 @@ DATA: list[dict[str, Any]] = [
                         "evpnRoutePaths": [
                             {
                                 "routeType": {
-                                    "active": True,
-                                    "valid": True,
+                                    "active": False,
+                                    "valid": False,
                                 },
                             },
                         ]
@@ -3039,7 +3039,7 @@ DATA: list[dict[str, Any]] = [
         },
     },
     {
-        "name": "failure-multiple-routes-multiple-paths-not-active",
+        "name": "success-multiple-path-and-have-one-active/valid",
         "test": VerifyEVPNType2Route,
         "eos_data": [
             {
@@ -3084,10 +3084,7 @@ DATA: list[dict[str, Any]] = [
         ],
         "inputs": {"vxlan_endpoints": [{"address": "192.168.20.102", "vni": 10020}]},
         "expected": {
-            "result": "failure",
-            "messages": [
-                "The following EVPN Type-2 routes do not have at least one valid and active path: ['RD: 10.1.0.6:500 mac-ip 10020 aac1.ab4e.bec2 192.168.20.102']"
-            ],
+            "result": "success",
         },
     },
     {
diff --git a/tests/units/test_device.py b/tests/units/test_device.py
index faf61448..17669df2 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 @@ if TYPE_CHECKING:
     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 @@ EQUALITY_PARAMS: list[ParameterSet] = [
         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 @@ ASYNCEAPI_COLLECT_PARAMS: list[ParameterSet] = [
             },
         },
         {"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 @@ ASYNCEAPI_COLLECT_PARAMS: list[ParameterSet] = [
         {"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 @@ class TestAntaDevice:
 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 8b7c50f1..d12d8594 100644
--- a/tests/units/test_models.py
+++ b/tests/units/test_models.py
@@ -64,6 +64,23 @@ class FakeTestWithUnsupportedCommand(AntaTest):
         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 @@ ANTATEST_DATA: list[dict[str, Any]] = [
             },
         },
     },
+    {
+        "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 @@ class TestAntaTest:
         assert test.result.custom_field == "a custom field"
 
 
-class TestAntaComamnd:
+class TestAntaCommand:
     """Test for anta.models.AntaCommand."""
 
     # ruff: noqa: B018
@@ -672,6 +701,32 @@ class TestAntaComamnd:
         )
         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