feat(anta): Added test case to verify Link Aggregation Control Protocol (LACP) functionality

anta/custom_types.py

@@ -21,6 +21,8 @@
 """Match EOS interface types like Ethernet1/1, Vlan1, Loopback1, etc."""
 REGEXP_TYPE_VXLAN_SRC_INTERFACE = r"^(Loopback)([0-9]|[1-9][0-9]{1,2}|[1-7][0-9]{3}|8[01][0-9]{2}|819[01])$"
 """Match Vxlan source interface like Loopback10."""
+REGEX_TYPE_PORTCHANNEL = r"^Port-Channel[0-9]{1,6}$"
+"""Match Port Channel interface like Port-Channel5."""
 REGEXP_TYPE_HOSTNAME = r"^(([a-zA-Z0-9]|[a-zA-Z0-9][a-zA-Z0-9\-]*[a-zA-Z0-9])\.)*([A-Za-z0-9]|[A-Za-z0-9][A-Za-z0-9\-]*[A-Za-z0-9])$"
 """Match hostname like `my-hostname`, `my-hostname-1`, `my-hostname-1-2`."""
 
@@ -135,6 +137,12 @@ def validate_regex(value: str) -> str:
     BeforeValidator(interface_autocomplete),
     BeforeValidator(interface_case_sensitivity),
 ]
+PortChannelInterface = Annotated[
+    str,
+    Field(pattern=REGEX_TYPE_PORTCHANNEL),
+    BeforeValidator(interface_autocomplete),
+    BeforeValidator(interface_case_sensitivity),
+]
 Afi = Literal["ipv4", "ipv6", "vpn-ipv4", "vpn-ipv6", "evpn", "rt-membership", "path-selection", "link-state"]
 Safi = Literal["unicast", "multicast", "labeled-unicast", "sr-te"]
 EncryptionAlgorithm = Literal["RSA", "ECDSA"]

anta/tests/interfaces.py

@@ -15,7 +15,7 @@
 from pydantic_extra_types.mac_address import MacAddress
 
 from anta import GITHUB_SUGGESTION
-from anta.custom_types import EthernetInterface, Interface, Percent, PositiveInteger
+from anta.custom_types import EthernetInterface, Interface, Percent, PortChannelInterface, PositiveInteger
 from anta.decorators import skip_on_platforms
 from anta.models import AntaCommand, AntaTemplate, AntaTest
 from anta.tools import custom_division, get_failed_logs, get_item, get_value
@@ -883,3 +883,107 @@ def test(self) -> None:
                         output["speed"] = f"{custom_division(output['speed'], BPS_GBPS_CONVERSIONS)}Gbps"
                 failed_log = get_failed_logs(expected_interface_output, actual_interface_output)
                 self.result.is_failure(f"For interface {intf}:{failed_log}\n")
+
+
+class VerifyLACPInterfacesStatus(AntaTest):
+    """Verifies the Link Aggregation Control Protocol (LACP) status of the provided interfaces.
+
+    - 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.)
+
+    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.
+
+    Examples
+    --------
+    ```yaml
+    anta.tests.interfaces:
+      - VerifyLACPInterfacesStatus:
+          interfaces:
+            - name: Ethernet1
+              portchannel: Port-Channel100
+    ```
+    """
+
+    name = "VerifyLACPInterfacesStatus"
+    description = "Verifies the Link Aggregation Control Protocol(LACP) status of the provided interfaces."
+    categories: ClassVar[list[str]] = ["interfaces"]
+    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaTemplate(template="show lacp interface {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]
+
+    @AntaTest.anta_test
+    def test(self) -> None:
+        """Main test function for VerifyLACPInterfacesStatus."""
+        self.result.is_success()
+
+        # 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
+
+            # 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}'.")
+                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)
+                continue
+
+            # Collecting actor and partner port details
+            actor_port_details = interface_details.get("actorPortState", {})
+            partner_port_details = interface_details.get("partnerPortState", {})
+
+            # Collecting actual interface details
+            actual_interface_output = {
+                "actor_port_details": {param: actor_port_details.get(param, "NotFound") for param in member_port_details},
+                "partner_port_details": {param: partner_port_details.get(param, "NotFound") for param in member_port_details},
+            }
+
+            # 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}
+
+            # 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)

examples/tests.yaml

@@ -221,6 +221,12 @@ anta.tests.interfaces:
         - name: Eth2
           auto: False
           speed: 2.5
+  - VerifyLACPInterfacesStatus:
+      interfaces:
+        - name: Ethernet5
+          portchannel: Port-Channel5
+        - name: Ethernet6
+          portchannel: Port-Channel5
 
 anta.tests.lanz:
   - VerifyLANZ:

tests/units/anta_tests/test_interfaces.py

@@ -21,6 +21,7 @@
     VerifyIpVirtualRouterMac,
     VerifyL2MTU,
     VerifyL3MTU,
