diff --git a/anta/custom_types.py b/anta/custom_types.py index 6747e7663..c1e1f6428 100644 --- a/anta/custom_types.py +++ b/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"] diff --git a/anta/tests/interfaces.py b/anta/tests/interfaces.py index dfbf15aa6..9ff1cf357 100644 --- a/anta/tests/interfaces.py +++ b/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) diff --git a/examples/tests.yaml b/examples/tests.yaml index f5a5ca46b..bb7d3b0d6 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 b8cf493da..c38ac89f2 100644 --- a/tests/units/anta_tests/test_interfaces.py +++ b/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" + ], + }, + }, ] diff --git a/tests/units/test_custom_types.py b/tests/units/test_custom_types.py index 8119849a6..e3dc09d25 100644 --- a/tests/units/test_custom_types.py +++ b/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: