Skip to content

Commit

Permalink
refactor(anta.tests): Nicer result failure messages interface(first 6…
Browse files Browse the repository at this point in the history 
…) test module 
  • Loading branch information
@geetanjalimanegslab
geetanjalimanegslab committed Feb 18, 2025
1 parent f3ccdfa commit 8e572a2
Show file tree
Hide file tree
Showing 2 changed files with 65 additions and 72 deletions.
84 changes: 38 additions & 46 deletions anta/tests/interfaces.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@

import re
from ipaddress import IPv4Interface
from typing import Any, ClassVar, TypeVar
from typing import ClassVar, TypeVar

from pydantic import BaseModel, Field, field_validator
from pydantic_extra_types.mac_address import MacAddress
Expand Down Expand Up @@ -63,32 +63,32 @@ class Input(AntaTest.Input):
@AntaTest.anta_test
def test(self) -> None:
"""Main test function for VerifyInterfaceUtilization."""
self.result.is_success()
duplex_full = "duplexFull"
failed_interfaces: dict[str, dict[str, float]] = {}
rates = self.instance_commands[0].json_output
interfaces = self.instance_commands[1].json_output

interface_data = []
for intf, rate in rates["interfaces"].items():
# The utilization logic has been implemented for full-duplex interfaces only
if ((duplex := (interface := interfaces["interfaces"][intf]).get("duplex", None)) is not None and duplex != duplex_full) or (
(members := interface.get("memberInterfaces", None)) is not None and any(stats["duplex"] != duplex_full for stats in members.values())
):
self.result.is_failure(f"Interface {intf} or one of its member interfaces is not Full-Duplex. VerifyInterfaceUtilization has not been implemented.")
return
# If interface is not full-duplex, test fails.
if all([(duplex := (interface := interfaces["interfaces"][intf]).get("duplex", None)), duplex != duplex_full]):
self.result.is_failure(f"Interface: {intf} - Interface utilization has not been implemented")
continue

# If any of the member-interface is not full-duplex, test fails.
if (members := interface.get("memberInterfaces", None)) is not None:
interface_data = [(member_interface, state) for member_interface, stats in members.items() if (state := stats["duplex"]) != duplex_full]
for member_interface in interface_data:
self.result.is_failure(f"Interface: {intf} MemberInterface: {member_interface[0]} - Not Full-Duplex - Actual: {member_interface[1]}")

if (bandwidth := interfaces["interfaces"][intf]["bandwidth"]) == 0:
self.logger.debug("Interface %s has been ignored due to null bandwidth value", intf)
continue

# If one or more interfaces have a usage above the threshold, test fails.
for bps_rate in ("inBpsRate", "outBpsRate"):
usage = rate[bps_rate] / bandwidth * 100
if usage > self.inputs.threshold:
failed_interfaces.setdefault(intf, {})[bps_rate] = usage

if not failed_interfaces:
self.result.is_success()
else:
self.result.is_failure(f"The following interfaces have a usage > {self.inputs.threshold}%: {failed_interfaces}")
self.result.is_failure(f"Interface: {intf} BPS Rate: {bps_rate} - Greater threshold usage - Expected: {self.inputs.threshold}% Actual: {usage}%")


class VerifyInterfaceErrors(AntaTest):
Expand All @@ -113,15 +113,12 @@ class VerifyInterfaceErrors(AntaTest):
@AntaTest.anta_test
def test(self) -> None:
"""Main test function for VerifyInterfaceErrors."""
self.result.is_success()
command_output = self.instance_commands[0].json_output
wrong_interfaces: list[dict[str, dict[str, int]]] = []
for interface, counters in command_output["interfaceErrorCounters"].items():
if any(value > 0 for value in counters.values()) and all(interface not in wrong_interface for wrong_interface in wrong_interfaces):
wrong_interfaces.append({interface: counters})
if not wrong_interfaces:
self.result.is_success()
else:
self.result.is_failure(f"The following interface(s) have non-zero error counters: {wrong_interfaces}")
counters_data = [f"{counter}: {value}" for counter, value in counters.items() if value > 0]
if counters_data:
self.result.is_failure(f"Interface: {interface} - Non-zero error counter(s) - {', '.join(counters_data)}")


class VerifyInterfaceDiscards(AntaTest):
Expand All @@ -146,14 +143,12 @@ class VerifyInterfaceDiscards(AntaTest):
@AntaTest.anta_test
def test(self) -> None:
"""Main test function for VerifyInterfaceDiscards."""
self.result.is_success()
command_output = self.instance_commands[0].json_output
wrong_interfaces: list[dict[str, dict[str, int]]] = []
for interface, outer_v in command_output["interfaces"].items():
wrong_interfaces.extend({interface: outer_v} for value in outer_v.values() if value > 0)
if not wrong_interfaces:
self.result.is_success()
else:
self.result.is_failure(f"The following interfaces have non 0 discard counter(s): {wrong_interfaces}")
for interface, interface_data in command_output["interfaces"].items():
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)}")


class VerifyInterfaceErrDisabled(AntaTest):
Expand All @@ -178,12 +173,11 @@ class VerifyInterfaceErrDisabled(AntaTest):
@AntaTest.anta_test
def test(self) -> None:
"""Main test function for VerifyInterfaceErrDisabled."""
self.result.is_success()
command_output = self.instance_commands[0].json_output
errdisabled_interfaces = [interface for interface, value in command_output["interfaceStatuses"].items() if value["linkStatus"] == "errdisabled"]
if errdisabled_interfaces:
self.result.is_failure(f"The following interfaces are in error disabled state: {errdisabled_interfaces}")
else:
self.result.is_success()
for interface, value in command_output["interfaceStatuses"].items():
if value["linkStatus"] == "errdisabled":
self.result.is_failure(f"Interface: {interface} - Link status Error disabled")


class VerifyInterfacesStatus(AntaTest):
Expand Down Expand Up @@ -255,16 +249,16 @@ def test(self) -> None:

# If line protocol status is provided, prioritize checking against both status and line protocol status
if interface.line_protocol_status:
if interface.status != status or interface.line_protocol_status != proto:
if any([interface.status != status, interface.line_protocol_status != proto]):
actual_state = f"Expected: {interface.status}/{interface.line_protocol_status}, Actual: {status}/{proto}"
self.result.is_failure(f"{interface.name} - {actual_state}")
self.result.is_failure(f"{interface.name} - Status mismatch - {actual_state}")

# If line protocol status is not provided and interface status is "up", expect both status and proto to be "up"
# If interface status is not "up", check only the interface status without considering line protocol status
elif interface.status == "up" and (status != "up" or proto != "up"):
self.result.is_failure(f"{interface.name} - Expected: up/up, Actual: {status}/{proto}")
elif all([interface.status == "up", status != "up" or proto != "up"]):
self.result.is_failure(f"{interface.name} - Status mismatch - Expected: up/up, Actual: {status}/{proto}")
elif interface.status != status:
self.result.is_failure(f"{interface.name} - Expected: {interface.status}, Actual: {status}")
self.result.is_failure(f"{interface.name} - Status mismatch - Expected: {interface.status}, Actual: {status}")


class VerifyStormControlDrops(AntaTest):
Expand All @@ -291,16 +285,14 @@ class VerifyStormControlDrops(AntaTest):
def test(self) -> None:
"""Main test function for VerifyStormControlDrops."""
command_output = self.instance_commands[0].json_output
storm_controlled_interfaces: dict[str, dict[str, Any]] = {}
storm_controlled_interfaces = []
self.result.is_success()

for interface, interface_dict in command_output["interfaces"].items():
for traffic_type, traffic_type_dict in interface_dict["trafficTypes"].items():
if "drop" in traffic_type_dict and traffic_type_dict["drop"] != 0:
storm_controlled_interface_dict = storm_controlled_interfaces.setdefault(interface, {})
storm_controlled_interface_dict.update({traffic_type: traffic_type_dict["drop"]})
if not storm_controlled_interfaces:
self.result.is_success()
else:
self.result.is_failure(f"The following interfaces have none 0 storm-control drop counters {storm_controlled_interfaces}")
storm_controlled_interfaces.append(f"{traffic_type}: {traffic_type_dict['drop']}")
self.result.is_failure(f"Interface: {interface} - Non-zero storm-control drop counter(s) - {', '.join(storm_controlled_interfaces)}")


class VerifyPortChannels(AntaTest):
Expand Down
53 changes: 27 additions & 26 deletions tests/units/anta_tests/test_interfaces.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -508,7 +508,10 @@
"inputs": {"threshold": 3.0},
"expected": {
"result": "failure",
"messages": ["The following interfaces have a usage > 3.0%: {'Ethernet1/1': {'inBpsRate': 10.0}, 'Port-Channel31': {'outBpsRate': 5.0}}"],
"messages": [
"Interface: Ethernet1/1 BPS Rate: inBpsRate - Greater threshold usage - Expected: 3.0% Actual: 10.0%",
"Interface: Port-Channel31 BPS Rate: outBpsRate - Greater threshold usage - Expected: 3.0% Actual: 5.0%",
],
},
},
{
Expand Down Expand Up @@ -653,7 +656,7 @@
"inputs": {"threshold": 70.0},
"expected": {
"result": "failure",
"messages": ["Interface Ethernet1/1 or one of its member interfaces is not Full-Duplex. VerifyInterfaceUtilization has not been implemented."],
"messages": ["Interface: Ethernet1/1 - Interface utilization has not been implemented"],
},
},
{
Expand Down Expand Up @@ -787,7 +790,7 @@
},
"memberInterfaces": {
"Ethernet3/1": {"bandwidth": 1000000000, "duplex": "duplexHalf"},
"Ethernet4/1": {"bandwidth": 1000000000, "duplex": "duplexFull"},
"Ethernet4/1": {"bandwidth": 1000000000, "duplex": "duplexHalf"},
},
"fallbackEnabled": False,
"fallbackEnabledType": "fallbackNone",
Expand All @@ -798,7 +801,10 @@
"inputs": {"threshold": 70.0},
"expected": {
"result": "failure",
"messages": ["Interface Port-Channel31 or one of its member interfaces is not Full-Duplex. VerifyInterfaceUtilization has not been implemented."],
"messages": [
"Interface: Port-Channel31 MemberInterface: Ethernet3/1 - Not Full-Duplex - Actual: duplexHalf",
"Interface: Port-Channel31 MemberInterface: Ethernet4/1 - Not Full-Duplex - Actual: duplexHalf",
],
},
},
{
Expand Down Expand Up @@ -830,9 +836,8 @@
"expected": {
"result": "failure",
"messages": [
"The following interface(s) have non-zero error counters: [{'Ethernet1': {'inErrors': 42, 'frameTooLongs': 0, 'outErrors': 0, 'frameTooShorts': 0,"
" 'fcsErrors': 0, 'alignmentErrors': 0, 'symbolErrors': 0}}, {'Ethernet6': {'inErrors': 0, 'frameTooLongs': 0, 'outErrors': 0, 'frameTooShorts':"
" 0, 'fcsErrors': 0, 'alignmentErrors': 666, 'symbolErrors': 0}}]",
"Interface: Ethernet1 - Non-zero error counter(s) - inErrors: 42",
"Interface: Ethernet6 - Non-zero error counter(s) - alignmentErrors: 666",
],
},
},
Expand All @@ -851,9 +856,8 @@
"expected": {
"result": "failure",
"messages": [
"The following interface(s) have non-zero error counters: [{'Ethernet1': {'inErrors': 42, 'frameTooLongs': 0, 'outErrors': 10, 'frameTooShorts': 0,"
" 'fcsErrors': 0, 'alignmentErrors': 0, 'symbolErrors': 0}}, {'Ethernet6': {'inErrors': 0, 'frameTooLongs': 0, 'outErrors': 0, 'frameTooShorts':"
" 0, 'fcsErrors': 0, 'alignmentErrors': 6, 'symbolErrors': 10}}]",
"Interface: Ethernet1 - Non-zero error counter(s) - inErrors: 42, outErrors: 10",
"Interface: Ethernet6 - Non-zero error counter(s) - alignmentErrors: 6, symbolErrors: 10",
],
},
},
Expand All @@ -870,10 +874,7 @@
"inputs": None,
"expected": {
"result": "failure",
"messages": [
"The following interface(s) have non-zero error counters: [{'Ethernet1': {'inErrors': 42, 'frameTooLongs': 0, 'outErrors': 2, 'frameTooShorts': 0,"
" 'fcsErrors': 0, 'alignmentErrors': 0, 'symbolErrors': 0}}]",
],
"messages": ["Interface: Ethernet1 - Non-zero error counter(s) - inErrors: 42, outErrors: 2"],
},
},
{
Expand Down Expand Up @@ -909,8 +910,8 @@
"expected": {
"result": "failure",
"messages": [
"The following interfaces have non 0 discard counter(s): [{'Ethernet2': {'outDiscards': 42, 'inDiscards': 0}},"
" {'Ethernet1': {'outDiscards': 0, 'inDiscards': 42}}]",
"Interface: Ethernet2 - Non-zero discard counter(s): outDiscards: 42",
"Interface: Ethernet1 - Non-zero discard counter(s): inDiscards: 42",
],
},
},
Expand Down Expand Up @@ -948,7 +949,7 @@
},
],
"inputs": None,
"expected": {"result": "failure", "messages": ["The following interfaces are in error disabled state: ['Management1', 'Ethernet8']"]},
"expected": {"result": "failure", "messages": ["Interface: Management1 - Link status Error disabled", "Interface: Ethernet8 - Link status Error disabled"]},
},
{
"name": "success",
Expand Down Expand Up @@ -1126,7 +1127,7 @@
"inputs": {"interfaces": [{"name": "Ethernet2", "status": "up"}, {"name": "Ethernet8", "status": "up"}, {"name": "Ethernet3", "status": "up"}]},
"expected": {
"result": "failure",
"messages": ["Ethernet8 - Expected: up/up, Actual: down/down"],
"messages": ["Ethernet8 - Status mismatch - Expected: up/up, Actual: down/down"],
},
},
{
Expand All @@ -1150,7 +1151,7 @@
},
"expected": {
"result": "failure",
"messages": ["Ethernet8 - Expected: up/up, Actual: up/down"],
"messages": ["Ethernet8 - Status mismatch - Expected: up/up, Actual: up/down"],
},
},
{
Expand All @@ -1166,7 +1167,7 @@
"inputs": {"interfaces": [{"name": "PortChannel100", "status": "up"}]},
"expected": {
"result": "failure",
"messages": ["Port-Channel100 - Expected: up/up, Actual: down/lowerLayerDown"],
"messages": ["Port-Channel100 - Status mismatch - Expected: up/up, Actual: down/lowerLayerDown"],
},
},
{
Expand All @@ -1191,8 +1192,8 @@
"expected": {
"result": "failure",
"messages": [
"Ethernet2 - Expected: up/down, Actual: up/unknown",
"Ethernet8 - Expected: up/up, Actual: up/down",
"Ethernet2 - Status mismatch - Expected: up/down, Actual: up/unknown",
"Ethernet8 - Status mismatch - Expected: up/up, Actual: up/down",
],
},
},
Expand All @@ -1218,9 +1219,9 @@
"expected": {
"result": "failure",
"messages": [
"Ethernet2 - Expected: down, Actual: up",
"Ethernet8 - Expected: down, Actual: up",
"Ethernet3 - Expected: down, Actual: up",
"Ethernet2 - Status mismatch - Expected: down, Actual: up",
"Ethernet8 - Status mismatch - Expected: down, Actual: up",
"Ethernet3 - Status mismatch - Expected: down, Actual: up",
],
},
},
Expand Down Expand Up @@ -1260,7 +1261,7 @@
},
],
"inputs": None,
"expected": {"result": "failure", "messages": ["The following interfaces have none 0 storm-control drop counters {'Ethernet1': {'broadcast': 666}}"]},
"expected": {"result": "failure", "messages": ["Interface: Ethernet1 - Non-zero storm-control drop counter(s) - broadcast: 666"]},
},
{
"name": "success",
Expand Down

0 comments on commit 8e572a2

Please sign in to comment.