Merge branch 'main' into issue-427

.github/workflows/code-testing.yml

@@ -119,6 +119,23 @@ jobs:
         run: pip install tox tox-gh-actions
       - name: "Run pytest via tox for ${{ matrix.python }}"
         run: tox
+  test-python-windows:
+    name: Pytest on 3.12 for windows
+    runs-on: windows-2022
+    needs: [lint-python, type-python]
+    env:
+      # Required to prevent asyncssh to fail.
+      USERNAME: WindowsUser
+    steps:
+      - uses: actions/checkout@v4
+      - name: Setup Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: 3.12
+      - name: Install dependencies
+        run: pip install tox tox-gh-actions
+      - name: Run pytest via tox for 3.12 on Windows
+        run: tox
   test-documentation:
     name: Build offline documentation for testing
     runs-on: ubuntu-20.04

.pre-commit-config.yaml

@@ -46,7 +46,7 @@ repos:
           - '<!--|  ~|  -->'
 
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.8.1
+    rev: v0.8.3
     hooks:
     - id: ruff
       name: Run Ruff linter

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/decorators.py

@@ -17,7 +17,8 @@
 F = TypeVar("F", bound=Callable[..., Any])
 
 
-def deprecated_test(new_tests: list[str] | None = None) -> Callable[[F], F]:
+# TODO: Remove this decorator in ANTA v2.0.0 in favor of deprecated_test_class
+def deprecated_test(new_tests: list[str] | None = None) -> Callable[[F], F]:  # pragma: no cover
     """Return a decorator to log a message of WARNING severity when a test is deprecated.
 
     Parameters
@@ -62,6 +63,57 @@ async def wrapper(*args: Any, **kwargs: Any) -> Any:
     return decorator
 
 
+def deprecated_test_class(new_tests: list[str] | None = None, removal_in_version: str | None = None) -> Callable[[type[AntaTest]], type[AntaTest]]:
+    """Return a decorator to log a message of WARNING severity when a test is deprecated.
+
+    Parameters
+    ----------
+    new_tests
+        A list of new test classes that should replace the deprecated test.
+    removal_in_version
+        A string indicating the version in which the test will be removed.
+
+    Returns
+    -------
+    Callable[[type], type]
+        A decorator that can be used to wrap test functions.
+
+    """
+
+    def decorator(cls: type[AntaTest]) -> type[AntaTest]:
+        """Actual decorator that logs the message.
+
+        Parameters
+        ----------
+        cls
+            The cls to be decorated.
+
+        Returns
+        -------
+        cls
+            The decorated cls.
+        """
+        orig_init = cls.__init__
+
+        def new_init(*args: Any, **kwargs: Any) -> None:
+            """Overload __init__ to generate a warning message for deprecation."""
+            if new_tests:
+                new_test_names = ", ".join(new_tests)
+                logger.warning("%s test is deprecated. Consider using the following new tests: %s.", cls.name, new_test_names)
+            else:
+                logger.warning("%s test is deprecated.", cls.name)
+            orig_init(*args, **kwargs)
+
+        if removal_in_version is not None:
+            cls.__removal_in_version = removal_in_version
+
+        # NOTE: we are ignoring mypy warning as we want to assign to a method here
+        cls.__init__ = new_init  # type: ignore[method-assign]
+        return cls
+
+    return decorator
+
+
 def skip_on_platforms(platforms: list[str]) -> Callable[[F], F]:
     """Return a decorator to skip a test based on the device's hardware model.
 

anta/device.py

