diff --git a/anta/tests/routing/bgp.py b/anta/tests/routing/bgp.py index af28f2ed2..71097c750 100644 --- a/anta/tests/routing/bgp.py +++ b/anta/tests/routing/bgp.py @@ -15,8 +15,8 @@ from anta.custom_types import Afi, Safi from anta.models import AntaCommand, AntaTemplate, AntaTest +from anta.tools.get_item import get_item from anta.tools.get_value import get_value -from anta.tools.utils import create_index # Need to keep List for pydantic in python 3.8 @@ -461,7 +461,7 @@ class VerifyBGPPeerMD5Auth(AntaTest): name = "VerifyBGPPeerMD5Auth" description = "Verifies the MD5 authentication and state of BGP peers in a specified VRF" categories = ["routing", "bgp"] - commands = [AntaCommand(command="show bgp neighbors")] + commands = [AntaCommand(command="show bgp neighbors vrf all")] class Input(AntaTest.Input): """ @@ -476,10 +476,10 @@ class BgpPeers(BaseModel): This class defines the details of a BGP peer. """ - peer: Union[IPv4Address, IPv6Address] - """IPv4/IPv6 BGP peer""" + peer_address: IPv4Address + """IPv4 address of a BGP 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: @@ -487,26 +487,22 @@ def test(self) -> None: # Iterate over each command for bgp_peer in self.inputs.bgp_peers: - peer = str(bgp_peer.peer) + peer = str(bgp_peer.peer_address) vrf = bgp_peer.vrf # Check if BGP output exists - if not (bgp_output := get_value(self.instance_commands[0].json_output, f"vrfs.{vrf}.peerList")): - failures[str(peer)] = {vrf: "Not Configured"} - continue - - bgp_index = create_index(bgp_output, "peerAddress") - bgp_output = bgp_index.get(peer) - if not bgp_output: - failures[str(peer)] = {vrf: "Not Configured"} + if ( + not (bgp_output := get_value(self.instance_commands[0].json_output, f"vrfs.{vrf}.peerList")) + or (bgp_output := get_item(bgp_output, "peerAddress", peer)) is None + ): + failures.setdefault("bgp_peers", {})[peer] = {vrf: "Not configured"} continue # Check if BGP peer state and authentication - bgp_output = bgp_output[0] state = bgp_output.get("state") md5_auth_enabled = bgp_output.get("md5AuthEnabled") if state != "Established" or not md5_auth_enabled: - failures[str(peer)] = {vrf: {"state": state, "md5_auth_enabled": md5_auth_enabled}} + failures.setdefault("bgp_peers", {})[peer] = {vrf: {"state": state, "md5_auth_enabled": md5_auth_enabled}} # Check if there are any failures if not failures: diff --git a/anta/tools/utils.py b/anta/tools/utils.py deleted file mode 100644 index 48319078d..000000000 --- a/anta/tools/utils.py +++ /dev/null @@ -1,36 +0,0 @@ -# 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. -""" -Toolkit for ANTA. -""" -from __future__ import annotations - -from collections import defaultdict -from typing import Any, Dict, List - - -def create_index(list_of_dicts: List[Dict[str, Any]], key: str) -> Dict[str, List[Dict[str, Any]]]: - """ - Creates an index for a list of dictionaries based on a specified key. - - Parameters: - list_of_dicts (List[Dict[str, Any]]): A list of dictionaries. - key (str): The key to be indexed. - - Returns: - Dict[str, List[Dict[str, Any]]]: A dictionary where each key is a unique value from the list_of_dicts[key] - and each value is a list of dictionaries that have that value for the key. - """ - - # Initialize a default dictionary - index = defaultdict(list) - - # Iterate over each dictionary in the list - for dict_item in list_of_dicts: - # Check if the key exists in the dictionary - if key in dict_item: - # Add the dictionary to the list of its key's value - index[dict_item[key]].append(dict_item) - - return index diff --git a/examples/tests.yaml b/examples/tests.yaml index 507d87d2b..93f2ecaeb 100644 --- a/examples/tests.yaml +++ b/examples/tests.yaml @@ -310,9 +310,9 @@ anta.tests.routing: - 10.1.255.4 - VerifyBGPPeerMD5Auth: bgp_peers: - - peer: 172.30.11.1 + - peer_address: 172.30.11.1 vrf: default - - peer: 172.30.11.5 + - peer_address: 172.30.11.5 vrf: default ospf: - VerifyOSPFNeighborState: diff --git a/tests/units/anta_tests/routing/test_bgp.py b/tests/units/anta_tests/routing/test_bgp.py index 5e2fda291..9ce941dcf 100644 --- a/tests/units/anta_tests/routing/test_bgp.py +++ b/tests/units/anta_tests/routing/test_bgp.py @@ -1253,16 +1253,29 @@ "md5AuthEnabled": True, } ] - } + }, + "CS": { + "peerList": [ + { + "peerAddress": "172.30.11.10", + "state": "Established", + "md5AuthEnabled": True, + } + ] + }, } } ], "inputs": { "bgp_peers": [ { - "peer": "172.30.11.1", + "peer_address": "172.30.11.1", "vrf": "default", - } + }, + { + "peer_address": "172.30.11.10", + "vrf": "CS", + }, ] }, "expected": {"result": "success"}, @@ -1270,11 +1283,25 @@ { "name": "failure-no-vrf", "test": VerifyBGPPeerMD5Auth, - "eos_data": [{"vrfs": {}}], + "eos_data": [ + { + "vrfs": { + "default": { + "peerList": [ + { + "peerAddress": "172.30.11.10", + "state": "Established", + "md5AuthEnabled": True, + } + ] + }, + } + } + ], "inputs": { "bgp_peers": [ { - "peer": "172.30.11.1", + "peer_address": "172.30.11.1", "vrf": "MGMT", } ] @@ -1282,7 +1309,8 @@ "expected": { "result": "failure", "messages": [ - "Following BGP peers are not configured, not established or MD5 authentication is not enabled:\n{'172.30.11.1': {'MGMT': 'Not Configured'}}" + "Following BGP peers are not configured, not established or MD5 authentication is not enabled:\n" + "{'bgp_peers': {'172.30.11.1': {'MGMT': 'Not configured'}}}" ], }, }, @@ -1300,22 +1328,36 @@ "md5AuthEnabled": True, } ] - } + }, + "CS": { + "peerList": [ + { + "peerAddress": "172.30.11.11", + "state": "Established", + "md5AuthEnabled": True, + } + ] + }, } } ], "inputs": { "bgp_peers": [ { - "peer": "172.30.11.10", + "peer_address": "172.30.11.10", "vrf": "default", - } + }, + { + "peer_address": "172.30.11.11", + "vrf": "default", + }, ] }, "expected": { "result": "failure", "messages": [ - "Following BGP peers are not configured, not established or MD5 authentication is not enabled:\n{'172.30.11.10': {'default': 'Not Configured'}}" + "Following BGP peers are not configured, not established or MD5 authentication is not enabled:\n" + "{'bgp_peers': {'172.30.11.10': {'default': 'Not configured'}, '172.30.11.11': {'default': 'Not configured'}}}" ], }, }, @@ -1333,23 +1375,37 @@ "md5AuthEnabled": True, } ] - } + }, + "MGMT": { + "peerList": [ + { + "peerAddress": "172.30.11.10", + "state": "Idle", + "md5AuthEnabled": False, + } + ] + }, } } ], "inputs": { "bgp_peers": [ { - "peer": "172.30.11.1", + "peer_address": "172.30.11.1", "vrf": "default", - } + }, + { + "peer_address": "172.30.11.10", + "vrf": "MGMT", + }, ] }, "expected": { "result": "failure", "messages": [ "Following BGP peers are not configured, not established or MD5 authentication is not enabled:\n" - "{'172.30.11.1': {'default': {'state': 'Idle', 'md5_auth_enabled': True}}}" + "{'bgp_peers': {'172.30.11.1': {'default': {'state': 'Idle', 'md5_auth_enabled': True}}, " + "'172.30.11.10': {'MGMT': {'state': 'Idle', 'md5_auth_enabled': False}}}}" ], }, }, @@ -1364,25 +1420,32 @@ { "peerAddress": "172.30.11.1", "state": "Established", - } + }, + {"peerAddress": "172.30.11.10", "state": "Established", "md5AuthEnabled": False}, ] - } + }, + "MGMT": {"peerList": [{"peerAddress": "172.30.11.11", "state": "Established", "md5AuthEnabled": False}]}, } } ], "inputs": { "bgp_peers": [ { - "peer": "172.30.11.1", + "peer_address": "172.30.11.1", "vrf": "default", - } + }, + { + "peer_address": "172.30.11.11", + "vrf": "MGMT", + }, ] }, "expected": { "result": "failure", "messages": [ "Following BGP peers are not configured, not established or MD5 authentication is not enabled:\n" - "{'172.30.11.1': {'default': {'state': 'Established', 'md5_auth_enabled': None}}}" + "{'bgp_peers': {'172.30.11.1': {'default': {'state': 'Established', 'md5_auth_enabled': None}}, " + "'172.30.11.11': {'MGMT': {'state': 'Established', 'md5_auth_enabled': False}}}}" ], }, },