Merge branch 'main' into issue_822

.github/workflows/code-testing.yml

@@ -46,7 +46,7 @@ jobs:
     runs-on: ubuntu-20.04
     strategy:
       matrix:
-        python-version: ["3.9", "3.10", "3.11", "3.12"]
+        python-version: ["3.9", "3.10", "3.11", "3.12", "3.13"]
     needs: file-changes
     steps:
       - uses: actions/checkout@v4
@@ -108,7 +108,7 @@ jobs:
     needs: [lint-python, type-python]
     strategy:
       matrix:
-        python: ["3.9", "3.10", "3.11", "3.12"]
+        python: ["3.9", "3.10", "3.11", "3.12", "3.13"]
     steps:
       - uses: actions/checkout@v4
       - name: Setup Python
@@ -149,4 +149,4 @@ jobs:
         uses: CodSpeedHQ/action@v3
         with:
           token: ${{ secrets.CODSPEED_TOKEN }}
-          run: pytest --codspeed --no-cov --log-cli-level INFO tests/benchmark
+          run: pytest --codspeed --no-cov --log-cli-level INFO tests/benchmark

.pre-commit-config.yaml

@@ -46,7 +46,7 @@ repos:
           - '<!--|  ~|  -->'
 
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.6.9
+    rev: v0.8.0
     hooks:
     - id: ruff
       name: Run Ruff linter
@@ -85,7 +85,7 @@ repos:
         types: [text]
 
   - repo: https://github.com/pre-commit/mirrors-mypy
-    rev: v1.11.2
+    rev: v1.13.0
     hooks:
       - id: mypy
         name: Check typing with mypy
@@ -100,7 +100,7 @@ repos:
         files: ^(anta|tests)/
 
   - repo: https://github.com/igorshubovych/markdownlint-cli
-    rev: v0.42.0
+    rev: v0.43.0
     hooks:
       - id: markdownlint
         name: Check Markdown files style.

anta/cli/__init__.py

@@ -35,7 +35,7 @@ def wrap() -> None:
 
     cli = build_cli(exc)
 
-__all__ = ["cli", "anta"]
+__all__ = ["anta", "cli"]
 
 if __name__ == "__main__":
     cli()

anta/cli/exec/commands.py

