From 2e4ab4472da9eb41c6e11ba2da1b7f84c4fc0da1 Mon Sep 17 00:00:00 2001 From: Mahesh Date: Wed, 20 Dec 2023 08:50:34 +0000 Subject: [PATCH 1/8] issue-499: Added testcase for BFD session and timers --- anta/tests/bfd.py | 124 ++++++++++++++++++++ examples/tests.yaml | 16 +++ tests/units/anta_tests/test_bfd.py | 176 +++++++++++++++++++++++++++++ 3 files changed, 316 insertions(+) create mode 100644 anta/tests/bfd.py create mode 100644 tests/units/anta_tests/test_bfd.py diff --git a/anta/tests/bfd.py b/anta/tests/bfd.py new file mode 100644 index 000000000..954a712df --- /dev/null +++ b/anta/tests/bfd.py @@ -0,0 +1,124 @@ +# Copyright (c) 2023 Arista Networks, Inc. +# Use of this source code is governed by the Apache License 2.0 +# that can be found in the LICENSE file. +""" +BFD test functions +""" +# Mypy does not understand AntaTest.Input typing +# mypy: disable-error-code=attr-defined +from __future__ import annotations + +from ipaddress import IPv4Address, IPv6Address, ip_address +from typing import Any, List, Union, cast + +from pydantic import BaseModel + +from anta.models import AntaCommand, AntaTemplate, AntaTest +from anta.tools.get_value import get_value + + +class VerifyBFDPeers(AntaTest): + """ + This class verifies if the BFD (Bidirectional Forwarding Detection) neighbor's sessions are UP and + if the timers are correctly set in the specified VRF (Virtual Routing and Forwarding). + + Expected results: + * success: The test will pass if BFD neighbors are UP and timers are correctly set in the specified VRF. + * failure: The test will fail if BFD neighbors are not found, status is not UP, or timers are not as expected in the specified VRF. + """ + + name = "VerifyBFDPeers" + description = "Verifies if BFD neighbor's sessions are UP and timers are correctly set in the specified VRF." + categories = ["bfd"] + commands = [AntaTemplate(template="show bfd peers dest-ip {neighbor} vrf {vrf} detail")] + + class Input(AntaTest.Input): + """ + This class defines the input parameters of the testcase. + """ + + bfd_neighbors: List["BFDNeighbors"] + """List of BFD neighbors""" + + class BFDNeighbors(BaseModel): + """ + This class defines the details of a BFD neighbor. + """ + + neighbor: Union[IPv4Address, IPv6Address] + """IPv4/IPv6 BFD neighbor""" + vrf: str = "default" + """VRF context""" + loopback: Union[IPv4Address, IPv6Address] + """Loopback IP address""" + tx_interval: int + """Tx interval of BFD neighbor""" + rx_interval: int + """Rx interval of BFD neighbor""" + multiplier: int + """Multiplier of BFD neighbor""" + + def render(self, template: AntaTemplate) -> list[AntaCommand]: + """ + This method renders the template with the BFD neighbor details. + """ + return [ + template.render( + neighbor=bfd_neighbor.neighbor, + vrf=bfd_neighbor.vrf, + loopback=bfd_neighbor.loopback, + tx_interval=bfd_neighbor.tx_interval, + rx_interval=bfd_neighbor.rx_interval, + multiplier=bfd_neighbor.multiplier, + ) + for bfd_neighbor in self.inputs.bfd_neighbors + ] + + @AntaTest.anta_test + def test(self) -> None: + """ + This method tests the BFD neighbors. + """ + failures: dict[str, Any] = {} + + # Iterating over command output for different neighbors + for command in self.instance_commands: + neighbor = cast(str, command.params.get("neighbor")) + vrf = cast(str, command.params.get("vrf")) + loopback = command.params.get("loopback") + tx_interval = command.params.get("tx_interval") + rx_interval = command.params.get("rx_interval") + multiplier = command.params.get("multiplier") + + # Verify BFD neighbor state and timers + if not ( + bfd_output := get_value( + command.json_output, + f"vrfs..{vrf}..{'ipv4Neighbors' if isinstance(ip_address(neighbor), IPv4Address) else 'ipv6Neighbors'}..{neighbor}" + f"..peers....types..multihop..peerStats..{loopback}", + separator="..", + ) + ): + failures[str(neighbor)] = {vrf: "Not Configured"} + continue + + bfd_details = bfd_output["peerStatsDetail"] + if ( + bfd_output["status"] != "up" + or bfd_details["operTxInterval"] != tx_interval + or bfd_details["operRxInterval"] != rx_interval + or bfd_details["detectMult"] != multiplier + ): + failures[str(neighbor)] = { + vrf: { + "status": bfd_output["status"], + "tx_interval": bfd_details["operTxInterval"], + "rx_interval": bfd_details["operRxInterval"], + "multiplier": bfd_details["detectMult"], + } + } + + if not failures: + self.result.is_success() + else: + self.result.is_failure(f"Following BFD neighbors are not UP, not configured or timers are not ok:\n{failures}") diff --git a/examples/tests.yaml b/examples/tests.yaml index e395ba872..571ca66ef 100644 --- a/examples/tests.yaml +++ b/examples/tests.yaml @@ -50,6 +50,22 @@ anta.tests.aaa: - commands - dot1x +anta.tests.bfd: + - VerifyBFDPeers: + bfd_neighbors: + - neighbor: 192.0.255.8 + vrf: default + loopback: 192.0.255.1 + tx_interval: 1200000 + rx_interval: 1200000 + multiplier: 3 + - neighbor: 192.0.255.7 + vrf: default + loopback: 192.0.255.1 + tx_interval: 1200000 + rx_interval: 1200000 + multiplier: 3 + anta.tests.configuration: - VerifyZeroTouch: - VerifyRunningConfigDiffs: diff --git a/tests/units/anta_tests/test_bfd.py b/tests/units/anta_tests/test_bfd.py new file mode 100644 index 000000000..52c47f94f --- /dev/null +++ b/tests/units/anta_tests/test_bfd.py @@ -0,0 +1,176 @@ +# Copyright (c) 2023 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.tests.bfd.py +""" +# pylint: disable=C0302 +from __future__ import annotations + +from typing import Any + +# pylint: disable=C0413 +# because of the patch above +from anta.tests.bfd import VerifyBFDPeers # noqa: E402 +from tests.lib.anta import test # noqa: F401; pylint: disable=W0611 + +DATA: list[dict[str, Any]] = [ + { + "name": "success", + "test": VerifyBFDPeers, + "eos_data": [ + { + "vrfs": { + "default": { + "ipv4Neighbors": { + "192.0.255.7": { + "peers": { + "": { + "types": { + "multihop": { + "peerStats": { + "192.0.255.1": { + "status": "up", + "peerStatsDetail": { + "operTxInterval": 1200000, + "operRxInterval": 1200000, + "detectMult": 3, + }, + } + } + } + } + } + } + } + } + } + } + } + ], + "inputs": { + "bfd_neighbors": [ + {"neighbor": "192.0.255.7", "vrf": "default", "loopback": "192.0.255.1", "tx_interval": 1200000, "rx_interval": 1200000, "multiplier": 3} + ] + }, + "expected": {"result": "success"}, + }, + { + "name": "failure-no-vrf", + "test": VerifyBFDPeers, + "eos_data": [{"vrfs": {}, "errors": ["VRF MGMT not configured."]}], + "inputs": { + "bfd_neighbors": [{"neighbor": "192.0.255.7", "vrf": "MGMT", "loopback": "192.0.255.1", "tx_interval": 1200000, "rx_interval": 1200000, "multiplier": 3}] + }, + "expected": { + "result": "failure", + "messages": ["Following BFD neighbors are not UP, not configured or timers are not ok:\n{'192.0.255.7': {'MGMT': 'Not Configured'}}"], + }, + }, + { + "name": "failure-no-neighbor", + "test": VerifyBFDPeers, + "eos_data": [{"vrfs": {}}], + "inputs": { + "bfd_neighbors": [ + {"neighbor": "192.0.255.70", "vrf": "default", "loopback": "192.0.255.1", "tx_interval": 1200000, "rx_interval": 1200000, "multiplier": 3} + ] + }, + "expected": { + "result": "failure", + "messages": ["Following BFD neighbors are not UP, not configured or timers are not ok:\n{'192.0.255.70': {'default': 'Not Configured'}}"], + }, + }, + { + "name": "failure-session-down", + "test": VerifyBFDPeers, + "eos_data": [ + { + "vrfs": { + "default": { + "ipv4Neighbors": { + "192.0.255.7": { + "peers": { + "": { + "types": { + "multihop": { + "peerStats": { + "192.0.255.1": { + "status": "down", + "peerStatsDetail": { + "operTxInterval": 1200000, + "operRxInterval": 1200000, + "detectMult": 3, + }, + } + } + } + } + } + } + } + } + } + } + } + ], + "inputs": { + "bfd_neighbors": [ + {"neighbor": "192.0.255.7", "vrf": "default", "loopback": "192.0.255.1", "tx_interval": 1200000, "rx_interval": 1200000, "multiplier": 3} + ] + }, + "expected": { + "result": "failure", + "messages": [ + "Following BFD neighbors are not UP, not configured or timers are not ok:\n" + "{'192.0.255.7': {'default': {'status': 'down', 'tx_interval': 1200000, 'rx_interval': 1200000, 'multiplier': 3}}}" + ], + }, + }, + { + "name": "failure-incorrect-timers", + "test": VerifyBFDPeers, + "eos_data": [ + { + "vrfs": { + "default": { + "ipv4Neighbors": { + "192.0.255.7": { + "peers": { + "": { + "types": { + "multihop": { + "peerStats": { + "192.0.255.1": { + "status": "up", + "peerStatsDetail": { + "operTxInterval": 1300000, + "operRxInterval": 1300000, + "detectMult": 4, + }, + } + } + } + } + } + } + } + } + } + } + } + ], + "inputs": { + "bfd_neighbors": [ + {"neighbor": "192.0.255.7", "vrf": "default", "loopback": "192.0.255.1", "tx_interval": 1200000, "rx_interval": 1200000, "multiplier": 3} + ] + }, + "expected": { + "result": "failure", + "messages": [ + "Following BFD neighbors are not UP, not configured or timers are not ok:\n" + "{'192.0.255.7': {'default': {'status': 'up', 'tx_interval': 1300000, 'rx_interval': 1300000, 'multiplier': 4}}}" + ], + }, + }, +] From 8e92dd99084f7fadc2ff34c06d949b32e2f08811 Mon Sep 17 00:00:00 2001 From: Mahesh Date: Wed, 20 Dec 2023 11:23:46 +0000 Subject: [PATCH 2/8] issue-499: improve code --- anta/tests/bfd.py | 48 ++++++++++++++++++------------ tests/units/anta_tests/test_bfd.py | 18 ++--------- 2 files changed, 32 insertions(+), 34 deletions(-) diff --git a/anta/tests/bfd.py b/anta/tests/bfd.py index 954a712df..63cf101cf 100644 --- a/anta/tests/bfd.py +++ b/anta/tests/bfd.py @@ -58,7 +58,7 @@ class BFDNeighbors(BaseModel): multiplier: int """Multiplier of BFD neighbor""" - def render(self, template: AntaTemplate) -> list[AntaCommand]: + def render(self, template: AntaTemplate) -> List[AntaCommand]: """ This method renders the template with the BFD neighbor details. """ @@ -74,11 +74,26 @@ def render(self, template: AntaTemplate) -> list[AntaCommand]: for bfd_neighbor in self.inputs.bfd_neighbors ] - @AntaTest.anta_test - def test(self) -> None: + def create_bfd_neighbor_key(self, vrf: str, neighbor: str) -> str: """ - This method tests the BFD neighbors. + Create a key for retrieving BFD neighbor information based on the VRF and neighbor's IP type. + + Parameters: + - vrf (str): Virtual Routing and Forwarding context. + - neighbor (str): IPv4 or IPv6 address of the BFD neighbor. + + Returns: + str: Key used to retrieve BFD neighbor information from the command output. + + Example: + >>> create_bfd_neighbor_key("default", "192.168.1.1") + 'vrfs..default..ipv4Neighbors..192.168.1.1..peers....types..multihop..peerStats' """ + ip_type = "ipv4" if isinstance(ip_address(neighbor), IPv4Address) else "ipv6" + return f"vrfs..{vrf}..{ip_type}Neighbors..{neighbor}..peers....types..multihop..peerStats" + + @AntaTest.anta_test + def test(self) -> None: failures: dict[str, Any] = {} # Iterating over command output for different neighbors @@ -90,25 +105,19 @@ def test(self) -> None: rx_interval = command.params.get("rx_interval") multiplier = command.params.get("multiplier") + bfd_key = self.create_bfd_neighbor_key(vrf, neighbor) + bfd_output = get_value(command.json_output, f"{bfd_key}..{loopback}", separator="..") + # Verify BFD neighbor state and timers - if not ( - bfd_output := get_value( - command.json_output, - f"vrfs..{vrf}..{'ipv4Neighbors' if isinstance(ip_address(neighbor), IPv4Address) else 'ipv6Neighbors'}..{neighbor}" - f"..peers....types..multihop..peerStats..{loopback}", - separator="..", - ) - ): + if not bfd_output: failures[str(neighbor)] = {vrf: "Not Configured"} continue bfd_details = bfd_output["peerStatsDetail"] - if ( - bfd_output["status"] != "up" - or bfd_details["operTxInterval"] != tx_interval - or bfd_details["operRxInterval"] != rx_interval - or bfd_details["detectMult"] != multiplier - ): + status_up = bfd_output["status"] == "up" + intervals_ok = bfd_details["operTxInterval"] == tx_interval and bfd_details["operRxInterval"] == rx_interval and bfd_details["detectMult"] == multiplier + + if not (status_up and intervals_ok): failures[str(neighbor)] = { vrf: { "status": bfd_output["status"], @@ -121,4 +130,5 @@ def test(self) -> None: if not failures: self.result.is_success() else: - self.result.is_failure(f"Following BFD neighbors are not UP, not configured or timers are not ok:\n{failures}") + message = f"Following BFD neighbors are not UP, not configured, or timers are not ok:\n{failures}" + self.result.is_failure(message) diff --git a/tests/units/anta_tests/test_bfd.py b/tests/units/anta_tests/test_bfd.py index 52c47f94f..2710fa96e 100644 --- a/tests/units/anta_tests/test_bfd.py +++ b/tests/units/anta_tests/test_bfd.py @@ -55,18 +55,6 @@ }, "expected": {"result": "success"}, }, - { - "name": "failure-no-vrf", - "test": VerifyBFDPeers, - "eos_data": [{"vrfs": {}, "errors": ["VRF MGMT not configured."]}], - "inputs": { - "bfd_neighbors": [{"neighbor": "192.0.255.7", "vrf": "MGMT", "loopback": "192.0.255.1", "tx_interval": 1200000, "rx_interval": 1200000, "multiplier": 3}] - }, - "expected": { - "result": "failure", - "messages": ["Following BFD neighbors are not UP, not configured or timers are not ok:\n{'192.0.255.7': {'MGMT': 'Not Configured'}}"], - }, - }, { "name": "failure-no-neighbor", "test": VerifyBFDPeers, @@ -78,7 +66,7 @@ }, "expected": { "result": "failure", - "messages": ["Following BFD neighbors are not UP, not configured or timers are not ok:\n{'192.0.255.70': {'default': 'Not Configured'}}"], + "messages": ["Following BFD neighbors are not UP, not configured, or timers are not ok:\n{'192.0.255.70': {'default': 'Not Configured'}}"], }, }, { @@ -122,7 +110,7 @@ "expected": { "result": "failure", "messages": [ - "Following BFD neighbors are not UP, not configured or timers are not ok:\n" + "Following BFD neighbors are not UP, not configured, or timers are not ok:\n" "{'192.0.255.7': {'default': {'status': 'down', 'tx_interval': 1200000, 'rx_interval': 1200000, 'multiplier': 3}}}" ], }, @@ -168,7 +156,7 @@ "expected": { "result": "failure", "messages": [ - "Following BFD neighbors are not UP, not configured or timers are not ok:\n" + "Following BFD neighbors are not UP, not configured, or timers are not ok:\n" "{'192.0.255.7': {'default': {'status': 'up', 'tx_interval': 1300000, 'rx_interval': 1300000, 'multiplier': 4}}}" ], }, From bc6a7afc322e6c9e1cf36a1fd58a2931c8f301a4 Mon Sep 17 00:00:00 2001 From: Mahesh Date: Wed, 20 Dec 2023 12:39:22 +0000 Subject: [PATCH 3/8] issue-499: fix tox issue for py38 --- anta/tests/bfd.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/anta/tests/bfd.py b/anta/tests/bfd.py index 63cf101cf..020bd89e4 100644 --- a/anta/tests/bfd.py +++ b/anta/tests/bfd.py @@ -58,6 +58,8 @@ class BFDNeighbors(BaseModel): multiplier: int """Multiplier of BFD neighbor""" + Input.model_rebuild() + def render(self, template: AntaTemplate) -> List[AntaCommand]: """ This method renders the template with the BFD neighbor details. From 1be46af786b77405c17f59ded610d9d461b23dea Mon Sep 17 00:00:00 2001 From: Mahesh Date: Wed, 20 Dec 2023 12:45:13 +0000 Subject: [PATCH 4/8] issue-499: removed model_rebuild method --- anta/tests/bfd.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/anta/tests/bfd.py b/anta/tests/bfd.py index 020bd89e4..63cf101cf 100644 --- a/anta/tests/bfd.py +++ b/anta/tests/bfd.py @@ -58,8 +58,6 @@ class BFDNeighbors(BaseModel): multiplier: int """Multiplier of BFD neighbor""" - Input.model_rebuild() - def render(self, template: AntaTemplate) -> List[AntaCommand]: """ This method renders the template with the BFD neighbor details. From 984d6874c2aa3069497ede8b705a100b2499adcb Mon Sep 17 00:00:00 2001 From: Mahesh Date: Tue, 26 Dec 2023 10:11:49 +0000 Subject: [PATCH 5/8] issue-499: updated the verification and split the testcase --- anta/tests/bfd.py | 217 ++++++++++++++++++++--------- tests/units/anta_tests/test_bfd.py | 140 ++++++++++++++----- 2 files changed, 258 insertions(+), 99 deletions(-) diff --git a/anta/tests/bfd.py b/anta/tests/bfd.py index 63cf101cf..39009e66e 100644 --- a/anta/tests/bfd.py +++ b/anta/tests/bfd.py @@ -9,7 +9,7 @@ from __future__ import annotations from ipaddress import IPv4Address, IPv6Address, ip_address -from typing import Any, List, Union, cast +from typing import Any, Dict, List, Union from pydantic import BaseModel @@ -17,118 +17,207 @@ from anta.tools.get_value import get_value -class VerifyBFDPeers(AntaTest): +def get_bfd_output(command: Any) -> Dict[str, Any] | None: """ - This class verifies if the BFD (Bidirectional Forwarding Detection) neighbor's sessions are UP and - if the timers are correctly set in the specified VRF (Virtual Routing and Forwarding). + This function extracts the BFD output for a given command. + + Parameters: + command: The command object which contains parameters and json_output. + + Returns: + Dict or None: The BFD output if it exists, otherwise None. + """ + peer = command.params.get("peer") + vrf = command.params.get("vrf") + source_address = command.params.get("source_address") + bfd_key = create_bfd_peer_key(vrf, peer) + bfd_output = get_value(command.json_output, f"{bfd_key}", separator="..") + if not bfd_output: + return None + + bfd_output = bfd_output.get("normal", bfd_output.get("multihop", {})).get("peerStats", {}).get(str(source_address), {}) + return bfd_output + + +def create_bfd_peer_key(vrf: str, peer: str) -> str: + """ + Create a key for retrieving BFD peer information based on the VRF and peer's IP type. + + Parameters: + - vrf (str): Virtual Routing and Forwarding context. + - peer (str): IPv4 or IPv6 address of the BFD peer. + + Returns: + str: Key used to retrieve BFD peer information from the command output. + + Example: + >>> create_bfd_peer_key("default", "192.168.1.1") + 'vrfs..default..ipv4Neighbors..192.168.1.1..peers....types..multihop..peerStats' + """ + ip_type = "ipv4" if isinstance(ip_address(peer), IPv4Address) else "ipv6" + return f"vrfs..{vrf}..{ip_type}Neighbors..{peer}..peers....types" + + +class VerifyBFDSpecificPeers(AntaTest): + """ + This class verifies if the BFD peer's sessions are UP and remote disc is non-zero in the specified VRF. + Session types are supported as normal and multihop. Default it will check for normal, if not exist then will check for multihop. Expected results: - * success: The test will pass if BFD neighbors are UP and timers are correctly set in the specified VRF. - * failure: The test will fail if BFD neighbors are not found, status is not UP, or timers are not as expected in the specified VRF. + * success: The test will pass if BFD peers are up and remote disc is non-zero in the specified VRF. + * failure: The test will fail if BFD peers are not found, the status is not UP or remote disc is zero in the specified VRF. """ - name = "VerifyBFDPeers" - description = "Verifies if BFD neighbor's sessions are UP and timers are correctly set in the specified VRF." + name = "VerifyBFDSpecificPeers" + description = "Verifies if the BFD peer's sessions are UP and remote disc is non-zero in the specified VRF." categories = ["bfd"] - commands = [AntaTemplate(template="show bfd peers dest-ip {neighbor} vrf {vrf} detail")] + commands = [AntaTemplate(template="show bfd peers dest-ip {peer} vrf {vrf}")] class Input(AntaTest.Input): """ This class defines the input parameters of the testcase. """ - bfd_neighbors: List["BFDNeighbors"] - """List of BFD neighbors""" + bfd_peers: List[BFDPeers] + """List of BFD peers""" - class BFDNeighbors(BaseModel): + class BFDPeers(BaseModel): """ - This class defines the details of a BFD neighbor. + This class defines the details of a BFD peer. """ - neighbor: Union[IPv4Address, IPv6Address] - """IPv4/IPv6 BFD neighbor""" + peer: Union[IPv4Address, IPv6Address] + """IPv4/IPv6 BFD peer""" vrf: str = "default" """VRF context""" - loopback: Union[IPv4Address, IPv6Address] - """Loopback IP address""" - tx_interval: int - """Tx interval of BFD neighbor""" - rx_interval: int - """Rx interval of BFD neighbor""" - multiplier: int - """Multiplier of BFD neighbor""" + source_address: Union[IPv4Address, IPv6Address] + """Source IP address of BFD peer""" def render(self, template: AntaTemplate) -> List[AntaCommand]: """ - This method renders the template with the BFD neighbor details. + This method renders the template with the BFD peer details. """ return [ template.render( - neighbor=bfd_neighbor.neighbor, - vrf=bfd_neighbor.vrf, - loopback=bfd_neighbor.loopback, - tx_interval=bfd_neighbor.tx_interval, - rx_interval=bfd_neighbor.rx_interval, - multiplier=bfd_neighbor.multiplier, + peer=bfd_peer.peer, + vrf=bfd_peer.vrf, + source_address=bfd_peer.source_address, ) - for bfd_neighbor in self.inputs.bfd_neighbors + for bfd_peer in self.inputs.bfd_peers ] - def create_bfd_neighbor_key(self, vrf: str, neighbor: str) -> str: + @AntaTest.anta_test + def test(self) -> None: + failures: dict[str, Any] = {} + + # Iterating over command output for different peers + for command in self.instance_commands: + peer = command.params.get("peer") + vrf = command.params.get("vrf") + bfd_output = get_bfd_output(command) + if not bfd_output: + failures[str(peer)] = {vrf: "Not Configured"} + continue + + # Verify BFD peer status and remote disc + if not (bfd_output.get("status") == "up" and bfd_output.get("remoteDisc") != 0): + failures[str(peer)] = {vrf: {"status": bfd_output.get("status"), "remote_disc": bfd_output.get("remoteDisc")}} + + if not failures: + self.result.is_success() + else: + self.result.is_failure(f"Following BFD peers are not configured, status is not up or remote disc is zero:\n{failures}") + + +class VerifyBFDPeersIntervals(AntaTest): + """ + This class verifies the timers of the BFD peers in the specified VRF. + Session types are supported as normal and multihop. Default it will check for normal, if not exist then will check for multihop. + + Expected results: + * success: The test will pass if the timers of the BFD peers are correct in the specified VRF. + * failure: The test will fail if the BFD peers are not found or their timers are incorrect in the specified VRF. + """ + + name = "VerifyBFDPeersIntervals" + description = "Verifies the timers of the BFD peers in the specified VRF." + categories = ["bfd"] + commands = [AntaTemplate(template="show bfd peers dest-ip {peer} vrf {vrf} detail")] + + class Input(AntaTest.Input): + """ + This class defines the input parameters of the testcase. """ - Create a key for retrieving BFD neighbor information based on the VRF and neighbor's IP type. - Parameters: - - vrf (str): Virtual Routing and Forwarding context. - - neighbor (str): IPv4 or IPv6 address of the BFD neighbor. + bfd_peers: List[BFDPeers] + """List of BFD peers""" - Returns: - str: Key used to retrieve BFD neighbor information from the command output. + class BFDPeers(BaseModel): + """ + This class defines the details of a BFD peer. + """ - Example: - >>> create_bfd_neighbor_key("default", "192.168.1.1") - 'vrfs..default..ipv4Neighbors..192.168.1.1..peers....types..multihop..peerStats' + peer: Union[IPv4Address, IPv6Address] + """IPv4/IPv6 BFD peer""" + vrf: str = "default" + """VRF context""" + source_address: Union[IPv4Address, IPv6Address] + """Source IP address of BFD peer""" + tx_interval: int + """Tx interval of BFD peer""" + rx_interval: int + """Rx interval of BFD peer""" + multiplier: int + """Multiplier of BFD peer""" + + def render(self, template: AntaTemplate) -> List[AntaCommand]: + """ + This method renders the template with the BFD peer details. """ - ip_type = "ipv4" if isinstance(ip_address(neighbor), IPv4Address) else "ipv6" - return f"vrfs..{vrf}..{ip_type}Neighbors..{neighbor}..peers....types..multihop..peerStats" + return [ + template.render( + peer=bfd_peer.peer, + vrf=bfd_peer.vrf, + source_address=bfd_peer.source_address, + tx_interval=bfd_peer.tx_interval, + rx_interval=bfd_peer.rx_interval, + multiplier=bfd_peer.multiplier, + ) + for bfd_peer in self.inputs.bfd_peers + ] @AntaTest.anta_test def test(self) -> None: failures: dict[str, Any] = {} - # Iterating over command output for different neighbors + # Iterating over command output for different peers for command in self.instance_commands: - neighbor = cast(str, command.params.get("neighbor")) - vrf = cast(str, command.params.get("vrf")) - loopback = command.params.get("loopback") + peer = command.params.get("peer") + vrf = command.params.get("vrf") tx_interval = command.params.get("tx_interval") rx_interval = command.params.get("rx_interval") multiplier = command.params.get("multiplier") - - bfd_key = self.create_bfd_neighbor_key(vrf, neighbor) - bfd_output = get_value(command.json_output, f"{bfd_key}..{loopback}", separator="..") - - # Verify BFD neighbor state and timers + bfd_output = get_bfd_output(command) if not bfd_output: - failures[str(neighbor)] = {vrf: "Not Configured"} + failures[str(peer)] = {vrf: "Not Configured"} continue - bfd_details = bfd_output["peerStatsDetail"] - status_up = bfd_output["status"] == "up" - intervals_ok = bfd_details["operTxInterval"] == tx_interval and bfd_details["operRxInterval"] == rx_interval and bfd_details["detectMult"] == multiplier + bfd_details = bfd_output.get("peerStatsDetail", {}) + intervals_ok = ( + bfd_details.get("operTxInterval") == tx_interval and bfd_details.get("operRxInterval") == rx_interval and bfd_details.get("detectMult") == multiplier + ) - if not (status_up and intervals_ok): - failures[str(neighbor)] = { + # Verify timers of BFD peer + if not intervals_ok: + failures[str(peer)] = { vrf: { - "status": bfd_output["status"], - "tx_interval": bfd_details["operTxInterval"], - "rx_interval": bfd_details["operRxInterval"], - "multiplier": bfd_details["detectMult"], + "tx_interval": bfd_details.get("operTxInterval"), + "rx_interval": bfd_details.get("operRxInterval"), + "multiplier": bfd_details.get("detectMult"), } } if not failures: self.result.is_success() else: - message = f"Following BFD neighbors are not UP, not configured, or timers are not ok:\n{failures}" - self.result.is_failure(message) + self.result.is_failure(f"Following BFD peers are not configured or timers are not correct:\n{failures}") diff --git a/tests/units/anta_tests/test_bfd.py b/tests/units/anta_tests/test_bfd.py index 2710fa96e..3c0d53d18 100644 --- a/tests/units/anta_tests/test_bfd.py +++ b/tests/units/anta_tests/test_bfd.py @@ -11,13 +11,13 @@ # pylint: disable=C0413 # because of the patch above -from anta.tests.bfd import VerifyBFDPeers # noqa: E402 +from anta.tests.bfd import VerifyBFDPeersIntervals, VerifyBFDSpecificPeers # noqa: E402 from tests.lib.anta import test # noqa: F401; pylint: disable=W0611 DATA: list[dict[str, Any]] = [ { "name": "success", - "test": VerifyBFDPeers, + "test": VerifyBFDPeersIntervals, "eos_data": [ { "vrfs": { @@ -30,12 +30,11 @@ "multihop": { "peerStats": { "192.0.255.1": { - "status": "up", "peerStatsDetail": { "operTxInterval": 1200000, "operRxInterval": 1200000, "detectMult": 3, - }, + } } } } @@ -49,29 +48,29 @@ } ], "inputs": { - "bfd_neighbors": [ - {"neighbor": "192.0.255.7", "vrf": "default", "loopback": "192.0.255.1", "tx_interval": 1200000, "rx_interval": 1200000, "multiplier": 3} + "bfd_peers": [ + {"peer": "192.0.255.7", "vrf": "default", "source_address": "192.0.255.1", "tx_interval": 1200000, "rx_interval": 1200000, "multiplier": 3} ] }, "expected": {"result": "success"}, }, { - "name": "failure-no-neighbor", - "test": VerifyBFDPeers, + "name": "failure-no-peer", + "test": VerifyBFDPeersIntervals, "eos_data": [{"vrfs": {}}], "inputs": { - "bfd_neighbors": [ - {"neighbor": "192.0.255.70", "vrf": "default", "loopback": "192.0.255.1", "tx_interval": 1200000, "rx_interval": 1200000, "multiplier": 3} + "bfd_peers": [ + {"peer": "192.0.255.70", "vrf": "default", "source_address": "192.0.255.1", "tx_interval": 1200000, "rx_interval": 1200000, "multiplier": 3} ] }, "expected": { "result": "failure", - "messages": ["Following BFD neighbors are not UP, not configured, or timers are not ok:\n{'192.0.255.70': {'default': 'Not Configured'}}"], + "messages": ["Following BFD peers are not configured or timers are not correct:\n{'192.0.255.70': {'default': 'Not Configured'}}"], }, }, { - "name": "failure-session-down", - "test": VerifyBFDPeers, + "name": "failure-incorrect-timers", + "test": VerifyBFDPeersIntervals, "eos_data": [ { "vrfs": { @@ -84,12 +83,11 @@ "multihop": { "peerStats": { "192.0.255.1": { - "status": "down", "peerStatsDetail": { - "operTxInterval": 1200000, - "operRxInterval": 1200000, - "detectMult": 3, - }, + "operTxInterval": 1300000, + "operRxInterval": 1300000, + "detectMult": 4, + } } } } @@ -103,21 +101,21 @@ } ], "inputs": { - "bfd_neighbors": [ - {"neighbor": "192.0.255.7", "vrf": "default", "loopback": "192.0.255.1", "tx_interval": 1200000, "rx_interval": 1200000, "multiplier": 3} + "bfd_peers": [ + {"peer": "192.0.255.7", "vrf": "default", "source_address": "192.0.255.1", "tx_interval": 1200000, "rx_interval": 1200000, "multiplier": 3} ] }, "expected": { "result": "failure", "messages": [ - "Following BFD neighbors are not UP, not configured, or timers are not ok:\n" - "{'192.0.255.7': {'default': {'status': 'down', 'tx_interval': 1200000, 'rx_interval': 1200000, 'multiplier': 3}}}" + "Following BFD peers are not configured or timers are not correct:\n" + "{'192.0.255.7': {'default': {'tx_interval': 1300000, 'rx_interval': 1300000, 'multiplier': 4}}}" ], }, }, { - "name": "failure-incorrect-timers", - "test": VerifyBFDPeers, + "name": "success", + "test": VerifyBFDSpecificPeers, "eos_data": [ { "vrfs": { @@ -131,11 +129,7 @@ "peerStats": { "192.0.255.1": { "status": "up", - "peerStatsDetail": { - "operTxInterval": 1300000, - "operRxInterval": 1300000, - "detectMult": 4, - }, + "remoteDisc": 108328132, } } } @@ -148,16 +142,92 @@ } } ], - "inputs": { - "bfd_neighbors": [ - {"neighbor": "192.0.255.7", "vrf": "default", "loopback": "192.0.255.1", "tx_interval": 1200000, "rx_interval": 1200000, "multiplier": 3} - ] + "inputs": {"bfd_peers": [{"peer": "192.0.255.7", "vrf": "default", "source_address": "192.0.255.1"}]}, + "expected": {"result": "success"}, + }, + { + "name": "failure-no-peer", + "test": VerifyBFDSpecificPeers, + "eos_data": [{"vrfs": {}}], + "inputs": {"bfd_peers": [{"peer": "192.0.255.70", "vrf": "default", "source_address": "192.0.255.1"}]}, + "expected": { + "result": "failure", + "messages": ["Following BFD peers are not configured, status is not up or remote disc is zero:\n{'192.0.255.70': {'default': 'Not Configured'}}"], + }, + }, + { + "name": "failure-session-down", + "test": VerifyBFDSpecificPeers, + "eos_data": [ + { + "vrfs": { + "default": { + "ipv4Neighbors": { + "192.0.255.7": { + "peers": { + "": { + "types": { + "multihop": { + "peerStats": { + "192.0.255.1": { + "status": "down", + "remoteDisc": 108328132, + } + } + } + } + } + } + } + } + } + } + } + ], + "inputs": {"bfd_peers": [{"peer": "192.0.255.7", "vrf": "default", "source_address": "192.0.255.1"}]}, + "expected": { + "result": "failure", + "messages": [ + "Following BFD peers are not configured, status is not up or remote disc is zero:\n" + "{'192.0.255.7': {'default': {'status': 'down', 'remote_disc': 108328132}}}" + ], }, + }, + { + "name": "failure-zero-remote-disk", + "test": VerifyBFDSpecificPeers, + "eos_data": [ + { + "vrfs": { + "default": { + "ipv4Neighbors": { + "192.0.255.7": { + "peers": { + "": { + "types": { + "multihop": { + "peerStats": { + "192.0.255.1": { + "status": "down", + "remoteDisc": 0, + } + } + } + } + } + } + } + } + } + } + } + ], + "inputs": {"bfd_peers": [{"peer": "192.0.255.7", "vrf": "default", "source_address": "192.0.255.1"}]}, "expected": { "result": "failure", "messages": [ - "Following BFD neighbors are not UP, not configured, or timers are not ok:\n" - "{'192.0.255.7': {'default': {'status': 'up', 'tx_interval': 1300000, 'rx_interval': 1300000, 'multiplier': 4}}}" + "Following BFD peers are not configured, status is not up or remote disc is zero:\n" + "{'192.0.255.7': {'default': {'status': 'down', 'remote_disc': 0}}}" ], }, }, From e292e1c5952699579791ce9ba03b6c3c9872ff41 Mon Sep 17 00:00:00 2001 From: Mahesh Date: Wed, 27 Dec 2023 10:49:05 +0000 Subject: [PATCH 6/8] issue-499: Updated the show command --- anta/tests/bfd.py | 120 +++++++++-------------------- examples/tests.yaml | 2 - tests/units/anta_tests/test_bfd.py | 102 ++++-------------------- 3 files changed, 51 insertions(+), 173 deletions(-) diff --git a/anta/tests/bfd.py b/anta/tests/bfd.py index 39009e66e..352d08864 100644 --- a/anta/tests/bfd.py +++ b/anta/tests/bfd.py @@ -13,32 +13,10 @@ from pydantic import BaseModel -from anta.models import AntaCommand, AntaTemplate, AntaTest +from anta.models import AntaCommand, AntaTest from anta.tools.get_value import get_value -def get_bfd_output(command: Any) -> Dict[str, Any] | None: - """ - This function extracts the BFD output for a given command. - - Parameters: - command: The command object which contains parameters and json_output. - - Returns: - Dict or None: The BFD output if it exists, otherwise None. - """ - peer = command.params.get("peer") - vrf = command.params.get("vrf") - source_address = command.params.get("source_address") - bfd_key = create_bfd_peer_key(vrf, peer) - bfd_output = get_value(command.json_output, f"{bfd_key}", separator="..") - if not bfd_output: - return None - - bfd_output = bfd_output.get("normal", bfd_output.get("multihop", {})).get("peerStats", {}).get(str(source_address), {}) - return bfd_output - - def create_bfd_peer_key(vrf: str, peer: str) -> str: """ Create a key for retrieving BFD peer information based on the VRF and peer's IP type. @@ -52,16 +30,15 @@ def create_bfd_peer_key(vrf: str, peer: str) -> str: Example: >>> create_bfd_peer_key("default", "192.168.1.1") - 'vrfs..default..ipv4Neighbors..192.168.1.1..peers....types..multihop..peerStats' + 'vrfs..default..ipv4Neighbors..192.168.1.1..peerStats..' """ ip_type = "ipv4" if isinstance(ip_address(peer), IPv4Address) else "ipv6" - return f"vrfs..{vrf}..{ip_type}Neighbors..{peer}..peers....types" + return f"vrfs..{vrf}..{ip_type}Neighbors..{peer}..peerStats.." class VerifyBFDSpecificPeers(AntaTest): """ This class verifies if the BFD peer's sessions are UP and remote disc is non-zero in the specified VRF. - Session types are supported as normal and multihop. Default it will check for normal, if not exist then will check for multihop. Expected results: * success: The test will pass if BFD peers are up and remote disc is non-zero in the specified VRF. @@ -69,9 +46,9 @@ class VerifyBFDSpecificPeers(AntaTest): """ name = "VerifyBFDSpecificPeers" - description = "Verifies if the BFD peer's sessions are UP and remote disc is non-zero in the specified VRF." + description = "Verifies the BFD peer's sessions and remote disc in the specified VRF." categories = ["bfd"] - commands = [AntaTemplate(template="show bfd peers dest-ip {peer} vrf {vrf}")] + commands = [AntaCommand(command="show bfd peers", revision=1)] class Input(AntaTest.Input): """ @@ -90,38 +67,26 @@ class BFDPeers(BaseModel): """IPv4/IPv6 BFD peer""" vrf: str = "default" """VRF context""" - source_address: Union[IPv4Address, IPv6Address] - """Source IP address of BFD peer""" - - def render(self, template: AntaTemplate) -> List[AntaCommand]: - """ - This method renders the template with the BFD peer details. - """ - return [ - template.render( - peer=bfd_peer.peer, - vrf=bfd_peer.vrf, - source_address=bfd_peer.source_address, - ) - for bfd_peer in self.inputs.bfd_peers - ] @AntaTest.anta_test def test(self) -> None: - failures: dict[str, Any] = {} + failures: Dict[str, Dict[str, Union[str, Dict[str, Any]]]] = {} + + # Iterating over BFD peers + for bfd_peer in self.inputs.bfd_peers: + peer = str(bfd_peer.peer) + vrf = bfd_peer.vrf + bfd_key = create_bfd_peer_key(vrf, peer) + bfd_output = get_value(self.instance_commands[0].json_output, f"{bfd_key}", separator="..") - # Iterating over command output for different peers - for command in self.instance_commands: - peer = command.params.get("peer") - vrf = command.params.get("vrf") - bfd_output = get_bfd_output(command) + # Check if BFD peer configured if not bfd_output: - failures[str(peer)] = {vrf: "Not Configured"} + failures[peer] = {vrf: "Not Configured"} continue - # Verify BFD peer status and remote disc + # Check BFD peer status and remote disc if not (bfd_output.get("status") == "up" and bfd_output.get("remoteDisc") != 0): - failures[str(peer)] = {vrf: {"status": bfd_output.get("status"), "remote_disc": bfd_output.get("remoteDisc")}} + failures[peer] = {vrf: {"status": bfd_output.get("status"), "remote_disc": bfd_output.get("remoteDisc")}} if not failures: self.result.is_success() @@ -132,7 +97,6 @@ def test(self) -> None: class VerifyBFDPeersIntervals(AntaTest): """ This class verifies the timers of the BFD peers in the specified VRF. - Session types are supported as normal and multihop. Default it will check for normal, if not exist then will check for multihop. Expected results: * success: The test will pass if the timers of the BFD peers are correct in the specified VRF. @@ -142,7 +106,7 @@ class VerifyBFDPeersIntervals(AntaTest): name = "VerifyBFDPeersIntervals" description = "Verifies the timers of the BFD peers in the specified VRF." categories = ["bfd"] - commands = [AntaTemplate(template="show bfd peers dest-ip {peer} vrf {vrf} detail")] + commands = [AntaCommand(command="show bfd peers detail", revision=1)] class Input(AntaTest.Input): """ @@ -161,8 +125,6 @@ class BFDPeers(BaseModel): """IPv4/IPv6 BFD peer""" vrf: str = "default" """VRF context""" - source_address: Union[IPv4Address, IPv6Address] - """Source IP address of BFD peer""" tx_interval: int """Tx interval of BFD peer""" rx_interval: int @@ -170,36 +132,23 @@ class BFDPeers(BaseModel): multiplier: int """Multiplier of BFD peer""" - def render(self, template: AntaTemplate) -> List[AntaCommand]: - """ - This method renders the template with the BFD peer details. - """ - return [ - template.render( - peer=bfd_peer.peer, - vrf=bfd_peer.vrf, - source_address=bfd_peer.source_address, - tx_interval=bfd_peer.tx_interval, - rx_interval=bfd_peer.rx_interval, - multiplier=bfd_peer.multiplier, - ) - for bfd_peer in self.inputs.bfd_peers - ] - @AntaTest.anta_test def test(self) -> None: - failures: dict[str, Any] = {} - - # Iterating over command output for different peers - for command in self.instance_commands: - peer = command.params.get("peer") - vrf = command.params.get("vrf") - tx_interval = command.params.get("tx_interval") - rx_interval = command.params.get("rx_interval") - multiplier = command.params.get("multiplier") - bfd_output = get_bfd_output(command) + failures: Dict[str, Dict[str, Union[str, Dict[str, Any]]]] = {} + + # Iterating over BFD peers + for bfd_peers in self.inputs.bfd_peers: + peer = str(bfd_peers.peer) + vrf = bfd_peers.vrf + tx_interval = bfd_peers.tx_interval + rx_interval = bfd_peers.rx_interval + multiplier = bfd_peers.multiplier + bfd_key = create_bfd_peer_key(vrf, peer) + bfd_output = get_value(self.instance_commands[0].json_output, f"{bfd_key}", separator="..") + + # Check if BFD peer configured if not bfd_output: - failures[str(peer)] = {vrf: "Not Configured"} + failures[peer] = {vrf: "Not Configured"} continue bfd_details = bfd_output.get("peerStatsDetail", {}) @@ -207,9 +156,9 @@ def test(self) -> None: bfd_details.get("operTxInterval") == tx_interval and bfd_details.get("operRxInterval") == rx_interval and bfd_details.get("detectMult") == multiplier ) - # Verify timers of BFD peer + # Check timers of BFD peer if not intervals_ok: - failures[str(peer)] = { + failures[peer] = { vrf: { "tx_interval": bfd_details.get("operTxInterval"), "rx_interval": bfd_details.get("operRxInterval"), @@ -217,6 +166,7 @@ def test(self) -> None: } } + # Check if any failures if not failures: self.result.is_success() else: diff --git a/examples/tests.yaml b/examples/tests.yaml index 571ca66ef..ec0b3303f 100644 --- a/examples/tests.yaml +++ b/examples/tests.yaml @@ -55,13 +55,11 @@ anta.tests.bfd: bfd_neighbors: - neighbor: 192.0.255.8 vrf: default - loopback: 192.0.255.1 tx_interval: 1200000 rx_interval: 1200000 multiplier: 3 - neighbor: 192.0.255.7 vrf: default - loopback: 192.0.255.1 tx_interval: 1200000 rx_interval: 1200000 multiplier: 3 diff --git a/tests/units/anta_tests/test_bfd.py b/tests/units/anta_tests/test_bfd.py index 3c0d53d18..f3413002b 100644 --- a/tests/units/anta_tests/test_bfd.py +++ b/tests/units/anta_tests/test_bfd.py @@ -24,20 +24,12 @@ "default": { "ipv4Neighbors": { "192.0.255.7": { - "peers": { + "peerStats": { "": { - "types": { - "multihop": { - "peerStats": { - "192.0.255.1": { - "peerStatsDetail": { - "operTxInterval": 1200000, - "operRxInterval": 1200000, - "detectMult": 3, - } - } - } - } + "peerStatsDetail": { + "operTxInterval": 1200000, + "operRxInterval": 1200000, + "detectMult": 3, } } } @@ -77,20 +69,12 @@ "default": { "ipv4Neighbors": { "192.0.255.7": { - "peers": { + "peerStats": { "": { - "types": { - "multihop": { - "peerStats": { - "192.0.255.1": { - "peerStatsDetail": { - "operTxInterval": 1300000, - "operRxInterval": 1300000, - "detectMult": 4, - } - } - } - } + "peerStatsDetail": { + "operTxInterval": 1300000, + "operRxInterval": 1300000, + "detectMult": 4, } } } @@ -122,18 +106,10 @@ "default": { "ipv4Neighbors": { "192.0.255.7": { - "peers": { + "peerStats": { "": { - "types": { - "multihop": { - "peerStats": { - "192.0.255.1": { - "status": "up", - "remoteDisc": 108328132, - } - } - } - } + "status": "up", + "remoteDisc": 108328132, } } } @@ -164,56 +140,10 @@ "default": { "ipv4Neighbors": { "192.0.255.7": { - "peers": { - "": { - "types": { - "multihop": { - "peerStats": { - "192.0.255.1": { - "status": "down", - "remoteDisc": 108328132, - } - } - } - } - } - } - } - } - } - } - } - ], - "inputs": {"bfd_peers": [{"peer": "192.0.255.7", "vrf": "default", "source_address": "192.0.255.1"}]}, - "expected": { - "result": "failure", - "messages": [ - "Following BFD peers are not configured, status is not up or remote disc is zero:\n" - "{'192.0.255.7': {'default': {'status': 'down', 'remote_disc': 108328132}}}" - ], - }, - }, - { - "name": "failure-zero-remote-disk", - "test": VerifyBFDSpecificPeers, - "eos_data": [ - { - "vrfs": { - "default": { - "ipv4Neighbors": { - "192.0.255.7": { - "peers": { + "peerStats": { "": { - "types": { - "multihop": { - "peerStats": { - "192.0.255.1": { - "status": "down", - "remoteDisc": 0, - } - } - } - } + "status": "down", + "remoteDisc": 0, } } } From 383d5f040176b955ad00fc0410726dbc29712988 Mon Sep 17 00:00:00 2001 From: Mahesh Kumar Date: Fri, 12 Jan 2024 09:34:19 +0000 Subject: [PATCH 7/8] issue-499: removed Ipv6 support --- anta/tests/bfd.py | 55 +++------- examples/tests.yaml | 14 ++- tests/units/anta_tests/test_bfd.py | 169 +++++++++++++++++++++++++---- 3 files changed, 176 insertions(+), 62 deletions(-) diff --git a/anta/tests/bfd.py b/anta/tests/bfd.py index 352d08864..62aeac69d 100644 --- a/anta/tests/bfd.py +++ b/anta/tests/bfd.py @@ -1,4 +1,4 @@ -# Copyright (c) 2023 Arista Networks, Inc. +# 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. """ @@ -8,8 +8,8 @@ # mypy: disable-error-code=attr-defined from __future__ import annotations -from ipaddress import IPv4Address, IPv6Address, ip_address -from typing import Any, Dict, List, Union +from ipaddress import IPv4Address +from typing import Any, List from pydantic import BaseModel @@ -17,25 +17,6 @@ from anta.tools.get_value import get_value -def create_bfd_peer_key(vrf: str, peer: str) -> str: - """ - Create a key for retrieving BFD peer information based on the VRF and peer's IP type. - - Parameters: - - vrf (str): Virtual Routing and Forwarding context. - - peer (str): IPv4 or IPv6 address of the BFD peer. - - Returns: - str: Key used to retrieve BFD peer information from the command output. - - Example: - >>> create_bfd_peer_key("default", "192.168.1.1") - 'vrfs..default..ipv4Neighbors..192.168.1.1..peerStats..' - """ - ip_type = "ipv4" if isinstance(ip_address(peer), IPv4Address) else "ipv6" - return f"vrfs..{vrf}..{ip_type}Neighbors..{peer}..peerStats.." - - class VerifyBFDSpecificPeers(AntaTest): """ This class verifies if the BFD peer's sessions are UP and remote disc is non-zero in the specified VRF. @@ -48,7 +29,7 @@ class VerifyBFDSpecificPeers(AntaTest): name = "VerifyBFDSpecificPeers" description = "Verifies the BFD peer's sessions and remote disc in the specified VRF." categories = ["bfd"] - commands = [AntaCommand(command="show bfd peers", revision=1)] + commands = [AntaCommand(command="show bfd peers")] class Input(AntaTest.Input): """ @@ -63,21 +44,20 @@ class BFDPeers(BaseModel): This class defines the details of a BFD peer. """ - peer: Union[IPv4Address, IPv6Address] - """IPv4/IPv6 BFD peer""" + peer_address: IPv4Address + """IPv4 address of a BFD peer""" vrf: str = "default" - """VRF context""" + """Optional VRF for BGP peer. If not provided, it defaults to `default`.""" @AntaTest.anta_test def test(self) -> None: - failures: Dict[str, Dict[str, Union[str, Dict[str, Any]]]] = {} + failures: dict[Any, Any] = {} # Iterating over BFD peers for bfd_peer in self.inputs.bfd_peers: - peer = str(bfd_peer.peer) + peer = str(bfd_peer.peer_address) vrf = bfd_peer.vrf - bfd_key = create_bfd_peer_key(vrf, peer) - bfd_output = get_value(self.instance_commands[0].json_output, f"{bfd_key}", separator="..") + 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: @@ -106,7 +86,7 @@ class VerifyBFDPeersIntervals(AntaTest): name = "VerifyBFDPeersIntervals" description = "Verifies the timers of the BFD peers in the specified VRF." categories = ["bfd"] - commands = [AntaCommand(command="show bfd peers detail", revision=1)] + commands = [AntaCommand(command="show bfd peers detail")] class Input(AntaTest.Input): """ @@ -121,10 +101,10 @@ class BFDPeers(BaseModel): This class defines the details of a BFD peer. """ - peer: Union[IPv4Address, IPv6Address] - """IPv4/IPv6 BFD peer""" + peer_address: IPv4Address + """IPv4 address of a BFD peer""" vrf: str = "default" - """VRF context""" + """Optional VRF for BGP peer. If not provided, it defaults to `default`.""" tx_interval: int """Tx interval of BFD peer""" rx_interval: int @@ -134,17 +114,16 @@ class BFDPeers(BaseModel): @AntaTest.anta_test def test(self) -> None: - failures: Dict[str, Dict[str, Union[str, Dict[str, Any]]]] = {} + failures: dict[Any, Any] = {} # Iterating over BFD peers for bfd_peers in self.inputs.bfd_peers: - peer = str(bfd_peers.peer) + peer = str(bfd_peers.peer_address) vrf = bfd_peers.vrf tx_interval = bfd_peers.tx_interval rx_interval = bfd_peers.rx_interval multiplier = bfd_peers.multiplier - bfd_key = create_bfd_peer_key(vrf, peer) - bfd_output = get_value(self.instance_commands[0].json_output, f"{bfd_key}", separator="..") + 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: diff --git a/examples/tests.yaml b/examples/tests.yaml index ec0b3303f..d3213ae8f 100644 --- a/examples/tests.yaml +++ b/examples/tests.yaml @@ -51,14 +51,20 @@ anta.tests.aaa: - dot1x anta.tests.bfd: - - VerifyBFDPeers: - bfd_neighbors: - - neighbor: 192.0.255.8 + - VerifyBFDSpecificPeers: + bfd_peers: + - peer_address: 192.0.255.8 + vrf: default + - peer_address: 192.0.255.7 + vrf: default + - VerifyBFDPeersIntervals: + bfd_peers: + - peer_address: 192.0.255.8 vrf: default tx_interval: 1200000 rx_interval: 1200000 multiplier: 3 - - neighbor: 192.0.255.7 + - peer_address: 192.0.255.7 vrf: default tx_interval: 1200000 rx_interval: 1200000 diff --git a/tests/units/anta_tests/test_bfd.py b/tests/units/anta_tests/test_bfd.py index f3413002b..3374fbf0f 100644 --- a/tests/units/anta_tests/test_bfd.py +++ b/tests/units/anta_tests/test_bfd.py @@ -1,4 +1,4 @@ -# Copyright (c) 2023 Arista Networks, Inc. +# 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. """ @@ -35,13 +35,29 @@ } } } - } + }, + "MGMT": { + "ipv4Neighbors": { + "192.0.255.70": { + "peerStats": { + "": { + "peerStatsDetail": { + "operTxInterval": 1200000, + "operRxInterval": 1200000, + "detectMult": 3, + } + } + } + } + } + }, } } ], "inputs": { "bfd_peers": [ - {"peer": "192.0.255.7", "vrf": "default", "source_address": "192.0.255.1", "tx_interval": 1200000, "rx_interval": 1200000, "multiplier": 3} + {"peer_address": "192.0.255.7", "vrf": "default", "tx_interval": 1200000, "rx_interval": 1200000, "multiplier": 3}, + {"peer_address": "192.0.255.70", "vrf": "MGMT", "tx_interval": 1200000, "rx_interval": 1200000, "multiplier": 3}, ] }, "expected": {"result": "success"}, @@ -49,15 +65,54 @@ { "name": "failure-no-peer", "test": VerifyBFDPeersIntervals, - "eos_data": [{"vrfs": {}}], + "eos_data": [ + { + "vrfs": { + "default": { + "ipv4Neighbors": { + "192.0.255.7": { + "peerStats": { + "": { + "peerStatsDetail": { + "operTxInterval": 1200000, + "operRxInterval": 1200000, + "detectMult": 3, + } + } + } + } + } + }, + "MGMT": { + "ipv4Neighbors": { + "192.0.255.71": { + "peerStats": { + "": { + "peerStatsDetail": { + "operTxInterval": 1200000, + "operRxInterval": 1200000, + "detectMult": 3, + } + } + } + } + } + }, + } + } + ], "inputs": { "bfd_peers": [ - {"peer": "192.0.255.70", "vrf": "default", "source_address": "192.0.255.1", "tx_interval": 1200000, "rx_interval": 1200000, "multiplier": 3} + {"peer_address": "192.0.255.7", "vrf": "CS", "tx_interval": 1200000, "rx_interval": 1200000, "multiplier": 3}, + {"peer_address": "192.0.255.70", "vrf": "MGMT", "tx_interval": 1200000, "rx_interval": 1200000, "multiplier": 3}, ] }, "expected": { "result": "failure", - "messages": ["Following BFD peers are not configured or timers are not correct:\n{'192.0.255.70': {'default': 'Not Configured'}}"], + "messages": [ + "Following BFD peers are not configured or timers are not correct:\n" + "{'192.0.255.7': {'CS': 'Not Configured'}, '192.0.255.70': {'MGMT': 'Not Configured'}}" + ], }, }, { @@ -72,28 +127,45 @@ "peerStats": { "": { "peerStatsDetail": { - "operTxInterval": 1300000, - "operRxInterval": 1300000, + "operTxInterval": 1200001, + "operRxInterval": 1200000, "detectMult": 4, } } } } } - } + }, + "MGMT": { + "ipv4Neighbors": { + "192.0.255.70": { + "peerStats": { + "": { + "peerStatsDetail": { + "operTxInterval": 120000, + "operRxInterval": 120000, + "detectMult": 5, + } + } + } + } + } + }, } } ], "inputs": { "bfd_peers": [ - {"peer": "192.0.255.7", "vrf": "default", "source_address": "192.0.255.1", "tx_interval": 1200000, "rx_interval": 1200000, "multiplier": 3} + {"peer_address": "192.0.255.7", "vrf": "default", "tx_interval": 1200000, "rx_interval": 1200000, "multiplier": 3}, + {"peer_address": "192.0.255.70", "vrf": "MGMT", "tx_interval": 1200000, "rx_interval": 1200000, "multiplier": 3}, ] }, "expected": { "result": "failure", "messages": [ "Following BFD peers are not configured or timers are not correct:\n" - "{'192.0.255.7': {'default': {'tx_interval': 1300000, 'rx_interval': 1300000, 'multiplier': 4}}}" + "{'192.0.255.7': {'default': {'tx_interval': 1200001, 'rx_interval': 1200000, 'multiplier': 4}}, " + "'192.0.255.70': {'MGMT': {'tx_interval': 120000, 'rx_interval': 120000, 'multiplier': 5}}}" ], }, }, @@ -114,21 +186,65 @@ } } } - } + }, + "MGMT": { + "ipv4Neighbors": { + "192.0.255.70": { + "peerStats": { + "": { + "status": "up", + "remoteDisc": 108328132, + } + } + } + } + }, } } ], - "inputs": {"bfd_peers": [{"peer": "192.0.255.7", "vrf": "default", "source_address": "192.0.255.1"}]}, + "inputs": {"bfd_peers": [{"peer_address": "192.0.255.7", "vrf": "default"}, {"peer_address": "192.0.255.70", "vrf": "MGMT"}]}, "expected": {"result": "success"}, }, { "name": "failure-no-peer", "test": VerifyBFDSpecificPeers, - "eos_data": [{"vrfs": {}}], - "inputs": {"bfd_peers": [{"peer": "192.0.255.70", "vrf": "default", "source_address": "192.0.255.1"}]}, + "eos_data": [ + { + "vrfs": { + "default": { + "ipv4Neighbors": { + "192.0.255.7": { + "peerStats": { + "": { + "status": "up", + "remoteDisc": 108328132, + } + } + } + } + }, + "MGMT": { + "ipv4Neighbors": { + "192.0.255.71": { + "peerStats": { + "": { + "status": "up", + "remoteDisc": 108328132, + } + } + } + } + }, + } + } + ], + "inputs": {"bfd_peers": [{"peer_address": "192.0.255.7", "vrf": "CS"}, {"peer_address": "192.0.255.70", "vrf": "MGMT"}]}, "expected": { "result": "failure", - "messages": ["Following BFD peers are not configured, status is not up or remote disc is zero:\n{'192.0.255.70': {'default': 'Not Configured'}}"], + "messages": [ + "Following BFD peers are not configured, status is not up or remote disc is zero:\n" + "{'192.0.255.7': {'CS': 'Not Configured'}, '192.0.255.70': {'MGMT': 'Not Configured'}}" + ], }, }, { @@ -142,22 +258,35 @@ "192.0.255.7": { "peerStats": { "": { - "status": "down", + "status": "Down", + "remoteDisc": 108328132, + } + } + } + } + }, + "MGMT": { + "ipv4Neighbors": { + "192.0.255.70": { + "peerStats": { + "": { + "status": "Down", "remoteDisc": 0, } } } } - } + }, } } ], - "inputs": {"bfd_peers": [{"peer": "192.0.255.7", "vrf": "default", "source_address": "192.0.255.1"}]}, + "inputs": {"bfd_peers": [{"peer_address": "192.0.255.7", "vrf": "default"}, {"peer_address": "192.0.255.70", "vrf": "MGMT"}]}, "expected": { "result": "failure", "messages": [ "Following BFD peers are not configured, status is not up or remote disc is zero:\n" - "{'192.0.255.7': {'default': {'status': 'down', 'remote_disc': 0}}}" + "{'192.0.255.7': {'default': {'status': 'Down', 'remote_disc': 108328132}}, " + "'192.0.255.70': {'MGMT': {'status': 'Down', 'remote_disc': 0}}}" ], }, }, From aac9f073526fab2fd5c2ef6bebaa8cf2667d4345 Mon Sep 17 00:00:00 2001 From: Mahesh Kumar Date: Thu, 25 Jan 2024 07:10:51 +0000 Subject: [PATCH 8/8] issue-499: Stricted input intervals and multiplier --- anta/custom_types.py | 2 ++ anta/tests/bfd.py | 43 ++++++++++++++++-------------- examples/tests.yaml | 8 +++--- tests/units/anta_tests/test_bfd.py | 16 +++++------ 4 files changed, 37 insertions(+), 32 deletions(-) diff --git a/anta/custom_types.py b/anta/custom_types.py index 3526ddcbe..21ca7566b 100644 --- a/anta/custom_types.py +++ b/anta/custom_types.py @@ -91,3 +91,5 @@ def bgp_multiprotocol_capabilities_abbreviations(value: str) -> str: RsaKeySize = Literal[2048, 3072, 4096] EcdsaKeySize = Literal[256, 384, 521] MultiProtocolCaps = Annotated[str, BeforeValidator(bgp_multiprotocol_capabilities_abbreviations)] +BfdInterval = Annotated[int, Field(ge=50, le=60000)] +BfdMultiplier = Annotated[int, Field(ge=3, le=50)] diff --git a/anta/tests/bfd.py b/anta/tests/bfd.py index 62aeac69d..382fb5cfb 100644 --- a/anta/tests/bfd.py +++ b/anta/tests/bfd.py @@ -13,35 +13,36 @@ from pydantic import BaseModel +from anta.custom_types import BfdInterval, BfdMultiplier from anta.models import AntaCommand, AntaTest from anta.tools.get_value import get_value class VerifyBFDSpecificPeers(AntaTest): """ - This class verifies if the BFD peer's sessions are UP and remote disc is non-zero in the specified VRF. + This class verifies if the IPv4 BFD peer's sessions are UP and remote disc is non-zero in the specified VRF. Expected results: - * success: The test will pass if BFD peers are up and remote disc is non-zero in the specified VRF. - * failure: The test will fail if BFD peers are not found, the status is not UP or remote disc is zero in the specified VRF. + * success: The test will pass if IPv4 BFD peers are up and remote disc is non-zero in the specified VRF. + * failure: The test will fail if IPv4 BFD peers are not found, the status is not UP or remote disc is zero in the specified VRF. """ name = "VerifyBFDSpecificPeers" - description = "Verifies the BFD peer's sessions and remote disc in the specified VRF." + description = "Verifies the IPv4 BFD peer's sessions and remote disc in the specified VRF." categories = ["bfd"] commands = [AntaCommand(command="show bfd peers")] class Input(AntaTest.Input): """ - This class defines the input parameters of the testcase. + This class defines the input parameters of the test case. """ bfd_peers: List[BFDPeers] - """List of BFD peers""" + """List of IPv4 BFD peers""" class BFDPeers(BaseModel): """ - This class defines the details of a BFD peer. + This class defines the details of an IPv4 BFD peer. """ peer_address: IPv4Address @@ -76,21 +77,21 @@ def test(self) -> None: class VerifyBFDPeersIntervals(AntaTest): """ - This class verifies the timers of the BFD peers in the specified VRF. + This class verifies the timers of the IPv4 BFD peers in the specified VRF. Expected results: - * success: The test will pass if the timers of the BFD peers are correct in the specified VRF. - * failure: The test will fail if the BFD peers are not found or their timers are incorrect in the specified VRF. + * success: The test will pass if the timers of the IPv4 BFD peers are correct in the specified VRF. + * failure: The test will fail if the IPv4 BFD peers are not found or their timers are incorrect in the specified VRF. """ name = "VerifyBFDPeersIntervals" - description = "Verifies the timers of the BFD peers in the specified VRF." + description = "Verifies the timers of the IPv4 BFD peers in the specified VRF." categories = ["bfd"] commands = [AntaCommand(command="show bfd peers detail")] class Input(AntaTest.Input): """ - This class defines the input parameters of the testcase. + This class defines the input parameters of the test case. """ bfd_peers: List[BFDPeers] @@ -98,18 +99,18 @@ class Input(AntaTest.Input): class BFDPeers(BaseModel): """ - This class defines the details of a BFD peer. + This class defines the details of an IPv4 BFD peer. """ peer_address: IPv4Address """IPv4 address of a BFD peer""" vrf: str = "default" """Optional VRF for BGP peer. If not provided, it defaults to `default`.""" - tx_interval: int - """Tx interval of BFD peer""" - rx_interval: int - """Rx interval of BFD peer""" - multiplier: int + tx_interval: BfdInterval + """Tx interval of BFD peer in milliseconds""" + rx_interval: BfdInterval + """Rx interval of BFD peer in milliseconds""" + multiplier: BfdMultiplier """Multiplier of BFD peer""" @AntaTest.anta_test @@ -120,8 +121,10 @@ def test(self) -> None: for bfd_peers in self.inputs.bfd_peers: peer = str(bfd_peers.peer_address) vrf = bfd_peers.vrf - tx_interval = bfd_peers.tx_interval - rx_interval = bfd_peers.rx_interval + + # Converting milliseconds intervals into actual value + tx_interval = bfd_peers.tx_interval * 1000 + rx_interval = bfd_peers.rx_interval * 1000 multiplier = bfd_peers.multiplier bfd_output = get_value(self.instance_commands[0].json_output, f"vrfs..{vrf}..ipv4Neighbors..{peer}..peerStats..", separator="..") diff --git a/examples/tests.yaml b/examples/tests.yaml index d3213ae8f..95f9368a0 100644 --- a/examples/tests.yaml +++ b/examples/tests.yaml @@ -61,13 +61,13 @@ anta.tests.bfd: bfd_peers: - peer_address: 192.0.255.8 vrf: default - tx_interval: 1200000 - rx_interval: 1200000 + tx_interval: 1200 + rx_interval: 1200 multiplier: 3 - peer_address: 192.0.255.7 vrf: default - tx_interval: 1200000 - rx_interval: 1200000 + tx_interval: 1200 + rx_interval: 1200 multiplier: 3 anta.tests.configuration: diff --git a/tests/units/anta_tests/test_bfd.py b/tests/units/anta_tests/test_bfd.py index 3374fbf0f..ff00199af 100644 --- a/tests/units/anta_tests/test_bfd.py +++ b/tests/units/anta_tests/test_bfd.py @@ -56,8 +56,8 @@ ], "inputs": { "bfd_peers": [ - {"peer_address": "192.0.255.7", "vrf": "default", "tx_interval": 1200000, "rx_interval": 1200000, "multiplier": 3}, - {"peer_address": "192.0.255.70", "vrf": "MGMT", "tx_interval": 1200000, "rx_interval": 1200000, "multiplier": 3}, + {"peer_address": "192.0.255.7", "vrf": "default", "tx_interval": 1200, "rx_interval": 1200, "multiplier": 3}, + {"peer_address": "192.0.255.70", "vrf": "MGMT", "tx_interval": 1200, "rx_interval": 1200, "multiplier": 3}, ] }, "expected": {"result": "success"}, @@ -103,8 +103,8 @@ ], "inputs": { "bfd_peers": [ - {"peer_address": "192.0.255.7", "vrf": "CS", "tx_interval": 1200000, "rx_interval": 1200000, "multiplier": 3}, - {"peer_address": "192.0.255.70", "vrf": "MGMT", "tx_interval": 1200000, "rx_interval": 1200000, "multiplier": 3}, + {"peer_address": "192.0.255.7", "vrf": "CS", "tx_interval": 1200, "rx_interval": 1200, "multiplier": 3}, + {"peer_address": "192.0.255.70", "vrf": "MGMT", "tx_interval": 1200, "rx_interval": 1200, "multiplier": 3}, ] }, "expected": { @@ -127,7 +127,7 @@ "peerStats": { "": { "peerStatsDetail": { - "operTxInterval": 1200001, + "operTxInterval": 1300000, "operRxInterval": 1200000, "detectMult": 4, } @@ -156,15 +156,15 @@ ], "inputs": { "bfd_peers": [ - {"peer_address": "192.0.255.7", "vrf": "default", "tx_interval": 1200000, "rx_interval": 1200000, "multiplier": 3}, - {"peer_address": "192.0.255.70", "vrf": "MGMT", "tx_interval": 1200000, "rx_interval": 1200000, "multiplier": 3}, + {"peer_address": "192.0.255.7", "vrf": "default", "tx_interval": 1200, "rx_interval": 1200, "multiplier": 3}, + {"peer_address": "192.0.255.70", "vrf": "MGMT", "tx_interval": 1200, "rx_interval": 1200, "multiplier": 3}, ] }, "expected": { "result": "failure", "messages": [ "Following BFD peers are not configured or timers are not correct:\n" - "{'192.0.255.7': {'default': {'tx_interval': 1200001, 'rx_interval': 1200000, 'multiplier': 4}}, " + "{'192.0.255.7': {'default': {'tx_interval': 1300000, 'rx_interval': 1200000, 'multiplier': 4}}, " "'192.0.255.70': {'MGMT': {'tx_interval': 120000, 'rx_interval': 120000, 'multiplier': 5}}}" ], },