Skip to content

Commit

Permalink
feat(anta): Added the test case to verify Inbound/outbound stats for …
Browse files Browse the repository at this point in the history
…BGP neighbors (#778)
  • Loading branch information
vitthalmagadum authored Aug 15, 2024
1 parent 2258078 commit c37c089
Show file tree
Hide file tree
Showing 4 changed files with 445 additions and 1 deletion.
30 changes: 30 additions & 0 deletions anta/custom_types.py
Original file line number Diff line number Diff line change
Expand Up @@ -167,4 +167,34 @@ def validate_regex(value: str) -> str:
Hostname = Annotated[str, Field(pattern=REGEXP_TYPE_HOSTNAME)]
Port = Annotated[int, Field(ge=1, le=65535)]
RegexString = Annotated[str, AfterValidator(validate_regex)]
BgpDropStats = Literal[
"inDropAsloop",
"inDropClusterIdLoop",
"inDropMalformedMpbgp",
"inDropOrigId",
"inDropNhLocal",
"inDropNhAfV6",
"prefixDroppedMartianV4",
"prefixDroppedMaxRouteLimitViolatedV4",
"prefixDroppedMartianV6",
"prefixDroppedMaxRouteLimitViolatedV6",
"prefixLuDroppedV4",
"prefixLuDroppedMartianV4",
"prefixLuDroppedMaxRouteLimitViolatedV4",
"prefixLuDroppedV6",
"prefixLuDroppedMartianV6",
"prefixLuDroppedMaxRouteLimitViolatedV6",
"prefixEvpnDroppedUnsupportedRouteType",
"prefixBgpLsDroppedReceptionUnsupported",
"outDropV4LocalAddr",
"outDropV6LocalAddr",
"prefixVpnIpv4DroppedImportMatchFailure",
"prefixVpnIpv4DroppedMaxRouteLimitViolated",
"prefixVpnIpv6DroppedImportMatchFailure",
"prefixVpnIpv6DroppedMaxRouteLimitViolated",
"prefixEvpnDroppedImportMatchFailure",
"prefixEvpnDroppedMaxRouteLimitViolated",
"prefixRtMembershipDroppedLocalAsReject",
"prefixRtMembershipDroppedMaxRouteLimitViolated",
]
BgpUpdateError = Literal["inUpdErrWithdraw", "inUpdErrIgnore", "inUpdErrDisableAfiSafi", "disabledAfiSafi", "lastUpdErrTime"]
89 changes: 88 additions & 1 deletion anta/tests/routing/bgp.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@
from pydantic.v1.utils import deep_update
from pydantic_extra_types.mac_address import MacAddress

from anta.custom_types import Afi, BgpUpdateError, MultiProtocolCaps, Safi, Vni
from anta.custom_types import Afi, BgpDropStats, BgpUpdateError, MultiProtocolCaps, Safi, Vni
from anta.models import AntaCommand, AntaTemplate, AntaTest
from anta.tools import get_item, get_value

Expand Down Expand Up @@ -1228,6 +1228,93 @@ def test(self) -> None:
self.result.is_failure(f"Following BGP peers are not configured or hold and keep-alive timers are not correct:\n{failures}")


class VerifyBGPPeerDropStats(AntaTest):
"""Verifies BGP NLRI drop statistics for the provided BGP IPv4 peer(s).
By default, all drop statistics counters will be checked for any non-zero values.
An optional list of specific drop statistics can be provided for granular testing.
Expected Results
----------------
* Success: The test will pass if the BGP peer's drop statistic(s) are zero.
* Failure: The test will fail if the BGP peer's drop statistic(s) are non-zero/Not Found or peer is not configured.
Examples
--------
```yaml
anta.tests.routing:
bgp:
- VerifyBGPPeerDropStats:
bgp_peers:
- peer_address: 172.30.11.1
vrf: default
drop_stats:
- inDropAsloop
- prefixEvpnDroppedUnsupportedRouteType
```
"""

name = "VerifyBGPPeerDropStats"
description = "Verifies the NLRI drop statistics of a BGP IPv4 peer(s)."
categories: ClassVar[list[str]] = ["bgp"]
commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaTemplate(template="show bgp neighbors {peer} vrf {vrf}", revision=3)]

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

bgp_peers: list[BgpPeer]
"""List of BGP peers"""

class BgpPeer(BaseModel):
"""Model for a BGP peer."""

peer_address: IPv4Address
"""IPv4 address of a BGP peer."""
vrf: str = "default"
"""Optional VRF for BGP peer. If not provided, it defaults to `default`."""
drop_stats: list[BgpDropStats] | None = None
"""Optional list of drop statistics to be verified. If not provided, test will verifies all the drop statistics."""

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]

@AntaTest.anta_test
def test(self) -> None:
"""Main test function for VerifyBGPPeerDropStats."""
failures: dict[Any, Any] = {}

for command, input_entry in zip(self.instance_commands, self.inputs.bgp_peers):
peer = command.params.peer
vrf = command.params.vrf
drop_statistics = input_entry.drop_stats

# Verify BGP peer
if not (peer_list := get_value(command.json_output, f"vrfs.{vrf}.peerList")) or (peer_detail := get_item(peer_list, "peerAddress", peer)) is None:
failures[peer] = {vrf: "Not configured"}
continue

# Verify BGP peer's drop stats
drop_stats_output = peer_detail.get("dropStats", {})

# In case drop stats not provided, It will check all drop statistics
if not drop_statistics:
drop_statistics = drop_stats_output

# Verify BGP peer's drop stats
drop_stats_not_ok = {
drop_stat: drop_stats_output.get(drop_stat, "Not Found") for drop_stat in drop_statistics if drop_stats_output.get(drop_stat, "Not Found")
}
if any(drop_stats_not_ok):
failures[peer] = {vrf: drop_stats_not_ok}

# Check if any failures
if not failures:
self.result.is_success()
else:
self.result.is_failure(f"The following BGP peers are not configured or have non-zero NLRI drop statistics counters:\n{failures}")


class VerifyBGPPeerUpdateErrors(AntaTest):
"""Verifies BGP update error counters for the provided BGP IPv4 peer(s).
Expand Down
11 changes: 11 additions & 0 deletions examples/tests.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -567,6 +567,17 @@ anta.tests.routing:
vrf: default
hold_time: 180
keep_alive_time: 60
- VerifyBGPPeerDropStats:
bgp_peers:
- peer_address: 10.101.0.4
vrf: default
drop_stats:
- inDropAsloop
- inDropClusterIdLoop
- inDropMalformedMpbgp
- inDropOrigId
- inDropNhLocal
- inDropNhAfV6
- VerifyBGPPeerUpdateErrors:
bgp_peers:
- peer_address: 10.100.0.8
Expand Down
Loading

0 comments on commit c37c089

Please sign in to comment.