Skip to content

Commit

Permalink
refactor(anta.tests)!: Update the pass/fail criteria for testcase Ver…
Browse files Browse the repository at this point in the history 
…ifyBFD (#504)
  • Loading branch information
@MaheshGSLAB
MaheshGSLAB authored Feb 23, 2024
1 parent 0871246 commit 4542750
Show file tree
Hide file tree
Showing 8 changed files with 331 additions and 150 deletions.
84 changes: 82 additions & 2 deletions anta/tests/bfd.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,10 +8,11 @@
# mypy: disable-error-code=attr-defined
from __future__ import annotations

from datetime import datetime
from ipaddress import IPv4Address
from typing import Any, List
from typing import Any, List, Optional

from pydantic import BaseModel
from pydantic import BaseModel, Field

from anta.custom_types import BfdInterval, BfdMultiplier
from anta.models import AntaCommand, AntaTest
Expand Down Expand Up @@ -153,3 +154,82 @@ def test(self) -> None:
self.result.is_success()
else:
self.result.is_failure(f"Following BFD peers are not configured or timers are not correct:\n{failures}")


class VerifyBFDPeersHealth(AntaTest):
"""
This class verifies the health of IPv4 BFD peers across all VRFs.
It checks that no BFD peer is in the down state and that the discriminator value of the remote system is not zero.
Optionally, it can also verify that BFD peers have not been down before a specified threshold of hours.
Expected results:
* Success: The test will pass if all IPv4 BFD peers are up, the discriminator value of each remote system is non-zero,
and the last downtime of each peer is above the defined threshold.
* Failure: The test will fail if any IPv4 BFD peer is down, the discriminator value of any remote system is zero,
or the last downtime of any peer is below the defined threshold.
"""

name = "VerifyBFDPeersHealth"
description = "Verifies the health of all IPv4 BFD peers."
categories = ["bfd"]
# revision 1 as later revision introduces additional nesting for type
commands = [AntaCommand(command="show bfd peers", revision=1), AntaCommand(command="show clock")]

class Input(AntaTest.Input):
"""
This class defines the input parameters of the test case.
"""

down_threshold: Optional[int] = Field(default=None, gt=0)
"""Optional down threshold in hours to check if a BFD peer was down before those hours or not."""

@AntaTest.anta_test
def test(self) -> None:
# Initialize failure strings
down_failures = []
up_failures = []

# Extract the current timestamp and command output
clock_output = self.instance_commands[1].json_output
current_timestamp = clock_output["utcTime"]
bfd_output = self.instance_commands[0].json_output

# set the initial result
self.result.is_success()

# Check if any IPv4 BFD peer is configured
ipv4_neighbors_exist = any(vrf_data["ipv4Neighbors"] for vrf_data in bfd_output["vrfs"].values())
if not ipv4_neighbors_exist:
self.result.is_failure("No IPv4 BFD peers are configured for any VRF.")
return

# Iterate over IPv4 BFD peers
for vrf, vrf_data in bfd_output["vrfs"].items():
for peer, neighbor_data in vrf_data["ipv4Neighbors"].items():
for peer_data in neighbor_data["peerStats"].values():
peer_status = peer_data["status"]
remote_disc = peer_data["remoteDisc"]
remote_disc_info = f" with remote disc {remote_disc}" if remote_disc == 0 else ""
last_down = peer_data["lastDown"]
hours_difference = (datetime.fromtimestamp(current_timestamp) - datetime.fromtimestamp(last_down)).total_seconds() / 3600

# Check if peer status is not up
if peer_status != "up":
down_failures.append(f"{peer} is {peer_status} in {vrf} VRF{remote_disc_info}.")

# Check if the last down is within the threshold
elif self.inputs.down_threshold and hours_difference < self.inputs.down_threshold:
up_failures.append(f"{peer} in {vrf} VRF was down {round(hours_difference)} hours ago{remote_disc_info}.")

# Check if remote disc is 0
elif remote_disc == 0:
up_failures.append(f"{peer} in {vrf} VRF has remote disc {remote_disc}.")

# Check if there are any failures
if down_failures:
down_failures_str = "\n".join(down_failures)
self.result.is_failure(f"Following BFD peers are not up:\n{down_failures_str}")
if up_failures:
up_failures_str = "\n".join(up_failures)
self.result.is_failure(f"\nFollowing BFD peers were down:\n{up_failures_str}")
25 changes: 0 additions & 25 deletions anta/tests/routing/generic.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -79,31 +79,6 @@ def test(self) -> None:
self.result.is_failure(f"routing-table has {total_routes} routes and not between min ({self.inputs.minimum}) and maximum ({self.inputs.maximum})")


class VerifyBFD(AntaTest):
"""
Verifies there is no BFD peer in down state (all VRF, IPv4 neighbors).
"""

name = "VerifyBFD"
description = "Verifies there is no BFD peer in down state (all VRF, IPv4 neighbors)."
categories = ["bfd"]
# revision 1 as later revision introduce additional nesting for type
commands = [AntaCommand(command="show bfd peers", revision=1)]

@AntaTest.anta_test
def test(self) -> None:
command_output = self.instance_commands[0].json_output
self.result.is_success()
for _, vrf_data in command_output["vrfs"].items():
for _, neighbor_data in vrf_data["ipv4Neighbors"].items():
for peer, peer_data in neighbor_data["peerStats"].items():
if (peer_status := peer_data["status"]) != "up":
failure_message = f"bfd state for peer '{peer}' is {peer_status} (expected up)."
if (peer_l3intf := peer_data.get("l3intf")) is not None and peer_l3intf != "":
failure_message += f" Interface: {peer_l3intf}."
self.result.is_failure(failure_message)


class VerifyRoutingTableEntry(AntaTest):
"""
This test verifies that the provided routes are present in the routing table of a specified VRF.
Expand Down
13 changes: 13 additions & 0 deletions docs/api/tests.bfd.md
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
<!--
~ 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.
-->

# ANTA catalog for bfd tests

::: anta.tests.bfd
options:
show_root_heading: false
show_root_toc_entry: false
merge_init_into_class: false
1 change: 1 addition & 0 deletions docs/api/tests.md
View file
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ This section describes all the available tests provided by ANTA package.


- [AAA](tests.aaa.md)
- [BFD](tests.bfd.md)
- [Configuration](tests.configuration.md)
- [Connectivity](tests.connectivity.md)
- [Field Notice](tests.field_notices.md)
Expand Down
2 changes: 2 additions & 0 deletions examples/tests.yaml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,8 @@ anta.tests.bfd:
tx_interval: 1200
rx_interval: 1200
multiplier: 3
- VerifyBFDPeersHealth:
down_threshold: 2

anta.tests.configuration:
- VerifyZeroTouch:
Expand Down
1 change: 1 addition & 0 deletions mkdocs.yml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -173,6 +173,7 @@ nav:
- Test Catalog Documentation:
- Overview: api/tests.md
- AAA: api/tests.aaa.md
- BFD: api/tests.bfd.md
- Configuration: api/tests.configuration.md
- Connectivity: api/tests.connectivity.md
- Field Notices: api/tests.field_notices.md
Expand Down
123 changes: 1 addition & 122 deletions tests/units/anta_tests/routing/test_generic.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@

from typing import Any

from anta.tests.routing.generic import VerifyBFD, VerifyRoutingProtocolModel, VerifyRoutingTableEntry, VerifyRoutingTableSize
from anta.tests.routing.generic import VerifyRoutingProtocolModel, VerifyRoutingTableEntry, VerifyRoutingTableSize
from tests.lib.anta import test # noqa: F401; pylint: disable=W0611

DATA: list[dict[str, Any]] = [
Expand Down Expand Up @@ -77,127 +77,6 @@
"messages": ["Minimum 666 is greater than maximum 42"],
},
},
{
"name": "success-no-peer",
"test": VerifyBFD,
"eos_data": [{"vrfs": {}}],
"inputs": None,
"expected": {"result": "success"},
},
{
"name": "success-peers-up",
"test": VerifyBFD,
"eos_data": [
{
"vrfs": {
"default": {
"ipv6Neighbors": {},
"ipv4Neighbors": {
"7.7.7.7": {
"peerStats": {
"": {
"status": "up",
"authType": "None",
"kernelIfIndex": 0,
"lastDiag": "diagNone",
"authProfileName": "",
"lastUp": 1683288421.669188,
"remoteDisc": 345332116,
"sessType": "sessionTypeMultihop",
"localDisc": 1654273918,
"lastDown": 0.0,
"l3intf": "",
"tunnelId": 0,
}
}
},
"10.3.0.1": {
"peerStats": {
"Ethernet1": {
"status": "up",
"authType": "None",
"kernelIfIndex": 11,
"lastDiag": "diagNone",
"authProfileName": "",
"lastUp": 1683288900.004889,
"remoteDisc": 1017672851,
"sessType": "sessionTypeNormal",
"localDisc": 4269977256,
"lastDown": 0.0,
"l3intf": "Ethernet1",
"tunnelId": 0,
}
}
},
},
"ipv4ReflectorNeighbors": {},
"ipv6ReflectorNeighbors": {},
"ipv6InitiatorNeighbors": {},
"ipv4InitiatorNeighbors": {},
}
}
}
],
"inputs": None,
"expected": {"result": "success"},
},
{
"name": "failure",
"test": VerifyBFD,
"eos_data": [
{
"vrfs": {
"default": {
"ipv6Neighbors": {},
"ipv4Neighbors": {
"7.7.7.7": {
"peerStats": {
"": {
"status": "down",
"authType": "None",
"kernelIfIndex": 0,
"lastDiag": "diagNone",
"authProfileName": "",
"lastUp": 1683288421.669188,
"remoteDisc": 345332116,
"sessType": "sessionTypeMultihop",
"localDisc": 1654273918,
"lastDown": 0.0,
"l3intf": "",
"tunnelId": 0,
}
}
},
"10.3.0.1": {
"peerStats": {
"Ethernet1": {
"status": "up",
"authType": "None",
"kernelIfIndex": 11,
"lastDiag": "diagNone",
"authProfileName": "",
"lastUp": 1683288900.004889,
"remoteDisc": 1017672851,
"sessType": "sessionTypeNormal",
"localDisc": 4269977256,
"lastDown": 0.0,
"l3intf": "Ethernet1",
"tunnelId": 0,
}
}
},
},
"ipv4ReflectorNeighbors": {},
"ipv6ReflectorNeighbors": {},
"ipv6InitiatorNeighbors": {},
"ipv4InitiatorNeighbors": {},
}
}
}
],
"inputs": None,
"expected": {"result": "failure", "messages": ["bfd state for peer '' is down (expected up)."]},
},
{
"name": "success",
"test": VerifyRoutingTableEntry,
Expand Down
Loading

0 comments on commit 4542750

Please sign in to comment.