fix(anta.tests): Cleaning up Interfaces tests module (VerifyLACPInterfacesStatus)

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/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)}")

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

tests/units/anta_tests/test_interfaces.py

@@ -2510,6 +2510,43 @@
         "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 @@
         "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 @@
         "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 @@
         "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",
             ],
         },
     },

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