Merge branch 'main' into anta-get-tests

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

@@ -0,0 +1,41 @@
+# 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 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})"

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/tests/connectivity.py

@@ -7,12 +7,12 @@
 # mypy: disable-error-code=attr-defined
 from __future__ import annotations
 
-from ipaddress import IPv4Address
 from typing import ClassVar
 
 from pydantic import BaseModel
 
 from anta.custom_types import Interface
+from anta.input_models.connectivity import Host
 from anta.models import AntaCommand, AntaTemplate, AntaTest
 
 
@@ -44,8 +44,7 @@ class VerifyReachability(AntaTest):
     """
 
     categories: ClassVar[list[str]] = ["connectivity"]
-    # Removing the <space> between '{size}' and '{df_bit}' to compensate the df-bit set default value
-    # i.e if df-bit kept disable then it will add redundant space in between the command
+    # Template uses '{size}{df_bit}' without space since df_bit includes leading space when enabled
     commands: ClassVar[list[AntaCommand | AntaTemplate]] = [
         AntaTemplate(template="ping vrf {vrf} {destination} source {source} size {size}{df_bit} repeat {repeat}", revision=1)
     ]
@@ -55,29 +54,13 @@ class Input(AntaTest.Input):
 
         hosts: list[Host]
         """List of host to ping."""
-
-        class Host(BaseModel):
-            """Model for a remote host to ping."""
-
-            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."""
+        Host: ClassVar[type[Host]] = Host
 
     def render(self, template: AntaTemplate) -> list[AntaCommand]:
         """Render the template for each host in the input list."""
         commands = []
         for host in self.inputs.hosts:
-            # Enables do not fragment bit in IP header if needed else keeping disable.
-            # Adding the <space> at start to compensate change in AntaTemplate
+            # df_bit includes leading space when enabled, empty string when disabled
             df_bit = " df-bit" if host.df_bit else ""
             command = template.render(destination=host.destination, source=host.source, vrf=host.vrf, repeat=host.repeat, size=host.size, df_bit=df_bit)
             commands.append(command)
@@ -86,20 +69,11 @@ def render(self, template: AntaTemplate) -> list[AntaCommand]:
     @AntaTest.anta_test
     def test(self) -> None:
         """Main test function for VerifyReachability."""
-        failures = []
-
-        for command in self.instance_commands:
-            src = command.params.source
-            dst = command.params.destination
-            repeat = command.params.repeat
-
-            if f"{repeat} received" not in command.json_output["messages"][0]:
-                failures.append((str(src), str(dst)))
+        self.result.is_success()
 
-        if not failures:
-            self.result.is_success()
-        else:
-            self.result.is_failure(f"Connectivity test failed for the following source-destination pairs: {failures}")
+        for command, host in zip(self.instance_commands, self.inputs.hosts):
+            if f"{host.repeat} received" not in command.json_output["messages"][0]:
+                self.result.is_failure(f"{host} - Unreachable")
 
 
 class VerifyLLDPNeighbors(AntaTest):

anta/tests/interfaces.py

@@ -9,14 +9,15 @@
 
 import re
 from ipaddress import IPv4Network
-from typing import Any, ClassVar, Literal
+from typing import Any, ClassVar
 
 from pydantic import BaseModel, Field
 from pydantic_extra_types.mac_address import MacAddress
 
 from anta import GITHUB_SUGGESTION
 from anta.custom_types import EthernetInterface, Interface, Percent, PortChannelInterface, PositiveInteger
 from anta.decorators import skip_on_platforms
+from anta.input_models.interfaces import InterfaceState
 from anta.models import AntaCommand, AntaTemplate, AntaTest
 from anta.tools import custom_division, get_failed_logs, get_item, get_value
 
@@ -183,16 +184,20 @@ def test(self) -> None:
 
 
 class VerifyInterfacesStatus(AntaTest):
-    """Verifies if the provided list of interfaces are all in the expected state.
+    """Verifies the operational states of specified interfaces to ensure they match expected configurations.
 
-    - If line protocol status is provided, prioritize checking against both status and line protocol status
-    - If line protocol status is not provided and interface status is "up", expect both status and line protocol to be "up"
-    - If interface status is not "up", check only the interface status without considering line protocol status
+    This test performs the following checks for each specified interface:
+
+      1. If `line_protocol_status` is defined, both `status` and `line_protocol_status` are verified for the specified interface.
+      2. If `line_protocol_status` is not provided but the `status` is "up", it is assumed that both the status and line protocol should be "up".
+      3. If the interface `status` is not "up", only the interface's status is validated, with no line protocol check performed.
 
     Expected Results
     ----------------
-    * Success: The test will pass if the provided interfaces are all in the expected state.
-    * Failure: The test will fail if any interface is not in the expected state.
+    * Success: If the interface status and line protocol status matches the expected operational state for all specified interfaces.
+    * Failure: If any of the following occur:
+        - The specified interface is not configured.
+        - The specified interface status and line protocol status does not match the expected operational state for any interface.
 
     Examples
     --------
@@ -219,30 +224,17 @@ class Input(AntaTest.Input):
 
         interfaces: list[InterfaceState]
         """List of interfaces with their expected state."""
