Repository: aristanetworks/anta (python)

Issue:
Add test case to verify Link Aggregation Control Protocol (LACP) functionality
**Verify LACP functionality using `show lacp interface <MLAG interface>`.**


**Pass/Fail Criteria**

1. Interface should be bundled in given port channel.
2. Aggregated link state should be `active, LongTimeout`( Indicates a longer interval for LACPDU transmission), `Aggregable, InSync` (Indicates actively participating in both transmit and receive operations), `Collecting and Distributing `(Indicates the interface is functioning correctly in aggregation group. i.e capable of receiving and sending the traffic.)

**Sample Output**

State: A = Active, P = Passive; S=ShortTimeout, L=LongTimeout;
       G = Aggregable, I = Individual; s+=InSync, s-=OutOfSync;
       C = Collecting, X = state machine expired,
       D = Distributing, d = default neighbor state

Port   Status  | Sys-id                                    Port# State   OperKey PortPri  Port#  State    OperKey  PortPriority  TimeoutMultiplier
------ --------|----------------------- ----- ------- ------- -------- ------ -------- -------- ------------- -----------------
Port Channel Port-Channel$MLAG_PO:
Et4 `Bundled` | 8000,c4-ca-2a-b1-b9-f3   221 `ALGs+CD ` 0x0064   32768    221  `ALGs+CD`  0x0064         32768

Pull request:
feat(anta): Added test case to verify Link Aggregation Control Protocol (LACP) functionality
# Description

**Verify LACP functionality using show lacp interface <MLAG interface>.**

- Interface should be bundled in given port channel.
- Aggregated link state should be active, LongTimeout( Indicates a longer interval for LACPDU transmission), Aggregable, InSync (Indicates actively participating in both transmit and receive operations), Collecting and Distributing (Indicates the interface is functioning correctly in aggregation group. i.e capable of receiving and sending the traffic.)

Fixes #762 

# 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`)
- [ ] I have made corresponding changes to the documentation
- [ ] 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/custom_types.py b/anta/custom_types.py
index 6747e766..c1e1f642 100644
--- a/anta/custom_types.py
+++ b/anta/custom_types.py
@@ -21,6 +21,8 @@ REGEXP_TYPE_EOS_INTERFACE = r"^(Dps|Ethernet|Fabric|Loopback|Management|Port-Cha
 """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 @@ VxlanSrcIntf = Annotated[
     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"]


Test patch:
diff --git a/anta/tests/interfaces.py b/anta/tests/interfaces.py
index dfbf15aa..9ff1cf35 100644
--- a/anta/tests/interfaces.py
+++ b/anta/tests/interfaces.py
@@ -15,7 +15,7 @@ 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, 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 @@ class VerifyInterfacesSpeed(AntaTest):
                         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)
diff --git a/examples/tests.yaml b/examples/tests.yaml
index f5a5ca46..bb7d3b0d 100644
--- a/examples/tests.yaml
+++ b/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:
diff --git a/tests/units/anta_tests/test_interfaces.py b/tests/units/anta_tests/test_interfaces.py
index b8cf493d..c38ac89f 100644
--- a/tests/units/anta_tests/test_interfaces.py
+++ b/tests/units/anta_tests/test_interfaces.py
@@ -21,6 +21,7 @@ from anta.tests.interfaces import (
     VerifyIpVirtualRouterMac,
     VerifyL2MTU,
     VerifyL3MTU,
+    VerifyLACPInterfacesStatus,
     VerifyLoopbackCount,
     VerifyPortChannels,
     VerifyStormControlDrops,
@@ -2441,4 +2442,127 @@ DATA: list[dict[str, Any]] = [
             ],
         },
     },
+    {
+        "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"
+            ],
+        },
+    },
 ]
diff --git a/tests/units/test_custom_types.py b/tests/units/test_custom_types.py
index 8119849a..e3dc09d2 100644
--- a/tests/units/test_custom_types.py
+++ b/tests/units/test_custom_types.py
@@ -17,6 +17,7 @@ import pytest
 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: