diff --git a/anta/custom_types.py b/anta/custom_types.py index 56c213977..153fd7011 100644 --- a/anta/custom_types.py +++ b/anta/custom_types.py @@ -198,3 +198,4 @@ def validate_regex(value: str) -> str: "prefixRtMembershipDroppedMaxRouteLimitViolated", ] BgpUpdateError = Literal["inUpdErrWithdraw", "inUpdErrIgnore", "inUpdErrDisableAfiSafi", "disabledAfiSafi", "lastUpdErrTime"] +BfdProtocol = Literal["bgp", "isis", "lag", "ospf", "ospfv3", "pim", "route-input", "static-bfd", "static-route", "vrrp", "vxlan"] diff --git a/anta/tests/bfd.py b/anta/tests/bfd.py index f19e9cc92..0b171a6d2 100644 --- a/anta/tests/bfd.py +++ b/anta/tests/bfd.py @@ -13,7 +13,7 @@ from pydantic import BaseModel, Field -from anta.custom_types import BfdInterval, BfdMultiplier +from anta.custom_types import BfdInterval, BfdMultiplier, BfdProtocol from anta.models import AntaCommand, AntaTest from anta.tools import get_value @@ -45,7 +45,7 @@ class VerifyBFDSpecificPeers(AntaTest): name = "VerifyBFDSpecificPeers" description = "Verifies the IPv4 BFD peer's sessions and remote disc in the specified VRF." categories: ClassVar[list[str]] = ["bfd"] - commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show bfd peers", revision=4)] + commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show bfd peers", revision=1)] class Input(AntaTest.Input): """Input model for the VerifyBFDSpecificPeers test.""" @@ -126,7 +126,7 @@ class VerifyBFDPeersIntervals(AntaTest): name = "VerifyBFDPeersIntervals" description = "Verifies the timers of the IPv4 BFD peers in the specified VRF." categories: ClassVar[list[str]] = ["bfd"] - commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show bfd peers detail", revision=4)] + commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show bfd peers detail", revision=1)] class Input(AntaTest.Input): """Input model for the VerifyBFDPeersIntervals test.""" @@ -285,3 +285,79 @@ def test(self) -> None: if up_failures: up_failures_str = "\n".join(up_failures) self.result.is_failure(f"\nFollowing BFD peers were down:\n{up_failures_str}") + + +class VerifyBFDPeersRegProtocols(AntaTest): + """Verifies that IPv4 BFD peer(s) have the specified protocol(s) registered. + + Expected Results + ---------------- + * Success: The test will pass if IPv4 BFD peers are registered with the specified protocol(s). + * Failure: The test will fail if IPv4 BFD peers are not found or the specified protocol(s) are not registered for the BFD peer(s). + + Examples + -------- + ```yaml + anta.tests.bfd: + - VerifyBFDPeersRegProtocols: + bfd_peers: + - peer_address: 192.0.255.7 + vrf: default + protocols: + - bgp + ``` + """ + + name = "VerifyBFDPeersRegProtocols" + description = "Verifies that IPv4 BFD peer(s) have the specified protocol(s) registered." + categories: ClassVar[list[str]] = ["bfd"] + commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show bfd peers detail", revision=1)] + + class Input(AntaTest.Input): + """Input model for the VerifyBFDPeersRegProtocols test.""" + + bfd_peers: list[BFDPeer] + """List of IPv4 BFD peers.""" + + class BFDPeer(BaseModel): + """Model for an IPv4 BFD peer.""" + + peer_address: IPv4Address + """IPv4 address of a BFD peer.""" + vrf: str = "default" + """Optional VRF for BFD peer. If not provided, it defaults to `default`.""" + protocols: list[BfdProtocol] + """List of protocols to be verified.""" + + @AntaTest.anta_test + def test(self) -> None: + """Main test function for VerifyBFDPeersRegProtocols.""" + # Initialize failure messages + failures: dict[Any, Any] = {} + + # Iterating over BFD peers, extract the parameters and command output + for bfd_peer in self.inputs.bfd_peers: + peer = str(bfd_peer.peer_address) + vrf = bfd_peer.vrf + protocols = bfd_peer.protocols + bfd_output = get_value( + self.instance_commands[0].json_output, + f"vrfs..{vrf}..ipv4Neighbors..{peer}..peerStats..", + separator="..", + ) + + # Check if BFD peer configured + if not bfd_output: + failures[peer] = {vrf: "Not Configured"} + continue + + # Check registered protocols + difference = set(protocols) - set(get_value(bfd_output, "peerStatsDetail.apps")) + + if difference: + failures[peer] = {vrf: sorted(difference)} + + if not failures: + self.result.is_success() + else: + self.result.is_failure(f"The following BFD peers are not configured or have non-registered protocol(s):\n{failures}") diff --git a/examples/tests.yaml b/examples/tests.yaml index c4248cf75..58161972f 100644 --- a/examples/tests.yaml +++ b/examples/tests.yaml @@ -83,6 +83,13 @@ anta.tests.bfd: multiplier: 3 - VerifyBFDPeersHealth: down_threshold: 2 + - VerifyBFDPeersRegProtocols: + bfd_peers: + - peer_address: 192.0.255.8 + vrf: default + protocols: + - bgp + - isis anta.tests.configuration: - VerifyZeroTouch: diff --git a/tests/units/anta_tests/test_bfd.py b/tests/units/anta_tests/test_bfd.py index 54dc7a05e..b3ab5609a 100644 --- a/tests/units/anta_tests/test_bfd.py +++ b/tests/units/anta_tests/test_bfd.py @@ -10,7 +10,7 @@ # pylint: disable=C0413 # because of the patch above -from anta.tests.bfd import VerifyBFDPeersHealth, VerifyBFDPeersIntervals, VerifyBFDSpecificPeers +from anta.tests.bfd import VerifyBFDPeersHealth, VerifyBFDPeersIntervals, VerifyBFDPeersRegProtocols, VerifyBFDSpecificPeers from tests.lib.anta import test # noqa: F401; pylint: disable=W0611 DATA: list[dict[str, Any]] = [ @@ -519,4 +519,133 @@ ], }, }, + { + "name": "success", + "test": VerifyBFDPeersRegProtocols, + "eos_data": [ + { + "vrfs": { + "default": { + "ipv4Neighbors": { + "192.0.255.7": { + "peerStats": { + "": { + "status": "up", + "remoteDisc": 108328132, + "peerStatsDetail": { + "role": "active", + "apps": ["ospf"], + }, + } + } + } + } + }, + "MGMT": { + "ipv4Neighbors": { + "192.0.255.70": { + "peerStats": { + "": { + "status": "up", + "remoteDisc": 108328132, + "peerStatsDetail": { + "role": "active", + "apps": ["bgp"], + }, + } + } + } + } + }, + } + } + ], + "inputs": { + "bfd_peers": [ + {"peer_address": "192.0.255.7", "vrf": "default", "protocols": ["ospf"]}, + {"peer_address": "192.0.255.70", "vrf": "MGMT", "protocols": ["bgp"]}, + ] + }, + "expected": {"result": "success"}, + }, + { + "name": "failure", + "test": VerifyBFDPeersRegProtocols, + "eos_data": [ + { + "vrfs": { + "default": { + "ipv4Neighbors": { + "192.0.255.7": { + "peerStats": { + "": { + "status": "up", + "peerStatsDetail": { + "role": "active", + "apps": ["ospf"], + }, + } + } + } + } + }, + "MGMT": { + "ipv4Neighbors": { + "192.0.255.70": { + "peerStats": { + "": { + "status": "up", + "remoteDisc": 0, + "peerStatsDetail": { + "role": "active", + "apps": ["bgp"], + }, + } + } + } + } + }, + } + } + ], + "inputs": { + "bfd_peers": [ + {"peer_address": "192.0.255.7", "vrf": "default", "protocols": ["isis"]}, + {"peer_address": "192.0.255.70", "vrf": "MGMT", "protocols": ["isis"]}, + ] + }, + "expected": { + "result": "failure", + "messages": [ + "The following BFD peers are not configured or have non-registered protocol(s):\n" + "{'192.0.255.7': {'default': ['isis']}, " + "'192.0.255.70': {'MGMT': ['isis']}}" + ], + }, + }, + { + "name": "failure-not-found", + "test": VerifyBFDPeersRegProtocols, + "eos_data": [ + { + "vrfs": { + "default": {}, + "MGMT": {}, + } + } + ], + "inputs": { + "bfd_peers": [ + {"peer_address": "192.0.255.7", "vrf": "default", "protocols": ["isis"]}, + {"peer_address": "192.0.255.70", "vrf": "MGMT", "protocols": ["isis"]}, + ] + }, + "expected": { + "result": "failure", + "messages": [ + "The following BFD peers are not configured or have non-registered protocol(s):\n" + "{'192.0.255.7': {'default': 'Not Configured'}, '192.0.255.70': {'MGMT': 'Not Configured'}}" + ], + }, + }, ]