Skip to content

Commit

Permalink
refactor(anta.tests): Enhance VerifyBGPExchangedRoutes to support opt…
Browse files Browse the repository at this point in the history 
…ional advertised and r… (#1025)

* Enhance VerifyBGPExchangedRoutes to support optional advertised and received routes

* Added helper function to reduce cognitive complexity

* Addressed review comments: updated helper function

* Update anta/tests/routing/bgp.py

---------

Co-authored-by: Carl Baillargeon <[email protected]>
  • Loading branch information
@vitthalmagadum @carl-baillargeon
vitthalmagadum and carl-baillargeon authored Feb 6, 2025
1 parent 5780126 commit 4095ecb
Show file tree
Hide file tree
Showing 4 changed files with 299 additions and 19 deletions.
39 changes: 24 additions & 15 deletions anta/tests/routing/bgp.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@
# mypy: disable-error-code=attr-defined
from __future__ import annotations

from typing import ClassVar, TypeVar
from typing import Any, ClassVar, TypeVar

from pydantic import field_validator

Expand Down Expand Up @@ -446,8 +446,6 @@ class VerifyBGPExchangedRoutes(AntaTest):
advertised_routes:
- 192.0.255.1/32
- 192.0.254.5/32
received_routes:
- 192.0.254.3/32
```
"""

Expand All @@ -469,7 +467,7 @@ class Input(AntaTest.Input):
def validate_bgp_peers(cls, bgp_peers: list[BgpPeer]) -> list[BgpPeer]:
"""Validate that 'advertised_routes' or 'received_routes' field is provided in each BGP peer."""
for peer in bgp_peers:
if peer.advertised_routes is None or peer.received_routes is None:
if peer.advertised_routes is None and peer.received_routes is None:
msg = f"{peer} 'advertised_routes' or 'received_routes' field missing in the input"
raise ValueError(msg)
return bgp_peers
Expand All @@ -478,6 +476,20 @@ def render(self, template: AntaTemplate) -> list[AntaCommand]:
"""Render the template for each BGP peer in the input list."""
return [template.render(peer=str(bgp_peer.peer_address), vrf=bgp_peer.vrf) for bgp_peer in self.inputs.bgp_peers]

def _validate_bgp_route_paths(self, peer: str, route_type: str, route: str, entries: dict[str, Any]) -> str | None:
"""Validate the BGP route paths."""
# Check if the route is found
if route in entries:
# Check if the route is active and valid
route_paths = entries[route]["bgpRoutePaths"][0]["routeType"]
is_active = route_paths["active"]
is_valid = route_paths["valid"]
if not is_active or not is_valid:
return f"{peer} {route_type} route: {route} - Valid: {is_valid}, Active: {is_active}"
return None

return f"{peer} {route_type} route: {route} - Not found"

@AntaTest.anta_test
def test(self) -> None:
"""Main test function for VerifyBGPExchangedRoutes."""
Expand All @@ -499,19 +511,16 @@ def test(self) -> None:

# Validate both advertised and received routes
for route_type, routes in zip(["Advertised", "Received"], [peer.advertised_routes, peer.received_routes]):
# Skipping the validation for routes if user input is None
if not routes:
continue

entries = command_output[route_type]
for route in routes:
# Check if the route is found
if str(route) not in entries:
self.result.is_failure(f"{peer} {route_type} route: {route} - Not found")
continue

# Check if the route is active and valid
route_paths = entries[str(route)]["bgpRoutePaths"][0]["routeType"]
is_active = route_paths["active"]
is_valid = route_paths["valid"]
if not is_active or not is_valid:
self.result.is_failure(f"{peer} {route_type} route: {route} - Valid: {is_valid}, Active: {is_active}")
# Check if the route is found. If yes then checks the route is active and valid
failure_msg = self._validate_bgp_route_paths(str(peer), route_type, str(route), entries)
if failure_msg:
self.result.is_failure(failure_msg)


class VerifyBGPPeerMPCaps(AntaTest):
Expand Down
2 changes: 0 additions & 2 deletions examples/tests.yaml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -397,8 +397,6 @@ anta.tests.routing.bgp:
advertised_routes:
- 192.0.255.1/32
- 192.0.254.5/32
received_routes:
- 192.0.254.3/32
- VerifyBGPNlriAcceptance:
# Verifies that all received NLRI are accepted for all AFI/SAFI configured for BGP IPv4 peer(s).
bgp_peers:
Expand Down
274 changes: 274 additions & 0 deletions tests/units/anta_tests/routing/test_bgp.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1111,6 +1111,139 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo
},
"expected": {"result": "success"},
},
{
"name": "success-advertised-route-validation-only",
"test": VerifyBGPExchangedRoutes,
"eos_data": [
{
"vrfs": {
"default": {
"bgpRouteEntries": {
"192.0.254.3/32": {
"bgpRoutePaths": [
{
"routeType": {
"valid": True,
"active": True,
},
}
]
},
"192.0.254.5/32": {
"bgpRoutePaths": [
{
"routeType": {
"valid": True,
"active": True,
},
}
]
},
},
}
}
},
{
"vrfs": {
"default": {
"bgpRouteEntries": {
"192.0.254.3/32": {
"bgpRoutePaths": [
{
"routeType": {
"valid": True,
"active": True,
},
}
]
},
"192.0.254.5/32": {
"bgpRoutePaths": [
{
"routeType": {
"valid": True,
"active": True,
},
}
]
},
},
}
}
},
{
"vrfs": {
"default": {
"bgpRouteEntries": {
"192.0.254.3/32": {
"bgpRoutePaths": [
{
"routeType": {
"valid": True,
"active": False,
},
}
],
},
"192.0.255.4/32": {
"bgpRoutePaths": [
{
"routeType": {
"valid": True,
"active": False,
},
}
],
},
},
}
}
},
{
"vrfs": {
"default": {
"bgpRouteEntries": {
"192.0.254.3/32": {
"bgpRoutePaths": [
{
"routeType": {
"valid": False,
"active": True,
},
}
],
},
"192.0.255.4/32": {
"bgpRoutePaths": [
{
"routeType": {
"valid": False,
"active": True,
},
}
],
},
},
}
}
},
],
"inputs": {
"bgp_peers": [
{
"peer_address": "172.30.11.1",
"vrf": "default",
"advertised_routes": ["192.0.254.5/32", "192.0.254.3/32"],
},
{
"peer_address": "172.30.11.5",
"vrf": "default",
"advertised_routes": ["192.0.254.3/32", "192.0.254.5/32"],
},
]
},
"expected": {"result": "success"},
},
{
"name": "failure-no-routes",
"test": VerifyBGPExchangedRoutes,
Expand Down Expand Up @@ -1329,6 +1462,147 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo
],
},
},
{
"name": "failure-invalid-or-inactive-routes-as-per-given-input",
"test": VerifyBGPExchangedRoutes,
"eos_data": [
{
"vrfs": {
"default": {
"bgpRouteEntries": {
"192.0.254.3/32": {
"bgpRoutePaths": [
{
"routeType": {
"valid": False,
"active": True,
},
}
]
},
"192.0.254.5/32": {
"bgpRoutePaths": [
{
"routeType": {
"valid": True,
"active": False,
},
}
]
},
},
}
}
},
{
"vrfs": {
"default": {
"bgpRouteEntries": {
"192.0.254.3/32": {
"bgpRoutePaths": [
{
"routeType": {
"valid": True,
"active": True,
},
}
]
},
"192.0.254.5/32": {
"bgpRoutePaths": [
{
"routeType": {
"valid": True,
"active": True,
},
}
]
},
},
}
}
},
{
"vrfs": {
"default": {
"bgpRouteEntries": {
"192.0.254.3/32": {
"bgpRoutePaths": [
{
"routeType": {
"valid": True,
"active": True,
},
}
],
},
"192.0.255.4/32": {
"bgpRoutePaths": [
{
"routeType": {
"valid": True,
"active": True,
},
}
],
},
},
}
}
},
{
"vrfs": {
"default": {
"bgpRouteEntries": {
"192.0.254.3/32": {
"bgpRoutePaths": [
{
"routeType": {
"valid": True,
"active": False,
},
}
],
},
"192.0.255.4/32": {
"bgpRoutePaths": [
{
"routeType": {
"valid": True,
"active": False,
},
}
],
},
},
}
}
},
],
"inputs": {
"bgp_peers": [
{
"peer_address": "172.30.11.1",
"vrf": "default",
"advertised_routes": ["192.0.254.3/32", "192.0.254.51/32"],
},
{
"peer_address": "172.30.11.5",
"vrf": "default",
"received_routes": ["192.0.254.3/32", "192.0.255.41/32"],
},
]
},
"expected": {
"result": "failure",
"messages": [
"Peer: 172.30.11.1 VRF: default Advertised route: 192.0.254.3/32 - Valid: False, Active: True",
"Peer: 172.30.11.1 VRF: default Advertised route: 192.0.254.51/32 - Not found",
"Peer: 172.30.11.5 VRF: default Received route: 192.0.254.3/32 - Valid: True, Active: False",
"Peer: 172.30.11.5 VRF: default Received route: 192.0.255.41/32 - Not found",
],
},
},
{
"name": "success",
"test": VerifyBGPPeerMPCaps,
Expand Down
3 changes: 1 addition & 2 deletions tests/units/input_models/routing/test_bgp.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -120,6 +120,7 @@ class TestVerifyBGPExchangedRoutesInput:
[{"peer_address": "172.30.255.5", "vrf": "default", "advertised_routes": ["192.0.254.5/32"], "received_routes": ["192.0.255.4/32"]}],
id="valid_both_received_advertised",
),
pytest.param([{"peer_address": "172.30.255.5", "vrf": "default", "advertised_routes": ["192.0.254.5/32"]}], id="valid_advertised_routes"),
],
)
def test_valid(self, bgp_peers: list[BgpPeer]) -> None:
Expand All @@ -130,8 +131,6 @@ def test_valid(self, bgp_peers: list[BgpPeer]) -> None:
("bgp_peers"),
[
pytest.param([{"peer_address": "172.30.255.5", "vrf": "default"}], id="invalid"),
pytest.param([{"peer_address": "172.30.255.5", "vrf": "default", "advertised_routes": ["192.0.254.5/32"]}], id="invalid_received_route"),
pytest.param([{"peer_address": "172.30.255.5", "vrf": "default", "received_routes": ["192.0.254.5/32"]}], id="invalid_advertised_route"),
],
)
def test_invalid(self, bgp_peers: list[BgpPeer]) -> None:
Expand Down

0 comments on commit 4095ecb

Please sign in to comment.