aristanetworks · geetanjalimanegslab · May 5, 2025 · May 5, 2025 · May 5, 2025 · May 13, 2025
diff --git a/anta/tests/interfaces.py b/anta/tests/interfaces.py
@@ -8,7 +8,8 @@
 from __future__ import annotations
 
 import re
-from typing import ClassVar, TypeVar
+from datetime import datetime, timezone
+from typing import Any, ClassVar, TypeVar
 
 from pydantic import Field, field_validator
 from pydantic_extra_types.mac_address import MacAddress
@@ -224,6 +225,7 @@ def test(self) -> None:
             # Verification is skipped if the interface is in the ignored interfaces list.
             if _is_interface_ignored(interface, self.inputs.ignored_interfaces):
                 continue
+
             counters_data = [f"{counter}: {value}" for counter, value in interface_data.items() if value > 0]
             if counters_data:
                 self.result.is_failure(f"Interface: {interface} - Non-zero discard counter(s): {', '.join(counters_data)}")
@@ -995,3 +997,142 @@ def test(self) -> None:
 
             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)}")
+
+
+class VerifyInterfacesCounters(AntaTest):
+    """Verifies the interfaces counter details.
+
+    Expected Results
+    ----------------
+    * Success: The test will pass if all interfaces have error counters below the expected threshold and the link status changes field matches the expected value,
+    * Failure: The test will fail if one or more interfaces have error counters below the expected threshold,
+     or if the link status changes field does not match the expected value (if provided).
+
+    Examples
+    --------
+    ```yaml
+    anta.tests.interfaces:
+      - VerifyInterfacesCounters:
+          ignored_interfaces:
+              - Management  # Ignore all Management interfaces
+              - Ethernet2.100
+              - Ethernet1/1
+          errors_threshold: 10
+          link_status_changes_threshold: 100
+    ```
+    """
+
+    categories: ClassVar[list[str]] = ["interfaces"]
+    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show interfaces", revision=1)]
+
+    class Input(AntaTest.Input):
+        """Input model for the VerifyInterfacesCounters test."""
+
+        ignored_interfaces: list[InterfaceType | Interface] = Field(default=["Dps", "Fabric", "Loopback", "Management", "Recirc-Channel", "Tunnel", "Vxlan"])
+        """A list of L3 interfaces or interfaces types like Loopback, Tunnel which will ignore all Loopback and Tunnel interfaces."""
+        errors_threshold: PositiveInteger = 0
+        """The max value for error threshold above which the test will fail."""
+        link_status_changes_threshold: PositiveInteger
+        """The max value for link status changes above which the test will fail."""
+
+    def _get_last_change_time_stamp(self, interface_time_stamp: float) -> int | None:
+        """Return the last change time stamp."""
+        now = datetime.now(timezone.utc)
+        if interface_time_stamp:
+            then = datetime.fromtimestamp(interface_time_stamp, tz=timezone.utc)
+            delta = now - then
+            return delta.days * 24 if delta.days < 1 else delta.days
+        return None
+
+    @AntaTest.anta_test
+    def test(self) -> None:
+        """Main test function for VerifyInterfacesCounters."""
+        self.result.is_success()
+        command_output = self.instance_commands[0].json_output
+
+        for interface, data in command_output["interfaces"].items():
+            # if the interface is in the ignored interfaces list or interface is sub-interface ignored the verification
+            if _is_interface_ignored(interface, self.inputs.ignored_interfaces) or "." in interface:
+                continue
+
+            last_state_timestamp = self._get_last_change_time_stamp(data.get("lastStatusChangeTimestamp"))
+            int_desc = data.get("description") if data.get("description") else None
+            error_counters = data.get("interfaceCounters", {})
+
+            # Verify the interface status
+            if data["lineProtocolStatus"] == "down" or data["lineProtocolStatus"] == "notPresent":
+                self.result.is_failure(
+                    f"Interface: {interface} Description: {int_desc} Downtime: {last_state_timestamp} - Incorrect state - Expected: up"
+                    f" Actual: {data['lineProtocolStatus']}"
+                )
+                continue
+
+            # Verify that link status changes are within the expected range
+            if error_counters.get("linkStatusChanges") > self.inputs.link_status_changes_threshold:
+                self.result.is_failure(
+                    f"Interface: {interface} Description: {int_desc} Uptime: {last_state_timestamp} - Link status changes mismatch -"
+                    f" Expected: < {self.inputs.link_status_changes_threshold} Actual: {error_counters['linkStatusChanges']}"
+                )
+
+            # Verify that interfaces input packet discard counters are non-zero
+            if error_counters.get("inDiscards") > self.inputs.errors_threshold:
+                self.result.is_failure(
+                    f"Interface: {interface} Description: {int_desc} Uptime: {last_state_timestamp} - Input packet discards counter(s) mismatch -"
+                    f" Expected: < {self.inputs.errors_threshold} Actual: {error_counters['inDiscards']}"
+                )
+
+            # Verify that interfaces output packet discard counters are non-zero
+            if error_counters.get("outDiscards") > self.inputs.errors_threshold:
+                self.result.is_failure(
+                    f"Interface: {interface} Description: {int_desc} Uptime: {last_state_timestamp} - Output packet discards counter(s) mismatch -"
+                    f" Expected: < {self.inputs.errors_threshold} Actual: {error_counters['outDiscards']}"
+                )
+
+            # Verify interface error counter details
+            self.verify_interface_error_counter_details(interface, error_counters, int_desc, last_state_timestamp)
+
+    def verify_interface_error_counter_details(self, interface: str, error_counters: dict[str, Any], int_desc: str | None, last_state_timestamp: int | None) -> None:
+        """Verify interface error counter details."""
+        # Delegate verification logic for the interface error counters
+        self.verify_input_output_error_counter_details(interface, error_counters, int_desc, last_state_timestamp)
+
+    def verify_input_output_error_counter_details(
+        self, interface: str, error_counters: dict[str, Any], int_desc: str | None, last_state_timestamp: int | None
+    ) -> None:
+        """Verify interface input and output error counter details."""
+        # Verify that input error counters are non-zero
+        input_counters_data = [
+            f"{counter}: {value}"
+            for counter, value in error_counters.get("inputErrorsDetail", {}).items()
+            if all([value > self.inputs.errors_threshold, counter != "rxPause"])
+        ]
+        if input_counters_data:
+            self.result.is_failure(
+                f"Interface: {interface} Description: {int_desc} Uptime: {last_state_timestamp} - Non-zero input error counter(s) - {', '.join(input_counters_data)}"
+            )
+
+        # Verify that output error counters are non-zero
+        output_counters_data = [
+            f"{counter}: {value}"
+            for counter, value in error_counters.get("outputErrorsDetail", {}).items()
+            if all([value > self.inputs.errors_threshold, counter != "txPause"])
+        ]
+        if output_counters_data:
+            self.result.is_failure(
+                f"Interface: {interface} Description: {int_desc} Uptime: {last_state_timestamp} - Non-zero output error counter(s) - "
+                f"{', '.join(output_counters_data)}"
+            )
+
+        # Verify that total input error counters are non-zero
+        if error_counters.get("totalInErrors") > self.inputs.errors_threshold:
+            self.result.is_failure(
+                f"Interface: {interface} Description: {int_desc} Uptime: {last_state_timestamp} - Total input error counter(s) mismatch -"
+                f" Expected: < {self.inputs.errors_threshold} Actual: {error_counters['totalInErrors']}"
+            )
+
+        # Verify that total output error counters are non-zero
+        if error_counters.get("totalOutErrors") > self.inputs.errors_threshold:
+            self.result.is_failure(
+                f"Interface: {interface} Description: {int_desc} Uptime: {last_state_timestamp} - Total output error counter(s) mismatch -"
+                f" Expected: < {self.inputs.errors_threshold} Actual: {error_counters['totalOutErrors']}"
+            )
diff --git a/examples/tests.yaml b/examples/tests.yaml
@@ -275,6 +275,14 @@ anta.tests.interfaces:
       ignored_interfaces:
         - Ethernet1
         - Port-Channel1
+  - VerifyInterfacesCounters:
+      # Verifies the interfaces counter details.
+      ignored_interfaces:
+          - Management  # Ignore all Management interfaces
+          - Ethernet2.100
+          - Ethernet1/1
+      errors_threshold: 10
+      link_status_changes_threshold: 100
   - VerifyInterfacesSpeed:
       # Verifies the speed, lanes, auto-negotiation status, and mode as full duplex for interfaces.
       interfaces: