Merge branch 'main' into fix/issue_948

anta/decorators.py

@@ -63,13 +63,15 @@ async def wrapper(*args: Any, **kwargs: Any) -> Any:
     return decorator
 
 
-def deprecated_test_class(new_tests: list[str] | None = None) -> Callable[[type[AntaTest]], type[AntaTest]]:
+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
     -------
@@ -102,6 +104,9 @@ def new_init(*args: Any, **kwargs: Any) -> None:
                 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

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

@@ -328,6 +328,8 @@ def test(self) -> None:
     # Optional class attributes
     name: ClassVar[str]
     description: ClassVar[str]
+    __removal_in_version: ClassVar[str]
+    """Internal class variable set by the `deprecated_test_class` decorator."""
 
     # Mandatory class attributes
     # TODO: find a way to tell mypy these are mandatory for child classes

anta/tests/interfaces.py

@@ -15,11 +15,11 @@
 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.custom_types import EthernetInterface, Interface, Percent, 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
+from anta.tools import custom_division, format_data, get_failed_logs, get_item, get_value
 
 BPS_GBPS_CONVERSIONS = 1000000000
 
@@ -848,17 +848,27 @@ def test(self) -> None:
 
 
 class VerifyLACPInterfacesStatus(AntaTest):
-    """Verifies the Link Aggregation Control Protocol (LACP) status of the provided interfaces.
+    """Verifies the Link Aggregation Control Protocol (LACP) status of the interface.
 
-    - Verifies that the interface is a member of the LACP port channel.
-    - Ensures that the synchronization is established.
-    - Ensures the interfaces are in the correct state for collecting and distributing traffic.
-    - Validates that LACP settings, such as timeouts, are correctly configured. (i.e The long timeout mode, also known as "slow" mode, is the default setting.)
+    This test performs the following checks for each specified interface:
+
+      1. Verifies that the interface is a member of the LACP port channel.
+      2. Verifies LACP port states and operational status:
+        - Activity: Active LACP mode (initiates)
+        - Timeout: Short (Fast Mode), Long (Slow Mode - default)
+        - Aggregation: Port aggregable
+        - Synchronization: Port in sync with partner
+        - Collecting: Incoming frames aggregating
+        - Distributing: Outgoing frames aggregating
 
     Expected Results
     ----------------
-    * Success: The test will pass if the provided interfaces are bundled in port channel and all specified parameters are correct.
-    * Failure: The test will fail if any interface is not bundled in port channel or any of specified parameter is not correct.
+    * Success: Interface is bundled and all LACP states match expected values for both actor and partner
+    * Failure: If any of the following occur:
+        - Interface or port channel is not configured.
+        - Interface is not bundled in port channel.
+        - Actor or partner port LACP states don't match expected configuration.
+        - LACP rate (timeout) mismatch when fast mode is configured.
 
     Examples
     --------
