Skip to content

Commit

Permalink
issue-509: Added testcase to verify bgp timers
Browse files Browse the repository at this point in the history
  • Loading branch information
MaheshGSLAB committed Jan 3, 2024
1 parent f68244e commit c6767ef
Show file tree
Hide file tree
Showing 4 changed files with 234 additions and 3 deletions.
71 changes: 70 additions & 1 deletion anta/tests/routing/bgp.py
Original file line number Diff line number Diff line change
@@ -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.
"""
Expand All @@ -16,6 +16,7 @@
from anta.custom_types import Afi, Safi
from anta.models import AntaCommand, AntaTemplate, AntaTest
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

Expand Down Expand Up @@ -447,3 +448,71 @@ def test(self) -> None:

if failures:
self.result.is_failure(f"Failures: {list(failures.values())}")


class VerifyBGPTimers(AntaTest):
"""
Verifies if the BGP peers are configured with the correct hold and keep-alive timers in the specified VRF.
Expected results:
* success: The test will pass if the hold and keep-alive timers are correct for BGP peers in the specified VRF.
* failure: The test will fail if BGP peers are not found or hold and keep-alive timers are not correct in the specified VRF.
"""

name = "VerifyBGPTimers"
description = "Verifies if the BGP peers are configured with the correct hold and keep alive timers in the specified VRF."
categories = ["routing", "bgp"]
commands = [AntaCommand(command="show bgp neighbors")]

class Input(AntaTest.Input):
"""
Input parameters for the test.
"""

bgp_peers: List[BgpPeers]
"""List of BGP peers"""

class BgpPeers(BaseModel):
"""
This class defines the details of a BGP peer.
"""

peer: Union[IPv4Address, IPv6Address]
"""IPv4/IPv6 BGP peer"""
vrf: str = "default"
"""VRF context"""
hold_time: int
"""BGP hold time in seconds"""
keep_alive_time: int
"""BGP keep-alive time in seconds"""

@AntaTest.anta_test
def test(self) -> None:
failures: dict[str, Any] = {}

# Iterate over each bgp peer
for bgp_peer in self.inputs.bgp_peers:
peer_address = str(bgp_peer.peer)
vrf = bgp_peer.vrf
hold_time = bgp_peer.hold_time
keep_alive_time = bgp_peer.keep_alive_time

# Verify BGP peer
if not (bgp_output := get_value(self.instance_commands[0].json_output, f"vrfs..{vrf}..peerList", separator="..")):
failures[str(peer_address)] = {vrf: "Not configured"}
continue

bgp_index = create_index(bgp_output, "peerAddress")
bgp_output = bgp_index.get(peer_address)
if not bgp_output:
failures[str(peer_address)] = {vrf: "Not configured"}
continue

# Verify BGP peer's hold and keep alive timers
bgp_output = bgp_output[0]
if bgp_output.get("holdTime") != hold_time or bgp_output.get("keepaliveTime") != keep_alive_time:
failures[peer_address] = {vrf: {"hold_time": bgp_output.get("holdTime"), "keep_alive_time": bgp_output.get("keepaliveTime")}}

if not failures:
self.result.is_success()
else:
self.result.is_failure(f"Following BGP peers are not configured or hold and keep-alive timers are not correct:\n{failures}")
34 changes: 34 additions & 0 deletions anta/tools/utils.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,34 @@
# Copyright (c) 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.
"""
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
10 changes: 10 additions & 0 deletions examples/tests.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -308,6 +308,16 @@ anta.tests.routing:
- 10.1.255.0
- 10.1.255.2
- 10.1.255.4
- VerifyBGPTimers:
bgp_peers:
- peer: 172.30.11.1
vrf: default
hold_time: 180
keep_alive_time: 60
- peer: 172.30.11.5
vrf: default
hold_time: 180
keep_alive_time: 60
ospf:
- VerifyOSPFNeighborState:
- VerifyOSPFNeighborCount:
Expand Down
122 changes: 120 additions & 2 deletions tests/units/anta_tests/routing/test_bgp.py
Original file line number Diff line number Diff line change
@@ -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.
"""
Expand All @@ -11,7 +11,7 @@

# pylint: disable=C0413
# because of the patch above
from anta.tests.routing.bgp import VerifyBGPPeerCount, VerifyBGPPeersHealth, VerifyBGPSpecificPeers # noqa: E402
from anta.tests.routing.bgp import VerifyBGPPeerCount, VerifyBGPPeersHealth, VerifyBGPSpecificPeers, VerifyBGPTimers # noqa: E402
from tests.lib.anta import test # noqa: F401; pylint: disable=W0611

DATA: list[dict[str, Any]] = [
Expand Down Expand Up @@ -1239,4 +1239,122 @@
],
},
},
{
"name": "success",
"test": VerifyBGPTimers,
"eos_data": [
{
"vrfs": {
"default": {
"peerList": [
{
"peerAddress": "172.30.11.1",
"holdTime": 180,
"keepaliveTime": 60,
}
]
}
}
}
],
"inputs": {
"bgp_peers": [
{
"peer": "172.30.11.1",
"vrf": "default",
"hold_time": 180,
"keep_alive_time": 60,
}
]
},
"expected": {"result": "success"},
},
{
"name": "failure-no-vrf",
"test": VerifyBGPTimers,
"eos_data": [{"vrfs": {}}],
"inputs": {
"bgp_peers": [
{
"peer": "172.30.11.17",
"vrf": "MGMT",
"hold_time": 180,
"keep_alive_time": 60,
}
]
},
"expected": {
"result": "failure",
"messages": ["Following BGP peers are not configured or hold and keep-alive timers are not correct:\n{'172.30.11.17': {'MGMT': 'Not configured'}}"],
},
},
{
"name": "failure-no-peer",
"test": VerifyBGPTimers,
"eos_data": [
{
"vrfs": {
"default": {
"peerList": [
{
"peerAddress": "172.30.11.1",
"holdTime": 180,
"keepaliveTime": 60,
}
]
}
}
}
],
"inputs": {
"bgp_peers": [
{
"peer": "172.30.11.10",
"vrf": "default",
"hold_time": 180,
"keep_alive_time": 60,
}
]
},
"expected": {
"result": "failure",
"messages": ["Following BGP peers are not configured or hold and keep-alive timers are not correct:\n{'172.30.11.10': {'default': 'Not configured'}}"],
},
},
{
"name": "failure-not-correct-timers",
"test": VerifyBGPTimers,
"eos_data": [
{
"vrfs": {
"default": {
"peerList": [
{
"peerAddress": "172.30.11.1",
"holdTime": 120,
"keepaliveTime": 40,
}
]
}
}
}
],
"inputs": {
"bgp_peers": [
{
"peer": "172.30.11.1",
"vrf": "default",
"hold_time": 180,
"keep_alive_time": 60,
}
]
},
"expected": {
"result": "failure",
"messages": [
"Following BGP peers are not configured or hold and keep-alive timers are not correct:\n"
"{'172.30.11.1': {'default': {'hold_time': 120, 'keep_alive_time': 40}}}"
],
},
},
]

0 comments on commit c6767ef

Please sign in to comment.