-
-        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."""
+        InterfaceState: ClassVar[type[InterfaceState]] = InterfaceState
 
     @AntaTest.anta_test
     def test(self) -> None:
         """Main test function for VerifyInterfacesStatus."""
-        command_output = self.instance_commands[0].json_output
-
         self.result.is_success()
 
-        intf_not_configured = []
-        intf_wrong_state = []
-
+        command_output = self.instance_commands[0].json_output
         for interface in self.inputs.interfaces:
             if (intf_status := get_value(command_output["interfaceDescriptions"], interface.name, separator="..")) is None:
-                intf_not_configured.append(interface.name)
+                self.result.is_failure(f"{interface.name} - Not configured")
                 continue
 
             status = "up" if intf_status["interfaceStatus"] in {"up", "connected"} else intf_status["interfaceStatus"]
@@ -251,18 +243,15 @@ def test(self) -> None:
             # If line protocol status is provided, prioritize checking against both status and line protocol status
             if interface.line_protocol_status:
                 if interface.status != status or interface.line_protocol_status != proto:
-                    intf_wrong_state.append(f"{interface.name} is {status}/{proto}")
+                    actual_state = f"Expected: {interface.status}/{interface.line_protocol_status}, Actual: {status}/{proto}"
+                    self.result.is_failure(f"{interface.name} - {actual_state}")
 
             # If line protocol status is not provided and interface status is "up", expect both status and proto to be "up"
             # If interface status is not "up", check only the interface status without considering line protocol status
-            elif (interface.status == "up" and (status != "up" or proto != "up")) or (interface.status != status):
-                intf_wrong_state.append(f"{interface.name} is {status}/{proto}")
-
-        if intf_not_configured:
-            self.result.is_failure(f"The following interface(s) are not configured: {intf_not_configured}")
-
-        if intf_wrong_state:
-            self.result.is_failure(f"The following interface(s) are not in the expected state: {intf_wrong_state}")
+            elif interface.status == "up" and (status != "up" or proto != "up"):
+                self.result.is_failure(f"{interface.name} - Expected: up/up, Actual: {status}/{proto}")
+            elif interface.status != status:
+                self.result.is_failure(f"{interface.name} - Expected: {interface.status}, Actual: {status}")
 
 
 class VerifyStormControlDrops(AntaTest):

docs/api/tests.connectivity.md

@@ -7,7 +7,10 @@ anta_title: ANTA catalog for connectivity tests
   ~ that can be found in the LICENSE file.
   -->
 
+# Tests
+
 ::: anta.tests.connectivity
+
     options:
       show_root_heading: false
       show_root_toc_entry: false
@@ -18,3 +21,15 @@ anta_title: ANTA catalog for connectivity tests
       filters:
         - "!test"
         - "!render"
+
+# Input models
+
+::: anta.input_models.connectivity
+
+    options:
+      show_root_heading: false
+      show_root_toc_entry: false
+      show_bases: false
+      anta_hide_test_module_description: true
+      show_labels: true
+      filters: ["!^__str__"]

docs/api/tests.interfaces.md

@@ -7,7 +7,10 @@ anta_title: ANTA catalog for interfaces tests
   ~ that can be found in the LICENSE file.
   -->
 
+# Tests
+
 ::: anta.tests.interfaces
+
     options:
       show_root_heading: false
       show_root_toc_entry: false
@@ -18,3 +21,15 @@ anta_title: ANTA catalog for interfaces tests
       filters:
         - "!test"
         - "!render"
+
+# Input models
+
+::: anta.input_models.interfaces
+
+    options:
+      show_root_heading: false
+      show_root_toc_entry: false
+      show_bases: false
+      anta_hide_test_module_description: true
+      show_labels: true
+      filters: ["!^__str__"]

docs/cli/exec.md

@@ -235,9 +235,10 @@ Options:
                           tag1,tag2,tag3.  [env var: ANTA_TAGS]
   -o, --output PATH       Path for test catalog  [default: ./tech-support]
   --latest INTEGER        Number of scheduled show-tech to retrieve
-  --configure             Ensure devices have 'aaa authorization exec default
-                          local' configured (required for SCP on EOS). THIS
-                          WILL CHANGE THE CONFIGURATION OF YOUR NETWORK.
+  --configure             [DEPRECATED] Ensure devices have 'aaa authorization
+                          exec default local' configured (required for SCP on
+                          EOS). THIS WILL CHANGE THE CONFIGURATION OF YOUR
+                          NETWORK.
   --help                  Show this message and exit.
 ```
 
@@ -248,7 +249,10 @@ When executed, this command fetches tech-support files and downloads them locall
 ANTA uses SCP to download files from devices and will not trust unknown SSH hosts by default. Add the SSH public keys of your devices to your `known_hosts` file or use the `anta --insecure` option to ignore SSH host keys validation.
 
 The configuration `aaa authorization exec default` must be present on devices to be able to use SCP.
-ANTA can automatically configure `aaa authorization exec default local` using the `anta exec collect-tech-support --configure` option.
+
+!!! warning Deprecation
+    ANTA can automatically configure `aaa authorization exec default local` using the `anta exec collect-tech-support --configure` option but this option is deprecated and will be removed in ANTA 2.0.0.
+
 If you require specific AAA configuration for `aaa authorization exec default`, like `aaa authorization exec default none` or `aaa authorization exec default group tacacs+`, you will need to configure it manually.
 
 The `--latest` option allows retrieval of a specific number of the most recent tech-support files.