diff --git a/anta/decorators.py b/anta/decorators.py
index 08fbd6532..043162323 100644
--- a/anta/decorators.py
+++ b/anta/decorators.py
@@ -63,13 +63,15 @@ async def wrapper(*args: Any, **kwargs: Any) -> Any:
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 new_init(*args: Any, **kwargs: Any) -> None:
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 5036156de..9e33a2c54 100644
--- a/anta/input_models/interfaces.py
+++ b/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
diff --git a/anta/models.py b/anta/models.py
index 4cebd997a..69f305e2f 100644
--- a/anta/models.py
+++ b/anta/models.py
@@ -328,6 +328,8 @@ def test(self) -> None:
# 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/anta/tests/interfaces.py b/anta/tests/interfaces.py
index dc6938110..b87c394f3 100644
--- a/anta/tests/interfaces.py
+++ b/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)}")
diff --git a/anta/tests/stun.py b/anta/tests/stun.py
index 925abd147..2be13c4b2 100644
--- a/anta/tests/stun.py
+++ b/anta/tests/stun.py
@@ -91,15 +91,13 @@ def test(self) -> None:
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/docs/templates/python/material/class.html.jinja b/docs/templates/python/material/class.html.jinja
index cf016c92e..cbf9fac22 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" %}
+
+{% for arg in dec.value.arguments | selectattr("name", "equalto", "removal_in_version") | list %}
+
+{% endfor %}
+
+{% for arg in dec.value.arguments | selectattr("name", "equalto", "new_tests") | list %}
+Replaced with: {{ arg.value.elements | map("replace", "'", "", 1) | map("replace", "'", "", 1) | join(", ") | safe }}
+{% endfor %}
+{% endif %}
+{% endfor %}
+{% endblock bases %}
diff --git a/examples/tests.yaml b/examples/tests.yaml
index e14b40823..d307215f2 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_interfaces.py b/tests/units/anta_tests/test_interfaces.py
index ac0530881..271683b0c 100644
--- a/tests/units/anta_tests/test_interfaces.py
+++ b/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",
],
},
},
diff --git a/tests/units/input_models/test_interfaces.py b/tests/units/input_models/test_interfaces.py
new file mode 100644
index 000000000..87d742d53
--- /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