@@ -255,7 +255,7 @@ class AsyncEOSDevice(AntaDevice):
 
     """
 
-    def __init__(
+    def __init__(  # noqa: PLR0913
         self,
         host: str,
         username: str,
@@ -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/input_models/avt.py

@@ -0,0 +1,36 @@
+# Copyright (c) 2023-2024 Arista Networks, Inc.
+# Use of this source code is governed by the Apache License 2.0
+# that can be found in the LICENSE file.
+"""Module containing input models for AVT tests."""
+
+from __future__ import annotations
+
+from ipaddress import IPv4Address
+
+from pydantic import BaseModel, ConfigDict
+
+
+class AVTPath(BaseModel):
+    """AVT (Adaptive Virtual Topology) model representing path details and associated information."""
+
+    model_config = ConfigDict(extra="forbid")
+    vrf: str = "default"
+    """VRF context. Defaults to `default`."""
+    avt_name: str
+    """The name of the Adaptive Virtual Topology (AVT)."""
+    destination: IPv4Address
+    """The IPv4 address of the destination peer in the AVT."""
+    next_hop: IPv4Address
+    """The IPv4 address of the next hop used to reach the AVT peer."""
+    path_type: str | None = None
+    """Specifies the type of path for the AVT. If not specified, both types 'direct' and 'multihop' are considered."""
+
+    def __str__(self) -> str:
+        """Return a human-readable string representation of the AVTPath for reporting.
+
+        Examples
+        --------
+        AVT CONTROL-PLANE-PROFILE VRF: default (Destination: 10.101.255.2, Next-hop: 10.101.255.1)
+
+        """
+        return f"AVT {self.avt_name} VRF: {self.vrf} (Destination: {self.destination}, Next-hop: {self.next_hop})"

anta/input_models/cvx.py

@@ -0,0 +1,19 @@
+# Copyright (c) 2023-2024 Arista Networks, Inc.
+# Use of this source code is governed by the Apache License 2.0
+# that can be found in the LICENSE file.
+"""Module containing input models for CVX tests."""
+
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import BaseModel
+
+from anta.custom_types import Hostname
+
+
+class CVXPeers(BaseModel):
+    """Model for a CVX Cluster Peer."""
+
+    peer_name: Hostname
+    registration_state: Literal["Connecting", "Connected", "Registration error", "Registration complete", "Unexpected peer state"] = "Registration complete"

anta/input_models/interfaces.py

@@ -7,17 +7,42 @@
 
 from typing import Literal
 
-from pydantic import BaseModel
+from pydantic import BaseModel, ConfigDict
 
-from anta.custom_types import Interface
+from anta.custom_types import Interface, PortChannelInterface
 
 
 class InterfaceState(BaseModel):
     """Model for an interface state."""
 
+    model_config = ConfigDict(extra="forbid")
     name: Interface
     """Interface to validate."""
-    status: Literal["up", "down", "adminDown"]
-    """Expected status of the interface."""
+    status: Literal["up", "down", "adminDown"] | None = None
+    """Expected status of the interface. Required field in the `VerifyInterfacesStatus` test."""
     line_protocol_status: Literal["up", "down", "testing", "unknown", "dormant", "notPresent", "lowerLayerDown"] | None = None
-    """Expected line protocol status of the interface."""
+    """Expected line protocol status of the interface. Optional field in the `VerifyInterfacesStatus` test."""
+    portchannel: PortChannelInterface | None = None
+    """Port-Channel in which the interface is bundled. Required field in the `VerifyLACPInterfacesStatus` test."""
+    lacp_rate_fast: bool = False
+    """Specifies the LACP timeout mode for the link aggregation group.
+
+    Options:
+    - True: Also referred to as fast mode.
+    - False: The default mode, also known as slow mode.
+
+    Can be enabled in the `VerifyLACPInterfacesStatus` tests.
+    """
+
+    def __str__(self) -> str:
+        """Return a human-readable string representation of the InterfaceState for reporting.
+
+        Examples
+        --------
+        - Interface: Ethernet1 Port-Channel: Port-Channel100
+        - Interface: Ethernet1
+        """
+        base_string = f"Interface: {self.name}"
+        if self.portchannel is not None:
+            base_string += f" Port-Channel: {self.portchannel}"
+        return base_string

anta/input_models/security.py

@@ -0,0 +1,61 @@
+# Copyright (c) 2023-2024 Arista Networks, Inc.
+# Use of this source code is governed by the Apache License 2.0
+# that can be found in the LICENSE file.
+"""Module containing input models for security tests."""
+
+from __future__ import annotations
+
+from ipaddress import IPv4Address
+from typing import Any
+from warnings import warn
+
+from pydantic import BaseModel, ConfigDict
+
+
+class IPSecPeer(BaseModel):
+    """IPSec (Internet Protocol Security) model represents the details of an IPv4 security peer."""
+
+    model_config = ConfigDict(extra="forbid")
+    peer: IPv4Address
+    """The IPv4 address of the security peer."""
+    vrf: str = "default"
+    """VRF context. Defaults to `default`."""
+    connections: list[IPSecConn] | None = None
+    """A list of IPv4 security connections associated with the peer. Defaults to None."""
+
+    def __str__(self) -> str:
+        """Return a string representation of the IPSecPeer model. Used in failure messages.
+
+        Examples
+        --------
+        - Peer: 1.1.1.1 VRF: default
+        """
+        return f"Peer: {self.peer} VRF: {self.vrf}"
+
+
+class IPSecConn(BaseModel):
+    """Details of an IPv4 security connection for a peer."""
+
+    model_config = ConfigDict(extra="forbid")
+    source_address: IPv4Address
+    """The IPv4 address of the source in the security connection."""
+    destination_address: IPv4Address
+    """The IPv4 address of the destination in the security connection."""
+
+
+class IPSecPeers(IPSecPeer):  # pragma: no cover
+    """Alias for the IPSecPeers model to maintain backward compatibility.
+
+    When initialized, it will emit a deprecation warning and call the IPSecPeer model.
+
+    TODO: Remove this class in ANTA v2.0.0.
+    """
+
+    def __init__(self, **data: Any) -> None:  # noqa: ANN401
+        """Initialize the IPSecPeer class, emitting a deprecation warning."""
+        warn(
+            message="IPSecPeers model is deprecated and will be removed in ANTA v2.0.0. Use the IPSecPeer model instead.",
+            category=DeprecationWarning,
+            stacklevel=2,
+        )
+        super().__init__(**data)

anta/input_models/stun.py

@@ -0,0 +1,35 @@
+# Copyright (c) 2023-2024 Arista Networks, Inc.
+# Use of this source code is governed by the Apache License 2.0
+# that can be found in the LICENSE file.
+"""Module containing input models for services tests."""
+
+from __future__ import annotations
+
+from ipaddress import IPv4Address
+
+from pydantic import BaseModel, ConfigDict
+
+from anta.custom_types import Port
+
+
+class StunClientTranslation(BaseModel):
+    """STUN (Session Traversal Utilities for NAT) model represents the configuration of an IPv4-based client translations."""
+
+    model_config = ConfigDict(extra="forbid")
+    source_address: IPv4Address
+    """The IPv4 address of the STUN client"""
+    source_port: Port = 4500
+    """The port number used by the STUN client for communication. Defaults to 4500."""
+    public_address: IPv4Address | None = None
+    """The public-facing IPv4 address of the STUN client, discovered via the STUN server."""
+    public_port: Port | None = None
+    """The public-facing port number of the STUN client, discovered via the STUN server."""
+
+    def __str__(self) -> str:
+        """Return a human-readable string representation of the StunClientTranslation for reporting.
+
+        Examples
+        --------
+        Client 10.0.0.1 Port: 4500
+        """
+        return f"Client {self.source_address} Port: {self.source_port}"