@@ -872,25 +882,14 @@ class VerifyLACPInterfacesStatus(AntaTest):
     """
 
     categories: ClassVar[list[str]] = ["interfaces"]
-    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaTemplate(template="show lacp interface {interface}", revision=1)]
+    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show lacp interface", revision=1)]
 
     class Input(AntaTest.Input):
         """Input model for the VerifyLACPInterfacesStatus test."""
 
-        interfaces: list[LACPInterface]
-        """List of LACP member interface."""
-
-        class LACPInterface(BaseModel):
-            """Model for an LACP member interface."""
-
-            name: EthernetInterface
-            """Ethernet interface to validate."""
-            portchannel: PortChannelInterface
-            """Port Channel in which the interface is bundled."""
-
-    def render(self, template: AntaTemplate) -> list[AntaCommand]:
-        """Render the template for each interface in the input list."""
-        return [template.render(interface=interface.name) for interface in self.inputs.interfaces]
+        interfaces: list[InterfaceState]
+        """List of interfaces with their expected state."""
+        InterfaceState: ClassVar[type[InterfaceState]] = InterfaceState
 
     @AntaTest.anta_test
     def test(self) -> None:
@@ -900,21 +899,17 @@ def test(self) -> None:
         # Member port verification parameters.
         member_port_details = ["activity", "aggregation", "synchronization", "collecting", "distributing", "timeout"]
 
-        # Iterating over command output for different interfaces
-        for command, input_entry in zip(self.instance_commands, self.inputs.interfaces):
-            interface = input_entry.name
-            portchannel = input_entry.portchannel
-
+        command_output = self.instance_commands[0].json_output
+        for interface in self.inputs.interfaces:
             # Verify if a PortChannel is configured with the provided interface
-            if not (interface_details := get_value(command.json_output, f"portChannels.{portchannel}.interfaces.{interface}")):
-                self.result.is_failure(f"Interface '{interface}' is not configured to be a member of LACP '{portchannel}'.")
+            if not (interface_details := get_value(command_output, f"portChannels..{interface.portchannel}..interfaces..{interface.name}", separator="..")):
+                self.result.is_failure(f"{interface} - Not configured")
                 continue
 
             # Verify the interface is bundled in port channel.
             actor_port_status = interface_details.get("actorPortStatus")
             if actor_port_status != "bundled":
-                message = f"For Interface {interface}:\nExpected `bundled` as the local port status, but found `{actor_port_status}` instead.\n"
-                self.result.is_failure(message)
+                self.result.is_failure(f"{interface} - Not bundled - Port Status: {actor_port_status}")
                 continue
 
             # Collecting actor and partner port details
@@ -929,21 +924,12 @@ def test(self) -> None:
 
             # Forming expected interface details
             expected_details = {param: param != "timeout" for param in member_port_details}
-            expected_interface_output = {"actor_port_details": expected_details, "partner_port_details": expected_details}
+            # Updating the short LACP timeout, if expected.
+            if interface.lacp_rate_fast:
+                expected_details["timeout"] = True
 
-            # Forming failure message
-            if actual_interface_output != expected_interface_output:
-                message = f"For Interface {interface}:\n"
-                actor_port_failed_log = get_failed_logs(
-                    expected_interface_output.get("actor_port_details", {}), actual_interface_output.get("actor_port_details", {})
-                )
-                partner_port_failed_log = get_failed_logs(
-                    expected_interface_output.get("partner_port_details", {}), actual_interface_output.get("partner_port_details", {})
-                )
-
-                if actor_port_failed_log:
-                    message += f"Actor port details:{actor_port_failed_log}\n"
-                if partner_port_failed_log:
-                    message += f"Partner port details:{partner_port_failed_log}\n"
-
-                self.result.is_failure(message)
+            if (act_port_details := actual_interface_output["actor_port_details"]) != expected_details:
+                self.result.is_failure(f"{interface} - Actor port details mismatch - {format_data(act_port_details)}")
+
+            if (part_port_details := actual_interface_output["partner_port_details"]) != expected_details:
+                self.result.is_failure(f"{interface} - Partner port details mismatch - {format_data(part_port_details)}")

anta/tests/stun.py

@@ -91,15 +91,13 @@ def test(self) -> None:
                 self.result.is_failure(f"{client_input} - Incorrect public-facing port - Expected: {input_public_port} Actual: {actual_public_port}")
 
 
-@deprecated_test_class(new_tests=["VerifyStunClientTranslation"])
+@deprecated_test_class(new_tests=["VerifyStunClientTranslation"], removal_in_version="v2.0.0")
 class VerifyStunClient(VerifyStunClientTranslation):
     """(Deprecated) Verifies the translation for a source address on a STUN client.
 
     Alias for the VerifyStunClientTranslation test to maintain backward compatibility.
     When initialized, it will emit a deprecation warning and call the VerifyStunClientTranslation test.
 
-    TODO: Remove this class in ANTA v2.0.0.
-
     Examples
     --------
     ```yaml
@@ -113,6 +111,8 @@ class VerifyStunClient(VerifyStunClientTranslation):
     ```
     """
 
+    # TODO: Remove this class in ANTA v2.0.0.
+
     # required to redefine name an description to overwrite parent class.
     name = "VerifyStunClient"
     description = "(Deprecated) Verifies the translation for a source address on a STUN client."

docs/templates/python/material/class.html.jinja

@@ -53,3 +53,21 @@
   {{ super() }}
 {%   endif %}
 {% endblock source %}
+
+{# overwrite block base to render some stuff on deprecation for anta_test #}
+{% block bases %}
+{{ super() }}
+
+{% for dec in class.decorators %}
+{% if dec.value.function.name == "deprecated_test_class" %}
+<img alt="Static Badge" src="https://img.shields.io/badge/DEPRECATED-yellow?style=flat&logoSize=auto">
+{% for arg in dec.value.arguments | selectattr("name", "equalto", "removal_in_version") | list %}
+<img alt="Static Badge" src="https://img.shields.io/badge/REMOVAL-{{ arg.value[1:-1] }}-grey?style=flat&logoSize=auto&labelColor=red">
+{% endfor %}
+<br/>
+{% for arg in dec.value.arguments | selectattr("name", "equalto", "new_tests") | list %}
+<strong>Replaced with:</strong> {{ arg.value.elements | map("replace", "'", "<code>", 1) | map("replace", "'", "</code>", 1) | join(", ") | safe }}
+{% endfor %}
+{% endif %}
+{% endfor %}
+{% endblock bases %}

examples/tests.yaml

@@ -253,7 +253,7 @@ anta.tests.interfaces:
       specific_mtu:
           - Ethernet1: 2500
   - VerifyLACPInterfacesStatus:
-      # Verifies the Link Aggregation Control Protocol (LACP) status of the provided interfaces.
+      # Verifies the Link Aggregation Control Protocol (LACP) status of the interface.
       interfaces:
         - name: Ethernet1
           portchannel: Port-Channel100