Repository: aristanetworks/anta (python)

Issue:
Fix: VerifyManagementCVX should not throw  an error
```
[16:44:29] CRITICAL Exception raised for test VerifyManagementCVX (on device spine2)                                                                                                                                                             logger.py:121
                    KeyError: 'clusterStatus'
```

Pull request:
fix(anta.tests): Adjust revision for show management cvx
# Description

Test is throwing error when run on a a device not running CVX at all. Looking at the unit tests, it feels the revision for the command should be `3` and not `1` (cf issue comments)

Fixes #953

# 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/anta/decorators.py b/anta/decorators.py
index 08fbd653..04316232 100644
--- a/anta/decorators.py
+++ b/anta/decorators.py
@@ -63,13 +63,15 @@ def deprecated_test(new_tests: list[str] | None = None) -> Callable[[F], F]:  #
     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 deprecated_test_class(new_tests: list[str] | None = None) -> Callable[[type[
                 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
diff --git a/anta/input_models/interfaces.py b/anta/input_models/interfaces.py
index 5036156d..9e33a2c5 100644
--- a/anta/input_models/interfaces.py
+++ b/anta/input_models/interfaces.py
@@ -7,17 +7,42 @@ from __future__ import annotations
 
 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
diff --git a/anta/models.py b/anta/models.py
index 4cebd997..69f305e2 100644
--- a/anta/models.py
+++ b/anta/models.py
@@ -328,6 +328,8 @@ class AntaTest(ABC):
     # 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
diff --git a/docs/templates/python/material/class.html.jinja b/docs/templates/python/material/class.html.jinja
index cf016c92..cbf9fac2 100644
--- a/docs/templates/python/material/class.html.jinja
+++ b/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 %}


Test patch:
diff --git a/anta/tests/cvx.py b/anta/tests/cvx.py
index 63ec336a..e80af2e4 100644
--- a/anta/tests/cvx.py
+++ b/anta/tests/cvx.py
@@ -10,6 +10,7 @@ from __future__ import annotations
 from typing import TYPE_CHECKING, ClassVar
 
 from anta.models import AntaCommand, AntaTest
+from anta.tools import get_value
 
 if TYPE_CHECKING:
     from anta.models import AntaTemplate
@@ -71,7 +72,7 @@ class VerifyManagementCVX(AntaTest):
     """
 
     categories: ClassVar[list[str]] = ["cvx"]
-    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show management cvx", revision=1)]
+    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show management cvx", revision=3)]
 
     class Input(AntaTest.Input):
         """Input model for the VerifyManagementCVX test."""
@@ -84,6 +85,5 @@ class VerifyManagementCVX(AntaTest):
         """Main test function for VerifyManagementCVX."""
         command_output = self.instance_commands[0].json_output
         self.result.is_success()
-        cluster_status = command_output["clusterStatus"]
-        if (cluster_state := cluster_status.get("enabled")) != self.inputs.enabled:
+        if (cluster_state := get_value(command_output, "clusterStatus.enabled")) != self.inputs.enabled:
             self.result.is_failure(f"Management CVX status is not valid: {cluster_state}")
diff --git a/anta/tests/interfaces.py b/anta/tests/interfaces.py
index dc693811..b87c394f 100644
--- a/anta/tests/interfaces.py
+++ b/anta/tests/interfaces.py
@@ -15,11 +15,11 @@ 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.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 @@ class VerifyInterfacesSpeed(AntaTest):
 
 
 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 @@ class VerifyLACPInterfacesStatus(AntaTest):
         # 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 @@ class VerifyLACPInterfacesStatus(AntaTest):
 
             # 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)}")
diff --git a/anta/tests/stun.py b/anta/tests/stun.py
index 925abd14..2be13c4b 100644
--- a/anta/tests/stun.py
+++ b/anta/tests/stun.py
@@ -91,15 +91,13 @@ class VerifyStunClientTranslation(AntaTest):
                 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."
diff --git a/examples/tests.yaml b/examples/tests.yaml
index e14b4082..d307215f 100644
--- a/examples/tests.yaml
+++ b/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
diff --git a/tests/units/anta_tests/test_cvx.py b/tests/units/anta_tests/test_cvx.py
index 0d4cec4e..03de7fda 100644
--- a/tests/units/anta_tests/test_cvx.py
+++ b/tests/units/anta_tests/test_cvx.py
@@ -140,10 +140,17 @@ DATA: list[dict[str, Any]] = [
         "expected": {"result": "success"},
     },
     {
-        "name": "failure",
+        "name": "failure - no enabled state",
         "test": VerifyManagementCVX,
         "eos_data": [{"clusterStatus": {}}],
         "inputs": {"enabled": False},
         "expected": {"result": "failure", "messages": ["Management CVX status is not valid: None"]},
     },
+    {
+        "name": "failure - no clusterStatus",
+        "test": VerifyManagementCVX,
+        "eos_data": [{}],
+        "inputs": {"enabled": False},
+        "expected": {"result": "failure", "messages": ["Management CVX status is not valid: None"]},
+    },
 ]
diff --git a/tests/units/anta_tests/test_interfaces.py b/tests/units/anta_tests/test_interfaces.py
index ac053088..271683b0 100644
--- a/tests/units/anta_tests/test_interfaces.py
+++ b/tests/units/anta_tests/test_interfaces.py
@@ -2510,6 +2510,43 @@ DATA: list[dict[str, Any]] = [
         "inputs": {"interfaces": [{"name": "Ethernet5", "portchannel": "Port-Channel5"}]},
         "expected": {"result": "success"},
     },
+    {
+        "name": "success-short-timeout",
+        "test": VerifyLACPInterfacesStatus,
+        "eos_data": [
+            {
+                "portChannels": {
+                    "Port-Channel5": {
+                        "interfaces": {
+                            "Ethernet5": {
+                                "actorPortStatus": "bundled",
+                                "partnerPortState": {
+                                    "activity": True,
+                                    "timeout": True,
+                                    "aggregation": True,
+                                    "synchronization": True,
+                                    "collecting": True,
+                                    "distributing": True,
+                                },
+                                "actorPortState": {
+                                    "activity": True,
+                                    "timeout": True,
+                                    "aggregation": True,
+                                    "synchronization": True,
+                                    "collecting": True,
+                                    "distributing": True,
+                                },
+                            }
+                        }
+                    }
+                },
+                "interface": "Ethernet5",
+                "orphanPorts": {},
+            }
+        ],
+        "inputs": {"interfaces": [{"name": "Ethernet5", "portchannel": "Port-Channel5", "lacp_rate_fast": True}]},
+        "expected": {"result": "success"},
+    },
     {
         "name": "failure-not-bundled",
         "test": VerifyLACPInterfacesStatus,
@@ -2531,7 +2568,7 @@ DATA: list[dict[str, Any]] = [
         "inputs": {"interfaces": [{"name": "Ethernet5", "portchannel": "Po5"}]},
         "expected": {
             "result": "failure",
-            "messages": ["For Interface Ethernet5:\nExpected `bundled` as the local port status, but found `No Aggregate` instead.\n"],
+            "messages": ["Interface: Ethernet5 Port-Channel: Port-Channel5 - Not bundled - Port Status: No Aggregate"],
         },
     },
     {
@@ -2545,7 +2582,7 @@ DATA: list[dict[str, Any]] = [
         "inputs": {"interfaces": [{"name": "Ethernet5", "portchannel": "Po 5"}]},
         "expected": {
             "result": "failure",
-            "messages": ["Interface 'Ethernet5' is not configured to be a member of LACP 'Port-Channel5'."],
+            "messages": ["Interface: Ethernet5 Port-Channel: Port-Channel5 - Not configured"],
         },
     },
     {
@@ -2586,13 +2623,55 @@ DATA: list[dict[str, Any]] = [
         "expected": {
             "result": "failure",
             "messages": [
-                "For Interface Ethernet5:\n"
-                "Actor port details:\nExpected `True` as the activity, but found `False` instead."
-                "\nExpected `True` as the aggregation, but found `False` instead."
-                "\nExpected `True` as the synchronization, but found `False` instead."
-                "\nPartner port details:\nExpected `True` as the activity, but found `False` instead.\n"
-                "Expected `True` as the aggregation, but found `False` instead.\n"
-                "Expected `True` as the synchronization, but found `False` instead.\n"
+                "Interface: Ethernet5 Port-Channel: Port-Channel5 - Actor port details mismatch - Activity: False, Aggregation: False, "
+                "Synchronization: False, Collecting: True, Distributing: True, Timeout: False",
+                "Interface: Ethernet5 Port-Channel: Port-Channel5 - Partner port details mismatch - Activity: False, Aggregation: False, "
+                "Synchronization: False, Collecting: True, Distributing: True, Timeout: False",
+            ],
+        },
+    },
+    {
+        "name": "failure-short-timeout",
+        "test": VerifyLACPInterfacesStatus,
+        "eos_data": [
+            {
+                "portChannels": {
+                    "Port-Channel5": {
+                        "interfaces": {
+                            "Ethernet5": {
+                                "actorPortStatus": "bundled",
+                                "partnerPortState": {
+                                    "activity": True,
+                                    "timeout": False,
+                                    "aggregation": True,
+                                    "synchronization": True,
+                                    "collecting": True,
+                                    "distributing": True,
+                                },
+                                "actorPortState": {
+                                    "activity": True,
+                                    "timeout": False,
+                                    "aggregation": True,
+                                    "synchronization": True,
+                                    "collecting": True,
+                                    "distributing": True,
+                                },
+                            }
+                        }
+                    }
+                },
+                "interface": "Ethernet5",
+                "orphanPorts": {},
+            }
+        ],
+        "inputs": {"interfaces": [{"name": "Ethernet5", "portchannel": "port-channel 5", "lacp_rate_fast": True}]},
+        "expected": {
+            "result": "failure",
+            "messages": [
+                "Interface: Ethernet5 Port-Channel: Port-Channel5 - Actor port details mismatch - Activity: True, Aggregation: True, "
+                "Synchronization: True, Collecting: True, Distributing: True, Timeout: False",
+                "Interface: Ethernet5 Port-Channel: Port-Channel5 - Partner port details mismatch - Activity: True, Aggregation: True, "
+                "Synchronization: True, Collecting: True, Distributing: True, Timeout: False",
             ],
         },
     },
diff --git a/tests/units/input_models/test_interfaces.py b/tests/units/input_models/test_interfaces.py
new file mode 100644
index 00000000..87d742d5
--- /dev/null
+++ b/tests/units/input_models/test_interfaces.py
@@ -0,0 +1,33 @@
+# 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.
+"""Tests for anta.input_models.interfaces.py."""
+
+# pylint: disable=C0302
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import pytest
+
+from anta.input_models.interfaces import InterfaceState
+
+if TYPE_CHECKING:
+    from anta.custom_types import Interface, PortChannelInterface
+
+
+class TestInterfaceState:
+    """Test anta.input_models.interfaces.InterfaceState."""
+
+    # pylint: disable=too-few-public-methods
+
+    @pytest.mark.parametrize(
+        ("name", "portchannel", "expected"),
+        [
+            pytest.param("Ethernet1", "Port-Channel42", "Interface: Ethernet1 Port-Channel: Port-Channel42", id="with port-channel"),
+            pytest.param("Ethernet1", None, "Interface: Ethernet1", id="no port-channel"),
+        ],
+    )
+    def test_valid__str__(self, name: Interface, portchannel: PortChannelInterface | None, expected: str) -> None:
+        """Test InterfaceState __str__."""
+        assert str(InterfaceState(name=name, portchannel=portchannel)) == expected