@@ -84,7 +84,10 @@ def snapshot(inventory: AntaInventory, tags: set[str] | None, commands_list: Pat
 )
 @click.option(
     "--configure",
-    help="Ensure devices have 'aaa authorization exec default local' configured (required for SCP on EOS). THIS WILL CHANGE THE CONFIGURATION OF YOUR NETWORK.",
+    help=(
+        "[DEPRECATED] Ensure devices have 'aaa authorization exec default local' configured (required for SCP on EOS). "
+        "THIS WILL CHANGE THE CONFIGURATION OF YOUR NETWORK."
+    ),
     default=False,
     is_flag=True,
     show_default=True,

anta/cli/exec/utils.py

@@ -128,6 +128,13 @@ async def collect(device: AntaDevice) -> None:
                     logger.error("Unable to collect tech-support on %s: configuration 'aaa authorization exec default local' is not present", device.name)
                     return
 
+                # TODO: ANTA 2.0.0
+                msg = (
+                    "[DEPRECATED] Using '--configure' for collecting show-techs is deprecated and will be removed in ANTA 2.0.0. "
+                    "Please add the required configuration on your devices before running this command from ANTA."
+                )
+                logger.warning(msg)
+
                 commands = []
                 # TODO: @mtache - add `config` field to `AntaCommand` object to handle this use case.
                 # Otherwise mypy complains about enable as it is only implemented for AsyncEOSDevice

anta/cli/nrfu/utils.py

@@ -116,8 +116,12 @@ def print_text(ctx: click.Context) -> None:
     """Print results as simple text."""
     console.print()
     for test in _get_result_manager(ctx).results:
-        message = f" ({test.messages[0]!s})" if len(test.messages) > 0 else ""
-        console.print(f"{test.name} :: {test.test} :: [{test.result}]{test.result.upper()}[/{test.result}]{message}", highlight=False)
+        if len(test.messages) <= 1:
+            message = test.messages[0] if len(test.messages) == 1 else ""
+            console.print(f"{test.name} :: {test.test} :: [{test.result}]{test.result.upper()}[/{test.result}]({message})", highlight=False)
+        else:  # len(test.messages) > 1
+            console.print(f"{test.name} :: {test.test} :: [{test.result}]{test.result.upper()}[/{test.result}]", highlight=False)
+            console.print("\n".join(f"    {message}" for message in test.messages), highlight=False)
 
 
 def print_jinja(results: ResultManager, template: pathlib.Path, output: pathlib.Path | None = None) -> None:

anta/input_models/__init__.py

@@ -0,0 +1,4 @@
+# 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.
+"""Package related to all ANTA tests input models."""

anta/input_models/bfd.py

@@ -0,0 +1,37 @@
+# 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 BFD tests."""
+
+from __future__ import annotations
+
+from ipaddress import IPv4Address
+
+from pydantic import BaseModel, ConfigDict
+
+from anta.custom_types import BfdInterval, BfdMultiplier, BfdProtocol
+
+
+class BFDPeer(BaseModel):
+    """BFD (Bidirectional Forwarding Detection) model representing the peer details.
+
+    Only IPv4 peers are supported for now.
+    """
+
+    model_config = ConfigDict(extra="forbid")
+    peer_address: IPv4Address
+    """IPv4 address of a BFD peer."""
+    vrf: str = "default"
+    """Optional VRF for the BFD peer. Defaults to `default`."""
+    tx_interval: BfdInterval | None = None
+    """Tx interval of BFD peer in milliseconds. Required field in the `VerifyBFDPeersIntervals` test."""
+    rx_interval: BfdInterval | None = None
+    """Rx interval of BFD peer in milliseconds. Required field in the `VerifyBFDPeersIntervals` test."""
+    multiplier: BfdMultiplier | None = None
+    """Multiplier of BFD peer. Required field in the `VerifyBFDPeersIntervals` test."""
+    protocols: list[BfdProtocol] | None = None
+    """List of protocols to be verified. Required field in the `VerifyBFDPeersRegProtocols` test."""
+
+    def __str__(self) -> str:
+        """Return a human-readable string representation of the BFDPeer for reporting."""
+        return f"Peer: {self.peer_address} VRF: {self.vrf}"

anta/input_models/connectivity.py

@@ -0,0 +1,83 @@
+# 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 connectivity tests."""
+
+from __future__ import annotations
+
+from ipaddress import IPv4Address
+from typing import Any
+from warnings import warn
+
+from pydantic import BaseModel, ConfigDict
+
+from anta.custom_types import Interface
+
+
+class Host(BaseModel):
+    """Model for a remote host to ping."""
+
+    model_config = ConfigDict(extra="forbid")
+    destination: IPv4Address
+    """IPv4 address to ping."""
+    source: IPv4Address | Interface
+    """IPv4 address source IP or egress interface to use."""
+    vrf: str = "default"
+    """VRF context. Defaults to `default`."""
+    repeat: int = 2
+    """Number of ping repetition. Defaults to 2."""
+    size: int = 100
+    """Specify datagram size. Defaults to 100."""
+    df_bit: bool = False
+    """Enable do not fragment bit in IP header. Defaults to False."""
+
+    def __str__(self) -> str:
+        """Return a human-readable string representation of the Host for reporting.
+
+        Examples
+        --------
+        Host 10.1.1.1 (src: 10.2.2.2, vrf: mgmt, size: 100B, repeat: 2)
+
+        """
+        df_status = ", df-bit: enabled" if self.df_bit else ""
+        return f"Host {self.destination} (src: {self.source}, vrf: {self.vrf}, size: {self.size}B, repeat: {self.repeat}{df_status})"
+
+
+class LLDPNeighbor(BaseModel):
+    """LLDP (Link Layer Discovery Protocol) model representing the port details and neighbor information."""
+
+    model_config = ConfigDict(extra="forbid")
+    port: Interface
+    """The LLDP port for the local device."""
+    neighbor_device: str
+    """The system name of the LLDP neighbor device."""
+    neighbor_port: Interface
+    """The LLDP port on the neighboring device."""
+
+    def __str__(self) -> str:
+        """Return a human-readable string representation of the LLDPNeighbor for reporting.
+
+        Examples
+        --------
+        Port Ethernet1 (Neighbor: DC1-SPINE2, Neighbor Port: Ethernet2)
+
+        """
+        return f"Port {self.port} (Neighbor: {self.neighbor_device}, Neighbor Port: {self.neighbor_port})"
+
+
+class Neighbor(LLDPNeighbor):  # pragma: no cover
+    """Alias for the LLDPNeighbor model to maintain backward compatibility.
+
+    When initialized, it will emit a deprecation warning and call the LLDPNeighbor model.
+
+    TODO: Remove this class in ANTA v2.0.0.
+    """
+
+    def __init__(self, **data: Any) -> None:  # noqa: ANN401
+        """Initialize the LLDPNeighbor class, emitting a depreciation warning."""
+        warn(
+            message="Neighbor model is deprecated and will be removed in ANTA v2.0.0. Use the LLDPNeighbor model instead.",
+            category=DeprecationWarning,
+            stacklevel=2,
+        )
+        super().__init__(**data)

anta/input_models/interfaces.py

@@ -0,0 +1,23 @@
+# 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 interface tests."""
+
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import BaseModel
+
+from anta.custom_types import Interface
+
+
+class InterfaceState(BaseModel):
+    """Model for an interface state."""
+
+    name: Interface
+    """Interface to validate."""
+    status: Literal["up", "down", "adminDown"]
+    """Expected status of the interface."""
+    line_protocol_status: Literal["up", "down", "testing", "unknown", "dormant", "notPresent", "lowerLayerDown"] | None = None
+    """Expected line protocol status of the interface."""

anta/input_models/routing/__init__.py

@@ -0,0 +1,4 @@
+# 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.
+"""Package related to routing tests input models."""

anta/input_models/routing/bgp.py

@@ -0,0 +1,130 @@
+# 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 routing BGP tests."""
+
+from __future__ import annotations
+
+from ipaddress import IPv4Address, IPv6Address
+from typing import TYPE_CHECKING, Any
+from warnings import warn
+
+from pydantic import BaseModel, ConfigDict, PositiveInt, model_validator
+
+from anta.custom_types import Afi, Safi
+
+if TYPE_CHECKING:
+    import sys
+
+    if sys.version_info >= (3, 11):
+        from typing import Self
+    else:
+        from typing_extensions import Self
+
+AFI_SAFI_EOS_KEY = {
+    ("ipv4", "unicast"): "ipv4Unicast",
+    ("ipv4", "multicast"): "ipv4Multicast",
+    ("ipv4", "labeled-unicast"): "ipv4MplsLabels",
+    ("ipv4", "sr-te"): "ipv4SrTe",
+    ("ipv6", "unicast"): "ipv6Unicast",
+    ("ipv6", "multicast"): "ipv6Multicast",
+    ("ipv6", "labeled-unicast"): "ipv6MplsLabels",
+    ("ipv6", "sr-te"): "ipv6SrTe",
+    ("vpn-ipv4", None): "ipv4MplsVpn",
+    ("vpn-ipv6", None): "ipv6MplsVpn",
+    ("evpn", None): "l2VpnEvpn",
+    ("rt-membership", None): "rtMembership",
+    ("path-selection", None): "dps",
+    ("link-state", None): "linkState",
+}
+"""Dictionary mapping AFI/SAFI to EOS key representation."""
+
+
+class BgpAddressFamily(BaseModel):
+    """Model for a BGP address family."""
+
+    model_config = ConfigDict(extra="forbid")
+    afi: Afi
+    """BGP Address Family Identifier (AFI)."""
+    safi: Safi | None = None
+    """BGP Subsequent Address Family Identifier (SAFI). Required when `afi` is `ipv4` or `ipv6`."""
+    vrf: str = "default"
+    """Optional VRF when `afi` is `ipv4` or `ipv6`. Defaults to `default`.
+
+    If the input `afi` is NOT `ipv4` or `ipv6` (e.g. `evpn`, `vpn-ipv4`, etc.), the `vrf` must be `default`.
+
+    These AFIs operate at a global level and do not use the VRF concept in the same way as IPv4/IPv6.
+    """
+    num_peers: PositiveInt | None = None
+    """Number of expected established BGP peers with negotiated AFI/SAFI. Required field in the `VerifyBGPPeerCount` test."""
+    peers: list[IPv4Address | IPv6Address] | None = None
+    """List of expected IPv4/IPv6 BGP peers supporting the AFI/SAFI. Required field in the `VerifyBGPSpecificPeers` test."""
+    check_tcp_queues: bool = True
+    """Flag to check if the TCP session queues are empty for a BGP peer. Defaults to `True`.
+
+    Can be disabled in the `VerifyBGPPeersHealth` and `VerifyBGPSpecificPeers` tests.
+    """
+    check_peer_state: bool = False
+    """Flag to check if the peers are established with negotiated AFI/SAFI. Defaults to `False`.
+
+    Can be enabled in the `VerifyBGPPeerCount` tests.
+    """
+
+    @model_validator(mode="after")
+    def validate_inputs(self) -> Self:
+        """Validate the inputs provided to the BgpAddressFamily class.
+
+        If `afi` is either `ipv4` or `ipv6`, `safi` must be provided.
+
+        If `afi` is not `ipv4` or `ipv6`, `safi` must NOT be provided and `vrf` must be `default`.
+        """
+        if self.afi in ["ipv4", "ipv6"]:
+            if self.safi is None:
+                msg = "'safi' must be provided when afi is ipv4 or ipv6"
+                raise ValueError(msg)
+        elif self.safi is not None:
+            msg = "'safi' must not be provided when afi is not ipv4 or ipv6"
+            raise ValueError(msg)
+        elif self.vrf != "default":
+            msg = "'vrf' must be default when afi is not ipv4 or ipv6"
+            raise ValueError(msg)
+        return self
+
+    @property
+    def eos_key(self) -> str:
+        """AFI/SAFI EOS key representation."""
+        # Pydantic handles the validation of the AFI/SAFI combination, so we can ignore error handling here.
+        return AFI_SAFI_EOS_KEY[(self.afi, self.safi)]
+
+    def __str__(self) -> str:
+        """Return a string representation of the BgpAddressFamily model. Used in failure messages.
+
+        Examples
+        --------
+        - AFI:ipv4 SAFI:unicast VRF:default
+        - AFI:evpn
+        """
+        base_string = f"AFI: {self.afi}"
+        if self.safi is not None:
+            base_string += f" SAFI: {self.safi}"
+        if self.afi in ["ipv4", "ipv6"]:
+            base_string += f" VRF: {self.vrf}"
+        return base_string
+
+
+class BgpAfi(BgpAddressFamily):  # pragma: no cover
+    """Alias for the BgpAddressFamily model to maintain backward compatibility.
+
+    When initialized, it will emit a deprecation warning and call the BgpAddressFamily model.
+
+    TODO: Remove this class in ANTA v2.0.0.
+    """
+
+    def __init__(self, **data: Any) -> None:  # noqa: ANN401
+        """Initialize the BgpAfi class, emitting a deprecation warning."""
+        warn(
+            message="BgpAfi model is deprecated and will be removed in ANTA v2.0.0. Use the BgpAddressFamily model instead.",
+            category=DeprecationWarning,
+            stacklevel=2,
+        )
+        super().__init__(**data)