+    VerifyLACPInterfacesStatus,
     VerifyLoopbackCount,
     VerifyPortChannels,
     VerifyStormControlDrops,
@@ -2441,4 +2442,127 @@
             ],
         },
     },
+    {
+        "name": "success",
+        "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-Channel5"}]},
+        "expected": {"result": "success"},
+    },
+    {
+        "name": "failure-not-bundled",
+        "test": VerifyLACPInterfacesStatus,
+        "eos_data": [
+            {
+                "portChannels": {
+                    "Port-Channel5": {
+                        "interfaces": {
+                            "Ethernet5": {
+                                "actorPortStatus": "No Aggregate",
+                            }
+                        }
+                    }
+                },
+                "interface": "Ethernet5",
+                "orphanPorts": {},
+            }
+        ],
+        "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"],
+        },
+    },
+    {
+        "name": "failure-no-details-found",
+        "test": VerifyLACPInterfacesStatus,
+        "eos_data": [
+            {
+                "portChannels": {"Port-Channel5": {"interfaces": {}}},
+            }
+        ],
+        "inputs": {"interfaces": [{"name": "Ethernet5", "portchannel": "Po 5"}]},
+        "expected": {
+            "result": "failure",
+            "messages": ["Interface 'Ethernet5' is not configured to be a member of LACP 'Port-Channel5'."],
+        },
+    },
+    {
+        "name": "failure-lacp-params",
+        "test": VerifyLACPInterfacesStatus,
+        "eos_data": [
+            {
+                "portChannels": {
+                    "Port-Channel5": {
+                        "interfaces": {
+                            "Ethernet5": {
+                                "actorPortStatus": "bundled",
+                                "partnerPortState": {
+                                    "activity": False,
+                                    "timeout": False,
+                                    "aggregation": False,
+                                    "synchronization": False,
+                                    "collecting": True,
+                                    "distributing": True,
+                                },
+                                "actorPortState": {
+                                    "activity": False,
+                                    "timeout": False,
+                                    "aggregation": False,
+                                    "synchronization": False,
+                                    "collecting": True,
+                                    "distributing": True,
+                                },
+                            }
+                        }
+                    }
+                },
+                "interface": "Ethernet5",
+                "orphanPorts": {},
+            }
+        ],
+        "inputs": {"interfaces": [{"name": "Ethernet5", "portchannel": "port-channel 5"}]},
+        "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"
+            ],
+        },
+    },
 ]

tests/units/test_custom_types.py

@@ -17,6 +17,7 @@
 from anta.custom_types import (
     REGEX_BGP_IPV4_MPLS_VPN,
     REGEX_BGP_IPV4_UNICAST,
+    REGEX_TYPE_PORTCHANNEL,
     REGEXP_BGP_IPV4_MPLS_LABELS,
     REGEXP_BGP_L2VPN_AFI,
     REGEXP_EOS_BLACKLIST_CMDS,
@@ -140,6 +141,22 @@ def test_regexp_type_vxlan_src_interface() -> None:
     assert re.match(REGEXP_TYPE_VXLAN_SRC_INTERFACE, "Loopback9000") is None
 
 
+def test_regexp_type_portchannel() -> None:
+    """Test REGEX_TYPE_PORTCHANNEL."""
+    # Test strings that should match the pattern
+    assert re.match(REGEX_TYPE_PORTCHANNEL, "Port-Channel5") is not None
+    assert re.match(REGEX_TYPE_PORTCHANNEL, "Port-Channel100") is not None
+    assert re.match(REGEX_TYPE_PORTCHANNEL, "Port-Channel999") is not None
+    assert re.match(REGEX_TYPE_PORTCHANNEL, "Port-Channel1000") is not None
+
+    # Test strings that should not match the pattern
+    assert re.match(REGEX_TYPE_PORTCHANNEL, "Port-Channel") is None
+    assert re.match(REGEX_TYPE_PORTCHANNEL, "Port_Channel") is None
+    assert re.match(REGEX_TYPE_PORTCHANNEL, "Port_Channel1000") is None
+    assert re.match(REGEX_TYPE_PORTCHANNEL, "Port_Channel5/1") is None
+    assert re.match(REGEX_TYPE_PORTCHANNEL, "Port-Channel-100") is None
+
+
 def test_regexp_type_hostname() -> None:
     """Test REGEXP_TYPE_HOSTNAME."""
     # Test strings that should match the pattern
@@ -200,6 +217,8 @@ def test_interface_autocomplete_success() -> None:
     assert interface_autocomplete("eth2") == "Ethernet2"
     assert interface_autocomplete("po3") == "Port-Channel3"
     assert interface_autocomplete("lo4") == "Loopback4"
+    assert interface_autocomplete("Po1000") == "Port-Channel1000"
+    assert interface_autocomplete("Po 1000") == "Port-Channel1000"
 
 
 def test_interface_autocomplete_no_alias() -> None: