Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat(anta): Added test to validate BGP peer TTL and Max TTL Hops details #1023

Open
wants to merge 6 commits into
base: main
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions anta/input_models/routing/bgp.py
Original file line number Diff line number Diff line change
Expand Up @@ -175,6 +175,10 @@ class BgpPeer(BaseModel):
"""The warning limit for the maximum routes. `0` means no warning.

Optional field in the `VerifyBGPPeerRouteLimit` test. If not provided, the test will not verify the warning limit."""
ttl_duration: int | None = Field(default=None, ge=1, le=255)
"""The TTL duration. Required field in the `VerifyBGPPeerTtlMultiHops` test."""
max_ttl_hops: int | None = Field(default=None, ge=1, le=255)
"""The Max TTL hops. Required field in the `VerifyBGPPeerTtlMultiHops` test."""

def __str__(self) -> str:
"""Return a human-readable string representation of the BgpPeer for reporting."""
Expand Down
80 changes: 80 additions & 0 deletions anta/tests/routing/bgp.py
Original file line number Diff line number Diff line change
Expand Up @@ -1893,3 +1893,83 @@ def test(self) -> None:
failure_msg = self._validate_redistribute_route(str(vrf_data), str(address_family), afi_safi_configs, route_info)
for msg in failure_msg:
self.result.is_failure(msg)


class VerifyBGPPeerTtlMultiHops(AntaTest):
"""Verifies BGP TTL, max-ttl-hops count for BGP IPv4 peer(s).

This test performs the following checks for each specified BGP peer:

1. Verifies the specified BGP peer exists in the bgp configuration.
2. Verifies the TTL and max-ttl-hops attribute matches the expected value.

Expected Results
----------------
* Success: The test will pass if all specified peer exist with TTL and max-ttl-hops attribute matching the expected values.
* Failure: If any of the following occur:
- A specified BGP peer is not found.
- A TTL or max-ttl-hops attribute doesn't match the expected value.

Examples
--------
```yaml
anta.tests.routing:
bgp:
- VerifyBGPPeerTtlMultiHops:
bgp_peers:
- peer_address: 172.30.11.1
vrf: default
ttl_duration: 3
max_ttl_hops: 3
- peer_address: 172.30.11.2
vrf: test
ttl_duration: 30
max_ttl_hops: 30
```
"""

categories: ClassVar[list[str]] = ["bgp"]
commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show ip bgp neighbors vrf all", revision=2)]

class Input(AntaTest.Input):
"""Input model for the VerifyBGPPeerTtlMultiHops test."""

bgp_peers: list[BgpPeer]
"""List of IPv4 peer(s)."""

@field_validator("bgp_peers")
@classmethod
def validate_bgp_peers(cls, bgp_peers: list[BgpPeer]) -> list[BgpPeer]:
"""Validate that 'ttl_duration' and 'max_ttl_hops' field is provided in each BGP peer."""
for peer in bgp_peers:
if peer.ttl_duration is None:
msg = f"{peer} 'ttl_duration' field missing in the input"
raise ValueError(msg)
if peer.max_ttl_hops is None:
msg = f"{peer} 'max_ttl_hops' field missing in the input"
raise ValueError(msg)

return bgp_peers

@AntaTest.anta_test
def test(self) -> None:
"""Main test function for VerifyBGPPeerTtlMultiHops."""
self.result.is_success()
command_output = self.instance_commands[0].json_output

for peer in self.inputs.bgp_peers:
peer_ip = str(peer.peer_address)
peer_list = get_value(command_output, f"vrfs.{peer.vrf}.peerList", default=[])

# Check if the peer is found
if (peer_details := get_item(peer_list, "peerAddress", peer_ip)) is None:
self.result.is_failure(f"{peer} - Not found")
continue

# Verify if the TTL duration matches the expected value.
if peer_details.get("ttl") != peer.ttl_duration:
self.result.is_failure(f"{peer} - TTL duration mismatch - Expected: {peer.ttl_duration}, Actual: {peer_details.get('ttl')}")

# Verify if the max-ttl-hops time matches the expected value.
if peer_details.get("maxTtlHops") != peer.max_ttl_hops:
self.result.is_failure(f"{peer} - MaxTtlHops mismatch - Expected: {peer.max_ttl_hops}, Actual: {peer_details.get('maxTtlHops')}")
11 changes: 11 additions & 0 deletions examples/tests.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -494,6 +494,17 @@ anta.tests.routing.bgp:
vrf: DEV
- peer_address: 10.1.255.4
vrf: DEV
- VerifyBGPPeerTtlMultiHops:
# Verifies BGP TTL, max-ttl-hops count for BGP IPv4 peer(s).
bgp_peers:
- peer_address: 172.30.11.1
vrf: default
ttl_duration: 3
max_ttl_hops: 3
- peer_address: 172.30.11.2
vrf: test
ttl_duration: 30
max_ttl_hops: 30
- VerifyBGPPeerUpdateErrors:
# Verifies BGP update error counters for the provided BGP IPv4 peer(s).
bgp_peers:
Expand Down
201 changes: 201 additions & 0 deletions tests/units/anta_tests/routing/test_bgp.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
VerifyBGPPeerSessionRibd,
VerifyBGPPeersHealth,
VerifyBGPPeersHealthRibd,
VerifyBGPPeerTtlMultiHops,
VerifyBGPPeerUpdateErrors,
VerifyBGPRedistribution,
VerifyBGPRouteECMP,
Expand Down Expand Up @@ -6118,4 +6119,204 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo
],
},
},
{
"name": "success",
"test": VerifyBGPPeerTtlMultiHops,
"eos_data": [
{
"vrfs": {
"default": {
"peerList": [
{
"peerAddress": "10.111.0.1",
"ttl": 2,
"maxTtlHops": 2,
},
{
"peerAddress": "10.111.0.2",
"ttl": 1,
"maxTtlHops": 1,
},
]
},
"Test": {"peerList": [{"peerAddress": "10.111.0.3", "ttl": 255, "maxTtlHops": 255}]},
}
}
],
"inputs": {
"bgp_peers": [
{
"peer_address": "10.111.0.1",
"vrf": "default",
"ttl_duration": 2,
"max_ttl_hops": 2,
},
{
"peer_address": "10.111.0.2",
"vrf": "default",
"ttl_duration": 1,
"max_ttl_hops": 1,
},
{
"peer_address": "10.111.0.3",
"vrf": "Test",
"ttl_duration": 255,
"max_ttl_hops": 255,
},
]
},
"expected": {"result": "success"},
},
{
"name": "failure-peer-not-found",
"test": VerifyBGPPeerTtlMultiHops,
"eos_data": [
{
"vrfs": {
"default": {
"peerList": [
{
"peerAddress": "10.111.0.4",
"ttl": 2,
"maxTtlHops": 2,
},
{
"peerAddress": "10.111.0.5",
"ttl": 1,
"maxTtlHops": 1,
},
]
},
"Test": {"peerList": [{"peerAddress": "10.111.0.6", "ttl": 255, "maxTtlHops": 255}]},
}
}
],
"inputs": {
"bgp_peers": [
{
"peer_address": "10.111.0.1",
"vrf": "default",
"ttl_duration": 2,
"max_ttl_hops": 2,
},
{
"peer_address": "10.111.0.2",
"vrf": "Test",
"ttl_duration": 255,
"max_ttl_hops": 255,
},
]
},
"expected": {"result": "failure", "messages": ["Peer: 10.111.0.1 VRF: default - Not found", "Peer: 10.111.0.2 VRF: Test - Not found"]},
},
{
"name": "failure-ttl-time-mismatch",
"test": VerifyBGPPeerTtlMultiHops,
"eos_data": [
{
"vrfs": {
"default": {
"peerList": [
{
"peerAddress": "10.111.0.1",
"ttl": 12,
"maxTtlHops": 2,
},
{
"peerAddress": "10.111.0.2",
"ttl": 120,
"maxTtlHops": 1,
},
]
},
"Test": {"peerList": [{"peerAddress": "10.111.0.3", "ttl": 205, "maxTtlHops": 255}]},
}
}
],
"inputs": {
"bgp_peers": [
{
"peer_address": "10.111.0.1",
"vrf": "default",
"ttl_duration": 2,
"max_ttl_hops": 2,
},
{
"peer_address": "10.111.0.2",
"vrf": "default",
"ttl_duration": 1,
"max_ttl_hops": 1,
},
{
"peer_address": "10.111.0.3",
"vrf": "Test",
"ttl_duration": 255,
"max_ttl_hops": 255,
},
]
},
"expected": {
"result": "failure",
"messages": [
"Peer: 10.111.0.1 VRF: default - TTL duration mismatch - Expected: 2, Actual: 12",
"Peer: 10.111.0.2 VRF: default - TTL duration mismatch - Expected: 1, Actual: 120",
"Peer: 10.111.0.3 VRF: Test - TTL duration mismatch - Expected: 255, Actual: 205",
],
},
},
{
"name": "failure-max-ttl-hops-mismatch",
"test": VerifyBGPPeerTtlMultiHops,
"eos_data": [
{
"vrfs": {
"default": {
"peerList": [
{
"peerAddress": "10.111.0.1",
"ttl": 2,
"maxTtlHops": 12,
},
{
"peerAddress": "10.111.0.2",
"ttl": 1,
"maxTtlHops": 100,
},
]
},
"Test": {"peerList": [{"peerAddress": "10.111.0.3", "ttl": 255, "maxTtlHops": 205}]},
}
}
],
"inputs": {
"bgp_peers": [
{
"peer_address": "10.111.0.1",
"vrf": "default",
"ttl_duration": 2,
"max_ttl_hops": 2,
},
{
"peer_address": "10.111.0.2",
"vrf": "default",
"ttl_duration": 1,
"max_ttl_hops": 1,
},
{
"peer_address": "10.111.0.3",
"vrf": "Test",
"ttl_duration": 255,
"max_ttl_hops": 255,
},
]
},
"expected": {
"result": "failure",
"messages": [
"Peer: 10.111.0.1 VRF: default - MaxTtlHops mismatch - Expected: 2, Actual: 12",
"Peer: 10.111.0.2 VRF: default - MaxTtlHops mismatch - Expected: 1, Actual: 100",
"Peer: 10.111.0.3 VRF: Test - MaxTtlHops mismatch - Expected: 255, Actual: 205",
],
},
},
]
27 changes: 27 additions & 0 deletions tests/units/input_models/routing/test_bgp.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
VerifyBGPPeerGroup,
VerifyBGPPeerMPCaps,
VerifyBGPPeerRouteLimit,
VerifyBGPPeerTtlMultiHops,
VerifyBGPRouteECMP,
VerifyBgpRouteMaps,
VerifyBGPRoutePaths,
Expand Down Expand Up @@ -435,3 +436,29 @@ def test_invalid(self, afi_safi: RedistributedAfiSafi, redistributed_routes: lis
def test_valid_str(self, afi_safi: RedistributedAfiSafi, redistributed_routes: list[Any], expected: str) -> None:
"""Test AddressFamilyConfig __str__."""
assert str(AddressFamilyConfig(afi_safi=afi_safi, redistributed_routes=redistributed_routes)) == expected


class TestVerifyBGPPeerTtlMultiHopsInput:
"""Test anta.tests.routing.bgp.VerifyBGPPeerTtlMultiHops.Input."""

@pytest.mark.parametrize(
("bgp_peers"),
[
pytest.param([{"peer_address": "172.30.255.5", "vrf": "default", "ttl_duration": 3, "max_ttl_hops": 3}], id="valid"),
],
)
def test_valid(self, bgp_peers: list[BgpPeer]) -> None:
"""Test VerifyBGPPeerTtlMultiHops.Input valid inputs."""
VerifyBGPPeerTtlMultiHops.Input(bgp_peers=bgp_peers)

@pytest.mark.parametrize(
("bgp_peers"),
[
pytest.param([{"peer_address": "172.30.255.5", "vrf": "default", "ttl_duration": None, "max_ttl_hops": 3}], id="invalid-ttl-time"),
pytest.param([{"peer_address": "172.30.255.6", "vrf": "default", "ttl_duration": 3, "max_ttl_hops": None}], id="invalid-max-ttl-hops"),
],
)
def test_invalid(self, bgp_peers: list[BgpPeer]) -> None:
"""Test VerifyBGPPeerTtlMultiHops.Input invalid inputs."""
with pytest.raises(ValidationError):
VerifyBGPPeerTtlMultiHops.Input(bgp_peers=bgp_peers)
Loading