From 891b8ba74e8be80e4b052c67e4351a23136149af Mon Sep 17 00:00:00 2001 From: Guillaume Mulocher Date: Tue, 24 Dec 2024 16:51:45 +0100 Subject: [PATCH 1/4] ci: Add environment to protect publishing (#979) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * ci: Add environment to protect publishing * Update .github/workflows/release.yml * Update .github/workflows/release.yml * Update .github/workflows/release.yml --------- Co-authored-by: Matthieu Tâche --- .github/workflows/release.yml | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index d32be46df..4b3b35705 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -7,8 +7,13 @@ on: jobs: pypi: - name: Publish version to Pypi servers + name: Publish Python 🐍 distribution 📦 to PyPI runs-on: ubuntu-latest + environment: + name: production + url: https://pypi.org/p/anta + permissions: + id-token: write steps: - name: Checkout code uses: actions/checkout@v4 @@ -19,11 +24,8 @@ jobs: - name: Build package run: | python -m build - - name: Publish package to Pypi + - name: Publish distribution 📦 to PyPI uses: pypa/gh-action-pypi-publish@release/v1 - with: - user: __token__ - password: ${{ secrets.PYPI_API_TOKEN }} release-coverage: name: Updated ANTA release coverage badge From d8d96903aada0f45538af2c8911fb45ec56670c8 Mon Sep 17 00:00:00 2001 From: Carl Baillargeon Date: Tue, 24 Dec 2024 10:57:34 -0500 Subject: [PATCH 2/4] fix(anta.tests): Second round of cleaning up BGP tests module (#914) --- anta/input_models/routing/bgp.py | 87 +- anta/reporter/md_reporter.py | 5 +- anta/tests/routing/bgp.py | 1030 ++++++++---------- examples/tests.yaml | 16 +- tests/units/anta_tests/routing/test_bgp.py | 956 +++------------- tests/units/input_models/routing/test_bgp.py | 144 ++- 6 files changed, 843 insertions(+), 1395 deletions(-) diff --git a/anta/input_models/routing/bgp.py b/anta/input_models/routing/bgp.py index a291809c6..57c821740 100644 --- a/anta/input_models/routing/bgp.py +++ b/anta/input_models/routing/bgp.py @@ -5,13 +5,14 @@ from __future__ import annotations -from ipaddress import IPv4Address, IPv6Address +from ipaddress import IPv4Address, IPv4Network, IPv6Address from typing import TYPE_CHECKING, Any from warnings import warn -from pydantic import BaseModel, ConfigDict, PositiveInt, model_validator +from pydantic import BaseModel, ConfigDict, Field, PositiveInt, model_validator +from pydantic_extra_types.mac_address import MacAddress -from anta.custom_types import Afi, Safi +from anta.custom_types import Afi, BgpDropStats, BgpUpdateError, MultiProtocolCaps, Safi, Vni if TYPE_CHECKING: import sys @@ -97,7 +98,7 @@ def eos_key(self) -> str: return AFI_SAFI_EOS_KEY[(self.afi, self.safi)] def __str__(self) -> str: - """Return a string representation of the BgpAddressFamily model. Used in failure messages. + """Return a human-readable string representation of the BgpAddressFamily for reporting. Examples -------- @@ -128,3 +129,81 @@ def __init__(self, **data: Any) -> None: # noqa: ANN401 stacklevel=2, ) super().__init__(**data) + + +class BgpPeer(BaseModel): + """Model for a BGP peer. + + Only IPv4 peers are supported for now. + """ + + model_config = ConfigDict(extra="forbid") + peer_address: IPv4Address + """IPv4 address of the BGP peer.""" + vrf: str = "default" + """Optional VRF for the BGP peer. Defaults to `default`.""" + advertised_routes: list[IPv4Network] | None = None + """List of advertised routes in CIDR format. Required field in the `VerifyBGPExchangedRoutes` test.""" + received_routes: list[IPv4Network] | None = None + """List of received routes in CIDR format. Required field in the `VerifyBGPExchangedRoutes` test.""" + capabilities: list[MultiProtocolCaps] | None = None + """List of BGP multiprotocol capabilities. Required field in the `VerifyBGPPeerMPCaps` test.""" + strict: bool = False + """If True, requires exact match of the provided BGP multiprotocol capabilities. + + Optional field in the `VerifyBGPPeerMPCaps` test. Defaults to False.""" + hold_time: int | None = Field(default=None, ge=3, le=7200) + """BGP hold time in seconds. Required field in the `VerifyBGPTimers` test.""" + keep_alive_time: int | None = Field(default=None, ge=0, le=3600) + """BGP keepalive time in seconds. Required field in the `VerifyBGPTimers` test.""" + drop_stats: list[BgpDropStats] | None = None + """List of drop statistics to be verified. + + Optional field in the `VerifyBGPPeerDropStats` test. If not provided, the test will verifies all drop statistics.""" + update_errors: list[BgpUpdateError] | None = None + """List of update error counters to be verified. + + Optional field in the `VerifyBGPPeerUpdateErrors` test. If not provided, the test will verifies all the update error counters.""" + inbound_route_map: str | None = None + """Inbound route map applied, defaults to None. Required field in the `VerifyBgpRouteMaps` test.""" + outbound_route_map: str | None = None + """Outbound route map applied, defaults to None. Required field in the `VerifyBgpRouteMaps` test.""" + maximum_routes: int | None = Field(default=None, ge=0, le=4294967294) + """The maximum allowable number of BGP routes, `0` means unlimited. Required field in the `VerifyBGPPeerRouteLimit` test""" + warning_limit: int | None = Field(default=None, ge=0, le=4294967294) + """Optional maximum routes warning limit. If not provided, it defaults to `0` meaning no warning limit.""" + + def __str__(self) -> str: + """Return a human-readable string representation of the BgpPeer for reporting.""" + return f"Peer: {self.peer_address} VRF: {self.vrf}" + + +class BgpNeighbor(BgpPeer): # pragma: no cover + """Alias for the BgpPeer model to maintain backward compatibility. + + When initialised, it will emit a deprecation warning and call the BgpPeer model. + + TODO: Remove this class in ANTA v2.0.0. + """ + + def __init__(self, **data: Any) -> None: # noqa: ANN401 + """Initialize the BgpPeer class, emitting a depreciation warning.""" + warn( + message="BgpNeighbor model is deprecated and will be removed in ANTA v2.0.0. Use the BgpPeer model instead.", + category=DeprecationWarning, + stacklevel=2, + ) + super().__init__(**data) + + +class VxlanEndpoint(BaseModel): + """Model for a VXLAN endpoint.""" + + address: IPv4Address | MacAddress + """IPv4 or MAC address of the VXLAN endpoint.""" + vni: Vni + """VNI of the VXLAN endpoint.""" + + def __str__(self) -> str: + """Return a human-readable string representation of the VxlanEndpoint for reporting.""" + return f"Address: {self.address} VNI: {self.vni}" diff --git a/anta/reporter/md_reporter.py b/anta/reporter/md_reporter.py index be3e86faf..94c4a8668 100644 --- a/anta/reporter/md_reporter.py +++ b/anta/reporter/md_reporter.py @@ -8,7 +8,7 @@ import logging import re from abc import ABC, abstractmethod -from typing import TYPE_CHECKING, ClassVar +from typing import TYPE_CHECKING, ClassVar, TextIO from anta.constants import MD_REPORT_TOC from anta.logger import anta_log_exception @@ -17,7 +17,6 @@ if TYPE_CHECKING: from collections.abc import Generator - from io import TextIOWrapper from pathlib import Path from anta.result_manager import ResultManager @@ -72,7 +71,7 @@ class MDReportBase(ABC): to generate and write content to the provided markdown file. """ - def __init__(self, mdfile: TextIOWrapper, results: ResultManager) -> None: + def __init__(self, mdfile: TextIO, results: ResultManager) -> None: """Initialize the MDReportBase with an open markdown file object to write to and a ResultManager instance. Parameters diff --git a/anta/tests/routing/bgp.py b/anta/tests/routing/bgp.py index f44729ec2..2a140ddb2 100644 --- a/anta/tests/routing/bgp.py +++ b/anta/tests/routing/bgp.py @@ -7,80 +7,16 @@ # mypy: disable-error-code=attr-defined from __future__ import annotations -from ipaddress import IPv4Address, IPv4Network -from typing import TYPE_CHECKING, Any, ClassVar +from typing import ClassVar, TypeVar -from pydantic import BaseModel, Field, field_validator, model_validator -from pydantic.v1.utils import deep_update -from pydantic_extra_types.mac_address import MacAddress +from pydantic import field_validator -from anta.custom_types import BgpDropStats, BgpUpdateError, MultiProtocolCaps, Vni -from anta.input_models.routing.bgp import BgpAddressFamily, BgpAfi +from anta.input_models.routing.bgp import BgpAddressFamily, BgpAfi, BgpNeighbor, BgpPeer, VxlanEndpoint from anta.models import AntaCommand, AntaTemplate, AntaTest from anta.tools import format_data, get_item, get_value -if TYPE_CHECKING: - import sys - - if sys.version_info >= (3, 11): - from typing import Self - else: - from typing_extensions import Self - - -def _add_bgp_routes_failure( - bgp_routes: list[str], bgp_output: dict[str, Any], peer: str, vrf: str, route_type: str = "advertised_routes" -) -> dict[str, dict[str, dict[str, dict[str, list[str]]]]]: - """Identify missing BGP routes and invalid or inactive route entries. - - This function checks the BGP output from the device against the expected routes. - - It identifies any missing routes as well as any routes that are invalid or inactive. The results are returned in a dictionary. - - Parameters - ---------- - bgp_routes - The list of expected routes. - bgp_output - The BGP output from the device. - peer - The IP address of the BGP peer. - vrf - The name of the VRF for which the routes need to be verified. - route_type - The type of BGP routes. Defaults to 'advertised_routes'. - - Returns - ------- - dict[str, dict[str, dict[str, dict[str, list[str]]]]] - A dictionary containing the missing routes and invalid or inactive routes. - - """ - # Prepare the failure routes dictionary - failure_routes: dict[str, dict[str, Any]] = {} - - # Iterate over the expected BGP routes - for route in bgp_routes: - str_route = str(route) - failure: dict[str, Any] = {"bgp_peers": {peer: {vrf: {route_type: {}}}}} - - # Check if the route is missing in the BGP output - if str_route not in bgp_output: - # If missing, add it to the failure routes dictionary - failure["bgp_peers"][peer][vrf][route_type][str_route] = "Not found" - failure_routes = deep_update(failure_routes, failure) - continue - - # Check if the route is active and valid - is_active = bgp_output[str_route]["bgpRoutePaths"][0]["routeType"]["valid"] - is_valid = bgp_output[str_route]["bgpRoutePaths"][0]["routeType"]["active"] - - # If the route is either inactive or invalid, add it to the failure routes dictionary - if not is_active or not is_valid: - failure["bgp_peers"][peer][vrf][route_type][str_route] = {"valid": is_valid, "active": is_active} - failure_routes = deep_update(failure_routes, failure) - - return failure_routes +# Using a TypeVar for the BgpPeer model since mypy thinks it's a ClassVar and not a valid type when used in field validators +T = TypeVar("T", bound=BgpPeer) def _check_bgp_neighbor_capability(capability_status: dict[str, bool]) -> bool: @@ -397,12 +333,20 @@ def test(self) -> None: class VerifyBGPExchangedRoutes(AntaTest): """Verifies the advertised and received routes of BGP peers. - The route type should be 'valid' and 'active' for a specified VRF. + This test performs the following checks for each specified peer: + + For each advertised and received route: + - Confirms that the route exists in the BGP route table. + - Verifies that the route is in an 'active' and 'valid' state. Expected Results ---------------- - * Success: If the BGP peers have correctly advertised and received routes of type 'valid' and 'active' for a specified VRF. - * Failure: If a BGP peer is not found, the expected advertised/received routes are not found, or the routes are not 'valid' or 'active'. + * Success: If all of the following conditions are met: + - All specified advertised/received routes are found in the BGP route table. + - All routes are in both 'active' and 'valid' states. + * Failure: If any of the following occur: + - An advertised/received route is not found in the BGP route table. + - Any route is not in an 'active' or 'valid' state. Examples -------- @@ -436,71 +380,87 @@ class VerifyBGPExchangedRoutes(AntaTest): class Input(AntaTest.Input): """Input model for the VerifyBGPExchangedRoutes test.""" - bgp_peers: list[BgpNeighbor] - """List of BGP neighbors.""" - - class BgpNeighbor(BaseModel): - """Model for a BGP neighbor.""" + bgp_peers: list[BgpPeer] + """List of BGP peers.""" + BgpNeighbor: ClassVar[type[BgpNeighbor]] = BgpNeighbor - peer_address: IPv4Address - """IPv4 address of a BGP peer.""" - vrf: str = "default" - """Optional VRF for BGP peer. If not provided, it defaults to `default`.""" - advertised_routes: list[IPv4Network] - """List of advertised routes in CIDR format.""" - received_routes: list[IPv4Network] - """List of received routes in CIDR format.""" + @field_validator("bgp_peers") + @classmethod + def validate_bgp_peers(cls, bgp_peers: list[BgpPeer]) -> list[BgpPeer]: + """Validate that 'advertised_routes' or 'received_routes' field is provided in each address family.""" + for peer in bgp_peers: + if peer.advertised_routes is None or peer.received_routes is None: + msg = f"{peer} 'advertised_routes' or 'received_routes' field missing in the input" + raise ValueError(msg) + return bgp_peers def render(self, template: AntaTemplate) -> list[AntaCommand]: - """Render the template for each BGP neighbor in the input list.""" + """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 VerifyBGPExchangedRoutes.""" - failures: dict[str, dict[str, Any]] = {"bgp_peers": {}} - - # Iterating over command output for different peers - for command in self.instance_commands: - peer = command.params.peer - vrf = command.params.vrf - for input_entry in self.inputs.bgp_peers: - if str(input_entry.peer_address) == peer and input_entry.vrf == vrf: - advertised_routes = input_entry.advertised_routes - received_routes = input_entry.received_routes - break - failure = {vrf: ""} - - # Verify if a BGP peer is configured with the provided vrf - if not (bgp_routes := get_value(command.json_output, f"vrfs.{vrf}.bgpRouteEntries")): - failure[vrf] = "Not configured" - failures["bgp_peers"][peer] = failure - continue + self.result.is_success() - # Validate advertised routes - if "advertised-routes" in command.command: - failure_routes = _add_bgp_routes_failure(advertised_routes, bgp_routes, peer, vrf) + num_peers = len(self.inputs.bgp_peers) - # Validate received routes - else: - failure_routes = _add_bgp_routes_failure(received_routes, bgp_routes, peer, vrf, route_type="received_routes") - failures = deep_update(failures, failure_routes) + # Process each peer and its corresponding command pair + for peer_idx, peer in enumerate(self.inputs.bgp_peers): + # For n peers, advertised routes are at indices 0 to n-1, and received routes are at indices n to 2n-1 + advertised_routes_cmd = self.instance_commands[peer_idx] + received_routes_cmd = self.instance_commands[peer_idx + num_peers] + + # Get the BGP route entries of each command + command_output = { + "Advertised": get_value(advertised_routes_cmd.json_output, f"vrfs.{peer.vrf}.bgpRouteEntries", default={}), + "Received": get_value(received_routes_cmd.json_output, f"vrfs.{peer.vrf}.bgpRouteEntries", default={}), + } + + # Validate both advertised and received routes + for route_type, routes in zip(["Advertised", "Received"], [peer.advertised_routes, peer.received_routes]): + 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 - if not failures["bgp_peers"]: - self.result.is_success() - else: - self.result.is_failure(f"Following BGP peers are not found or routes are not exchanged properly:\n{failures}") + # 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}") class VerifyBGPPeerMPCaps(AntaTest): - """Verifies the multiprotocol capabilities of a BGP peer in a specified VRF. + """Verifies the multiprotocol capabilities of BGP peers. + + This test performs the following checks for each specified peer: - Supports `strict: True` to verify that only the specified capabilities are configured, requiring an exact match. + 1. Confirms that the specified VRF is configured. + 2. Verifies that the peer exists in the BGP configuration. + 3. For each specified capability: + - Validates that the capability is present in the peer configuration. + - Confirms that the capability is advertised, received, and enabled. + 4. When strict mode is enabled (`strict: true`): + - Verifies that only the specified capabilities are configured. + - Ensures an exact match between configured and expected capabilities. Expected Results ---------------- - * Success: The test will pass if the BGP peer's multiprotocol capabilities are advertised, received, and enabled in the specified VRF. - * Failure: The test will fail if BGP peers are not found or multiprotocol capabilities are not advertised, received, and enabled in the specified VRF. + * Success: If all of the following conditions are met: + - The specified VRF is configured. + - All specified peers are found in the BGP configuration. + - All specified capabilities are present and properly negotiated. + - In strict mode, only the specified capabilities are configured. + * Failure: If any of the following occur: + - The specified VRF is not configured. + - A specified peer is not found in the BGP configuration. + - A specified capability is not found. + - A capability is not properly negotiated (not advertised, received, or enabled). + - In strict mode, additional or missing capabilities are detected. Examples -------- @@ -517,7 +477,6 @@ class VerifyBGPPeerMPCaps(AntaTest): ``` """ - description = "Verifies the multiprotocol capabilities of a BGP peer." categories: ClassVar[list[str]] = ["bgp"] commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show bgp neighbors vrf all", revision=3)] @@ -526,78 +485,77 @@ class Input(AntaTest.Input): bgp_peers: list[BgpPeer] """List of BGP peers""" + BgpPeer: ClassVar[type[BgpPeer]] = BgpPeer - 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`.""" - strict: bool = False - """If True, requires exact matching of provided capabilities. Defaults to False.""" - capabilities: list[MultiProtocolCaps] - """List of multiprotocol capabilities to be verified.""" + @field_validator("bgp_peers") + @classmethod + def validate_bgp_peers(cls, bgp_peers: list[T]) -> list[T]: + """Validate that 'capabilities' field is provided in each address family.""" + for peer in bgp_peers: + if peer.capabilities is None: + msg = f"{peer} 'capabilities' field missing in the input" + raise ValueError(msg) + return bgp_peers @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyBGPPeerMPCaps.""" - failures: dict[str, Any] = {"bgp_peers": {}} - - # Iterate over each bgp peer. - for bgp_peer in self.inputs.bgp_peers: - peer = str(bgp_peer.peer_address) - vrf = bgp_peer.vrf - capabilities = bgp_peer.capabilities - failure: dict[str, dict[str, dict[str, Any]]] = {"bgp_peers": {peer: {vrf: {}}}} - - # Check if BGP output exists. - 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 - ): - failure["bgp_peers"][peer][vrf] = {"status": "Not configured"} - failures = deep_update(failures, failure) + self.result.is_success() + + output = self.instance_commands[0].json_output + + for peer in self.inputs.bgp_peers: + peer_ip = str(peer.peer_address) + + # Check if the VRF is configured + if (vrf_output := get_value(output, f"vrfs.{peer.vrf}")) is None: + self.result.is_failure(f"{peer} - VRF not configured") + continue + + # Check if the peer is found + if (peer_data := get_item(vrf_output["peerList"], "peerAddress", peer_ip)) is None: + self.result.is_failure(f"{peer} - Not found") continue - # Fetching the capabilities output. - bgp_output = get_value(bgp_output, "neighborCapabilities.multiprotocolCaps") + # Fetching the multiprotocol capabilities + act_mp_caps = get_value(peer_data, "neighborCapabilities.multiprotocolCaps") - if bgp_peer.strict and sorted(capabilities) != sorted(bgp_output): - failure["bgp_peers"][peer][vrf] = { - "status": f"Expected only `{', '.join(capabilities)}` capabilities should be listed but found `{', '.join(bgp_output)}` instead." - } - failures = deep_update(failures, failure) + # If strict is True, check if only the specified capabilities are configured + if peer.strict and sorted(peer.capabilities) != sorted(act_mp_caps): + self.result.is_failure(f"{peer} - Mismatch - Expected: {', '.join(peer.capabilities)} Actual: {', '.join(act_mp_caps)}") continue # Check each capability - for capability in capabilities: - capability_output = bgp_output.get(capability) + for capability in peer.capabilities: + # Check if the capability is found + if (capability_status := get_value(act_mp_caps, capability)) is None: + self.result.is_failure(f"{peer} - {capability} not found") - # Check if capabilities are missing - if not capability_output: - failure["bgp_peers"][peer][vrf][capability] = "not found" - failures = deep_update(failures, failure) + # Check if the capability is advertised, received, and enabled + elif not _check_bgp_neighbor_capability(capability_status): + self.result.is_failure(f"{peer} - {capability} not negotiated - {format_data(capability_status)}") - # Check if capabilities are not advertised, received, or enabled - elif not all(capability_output.get(prop, False) for prop in ["advertised", "received", "enabled"]): - failure["bgp_peers"][peer][vrf][capability] = capability_output - failures = deep_update(failures, failure) - # Check if there are any failures - if not failures["bgp_peers"]: - self.result.is_success() - else: - self.result.is_failure(f"Following BGP peer multiprotocol capabilities are not found or not ok:\n{failures}") +class VerifyBGPPeerASNCap(AntaTest): + """Verifies the four octet ASN capability of BGP peers. + This test performs the following checks for each specified peer: -class VerifyBGPPeerASNCap(AntaTest): - """Verifies the four octet asn capabilities of a BGP peer in a specified VRF. + 1. Confirms that the specified VRF is configured. + 2. Verifies that the peer exists in the BGP configuration. + 3. Validates that the capability is present in the peer configuration. + 4. Confirms that the capability is advertised, received, and enabled. Expected Results ---------------- - * Success: The test will pass if BGP peer's four octet asn capabilities are advertised, received, and enabled in the specified VRF. - * Failure: The test will fail if BGP peers are not found or four octet asn capabilities are not advertised, received, and enabled in the specified VRF. + * Success: If all of the following conditions are met: + - All specified peers are found in the BGP configuration. + - The four octet ASN capability is present in each peer configuration. + - The capability is properly negotiated (advertised, received, and enabled) for all peers. + * Failure: If any of the following occur: + - A specified peer is not found in the BGP configuration. + - The four octet ASN capability is not present for a peer. + - The capability is not properly negotiated (not advertised, received, or enabled) for any peer. Examples -------- @@ -611,7 +569,6 @@ class VerifyBGPPeerASNCap(AntaTest): ``` """ - description = "Verifies the four octet asn capabilities of a BGP peer." categories: ClassVar[list[str]] = ["bgp"] commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show bgp neighbors vrf all", revision=3)] @@ -620,61 +577,54 @@ class Input(AntaTest.Input): 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`.""" + BgpPeer: ClassVar[type[BgpPeer]] = BgpPeer @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyBGPPeerASNCap.""" - failures: dict[str, Any] = {"bgp_peers": {}} - - # Iterate over each bgp peer - for bgp_peer in self.inputs.bgp_peers: - peer = str(bgp_peer.peer_address) - vrf = bgp_peer.vrf - failure: dict[str, dict[str, dict[str, Any]]] = {"bgp_peers": {peer: {vrf: {}}}} - - # Check if BGP output exists - 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 - ): - failure["bgp_peers"][peer][vrf] = {"status": "Not configured"} - failures = deep_update(failures, failure) - continue + self.result.is_success() + + output = self.instance_commands[0].json_output - bgp_output = get_value(bgp_output, "neighborCapabilities.fourOctetAsnCap") + for peer in self.inputs.bgp_peers: + peer_ip = str(peer.peer_address) + peer_list = get_value(output, f"vrfs.{peer.vrf}.peerList", default=[]) - # Check if four octet asn capabilities are found - if not bgp_output: - failure["bgp_peers"][peer][vrf] = {"fourOctetAsnCap": "not found"} - failures = deep_update(failures, failure) + # Check if the peer is found + if (peer_data := get_item(peer_list, "peerAddress", peer_ip)) is None: + self.result.is_failure(f"{peer} - Not found") + continue - # Check if capabilities are not advertised, received, or enabled - elif not all(bgp_output.get(prop, False) for prop in ["advertised", "received", "enabled"]): - failure["bgp_peers"][peer][vrf] = {"fourOctetAsnCap": bgp_output} - failures = deep_update(failures, failure) + # Check if the 4-octet ASN capability is found + if (capablity_status := get_value(peer_data, "neighborCapabilities.fourOctetAsnCap")) is None: + self.result.is_failure(f"{peer} - 4-octet ASN capability not found") + continue - # Check if there are any failures - if not failures["bgp_peers"]: - self.result.is_success() - else: - self.result.is_failure(f"Following BGP peer four octet asn capabilities are not found or not ok:\n{failures}") + # Check if the 4-octet ASN capability is advertised, received, and enabled + if not _check_bgp_neighbor_capability(capablity_status): + self.result.is_failure(f"{peer} - 4-octet ASN capability not negotiated - {format_data(capablity_status)}") class VerifyBGPPeerRouteRefreshCap(AntaTest): """Verifies the route refresh capabilities of a BGP peer in a specified VRF. + This test performs the following checks for each specified peer: + + 1. Confirms that the specified VRF is configured. + 2. Verifies that the peer exists in the BGP configuration. + 3. Validates that the route refresh capability is present in the peer configuration. + 4. Confirms that the capability is advertised, received, and enabled. + Expected Results ---------------- - * Success: The test will pass if the BGP peer's route refresh capabilities are advertised, received, and enabled in the specified VRF. - * Failure: The test will fail if BGP peers are not found or route refresh capabilities are not advertised, received, and enabled in the specified VRF. + * Success: If all of the following conditions are met: + - All specified peers are found in the BGP configuration. + - The route refresh capability is present in each peer configuration. + - The capability is properly negotiated (advertised, received, and enabled) for all peers. + * Failure: If any of the following occur: + - A specified peer is not found in the BGP configuration. + - The route refresh capability is not present for a peer. + - The capability is not properly negotiated (not advertised, received, or enabled) for any peer. Examples -------- @@ -688,7 +638,6 @@ class VerifyBGPPeerRouteRefreshCap(AntaTest): ``` """ - description = "Verifies the route refresh capabilities of a BGP peer." categories: ClassVar[list[str]] = ["bgp"] commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show bgp neighbors vrf all", revision=3)] @@ -697,61 +646,54 @@ class Input(AntaTest.Input): 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`.""" + BgpPeer: ClassVar[type[BgpPeer]] = BgpPeer @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyBGPPeerRouteRefreshCap.""" - failures: dict[str, Any] = {"bgp_peers": {}} - - # Iterate over each bgp peer - for bgp_peer in self.inputs.bgp_peers: - peer = str(bgp_peer.peer_address) - vrf = bgp_peer.vrf - failure: dict[str, dict[str, dict[str, Any]]] = {"bgp_peers": {peer: {vrf: {}}}} - - # Check if BGP output exists - 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 - ): - failure["bgp_peers"][peer][vrf] = {"status": "Not configured"} - failures = deep_update(failures, failure) - continue + self.result.is_success() - bgp_output = get_value(bgp_output, "neighborCapabilities.routeRefreshCap") + output = self.instance_commands[0].json_output - # Check if route refresh capabilities are found - if not bgp_output: - failure["bgp_peers"][peer][vrf] = {"routeRefreshCap": "not found"} - failures = deep_update(failures, failure) + for peer in self.inputs.bgp_peers: + peer_ip = str(peer.peer_address) + peer_list = get_value(output, f"vrfs.{peer.vrf}.peerList", default=[]) - # Check if capabilities are not advertised, received, or enabled - elif not all(bgp_output.get(prop, False) for prop in ["advertised", "received", "enabled"]): - failure["bgp_peers"][peer][vrf] = {"routeRefreshCap": bgp_output} - failures = deep_update(failures, failure) + # Check if the peer is found + if (peer_data := get_item(peer_list, "peerAddress", peer_ip)) is None: + self.result.is_failure(f"{peer} - Not found") + continue + + # Check if the route refresh capability is found + if (capablity_status := get_value(peer_data, "neighborCapabilities.routeRefreshCap")) is None: + self.result.is_failure(f"{peer} - Route refresh capability not found") + continue - # Check if there are any failures - if not failures["bgp_peers"]: - self.result.is_success() - else: - self.result.is_failure(f"Following BGP peer route refresh capabilities are not found or not ok:\n{failures}") + # Check if the route refresh capability is advertised, received, and enabled + if not _check_bgp_neighbor_capability(capablity_status): + self.result.is_failure(f"{peer} - Route refresh capability not negotiated - {format_data(capablity_status)}") class VerifyBGPPeerMD5Auth(AntaTest): """Verifies the MD5 authentication and state of IPv4 BGP peers in a specified VRF. + This test performs the following checks for each specified peer: + + 1. Confirms that the specified VRF is configured. + 2. Verifies that the peer exists in the BGP configuration. + 3. Validates that the BGP session is in `Established` state. + 4. Confirms that MD5 authentication is enabled for the peer. + Expected Results ---------------- - * Success: The test will pass if IPv4 BGP peers are configured with MD5 authentication and state as established in the specified VRF. - * Failure: The test will fail if IPv4 BGP peers are not found, state is not as established or MD5 authentication is not enabled in the specified VRF. + * Success: If all of the following conditions are met: + - All specified peers are found in the BGP configuration. + - All peers are in `Established` state. + - MD5 authentication is enabled for all peers. + * Failure: If any of the following occur: + - A specified peer is not found in the BGP configuration. + - A peer's session state is not `Established`. + - MD5 authentication is not enabled for a peer. Examples -------- @@ -767,7 +709,6 @@ class VerifyBGPPeerMD5Auth(AntaTest): ``` """ - description = "Verifies the MD5 authentication and state of a BGP peer." categories: ClassVar[list[str]] = ["bgp"] commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show bgp neighbors vrf all", revision=3)] @@ -776,56 +717,49 @@ class Input(AntaTest.Input): bgp_peers: list[BgpPeer] """List of IPv4 BGP peers.""" - - class BgpPeer(BaseModel): - """Model for a BGP peer.""" - - peer_address: IPv4Address - """IPv4 address of BGP peer.""" - vrf: str = "default" - """Optional VRF for BGP peer. If not provided, it defaults to `default`.""" + BgpPeer: ClassVar[type[BgpPeer]] = BgpPeer @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyBGPPeerMD5Auth.""" - failures: dict[str, Any] = {"bgp_peers": {}} - - # Iterate over each command - for bgp_peer in self.inputs.bgp_peers: - peer = str(bgp_peer.peer_address) - vrf = bgp_peer.vrf - failure: dict[str, dict[str, dict[str, Any]]] = {"bgp_peers": {peer: {vrf: {}}}} - - # Check if BGP output exists - 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 - ): - failure["bgp_peers"][peer][vrf] = {"status": "Not configured"} - failures = deep_update(failures, failure) - continue + self.result.is_success() - # Check if BGP peer state and authentication - state = bgp_output.get("state") - md5_auth_enabled = bgp_output.get("md5AuthEnabled") - if state != "Established" or not md5_auth_enabled: - failure["bgp_peers"][peer][vrf] = {"state": state, "md5_auth_enabled": md5_auth_enabled} - failures = deep_update(failures, failure) + output = self.instance_commands[0].json_output + + for peer in self.inputs.bgp_peers: + peer_ip = str(peer.peer_address) + peer_list = get_value(output, f"vrfs.{peer.vrf}.peerList", default=[]) + + # Check if the peer is found + if (peer_data := get_item(peer_list, "peerAddress", peer_ip)) is None: + self.result.is_failure(f"{peer} - Not found") + continue - # Check if there are any failures - if not failures["bgp_peers"]: - self.result.is_success() - else: - self.result.is_failure(f"Following BGP peers are not configured, not established or MD5 authentication is not enabled:\n{failures}") + # Check BGP peer state and MD5 authentication + state = peer_data.get("state") + md5_auth_enabled = peer_data.get("md5AuthEnabled") + if state != "Established": + self.result.is_failure(f"{peer} - Session state is not established - State: {state}") + if not md5_auth_enabled: + self.result.is_failure(f"{peer} - Session does not have MD5 authentication enabled") class VerifyEVPNType2Route(AntaTest): """Verifies the EVPN Type-2 routes for a given IPv4 or MAC address and VNI. + This test performs the following checks for each specified VXLAN endpoint: + + 1. Verifies that the endpoint exists in the BGP EVPN table. + 2. Confirms that at least one EVPN Type-2 route with a valid and active path exists. + Expected Results ---------------- - * Success: If all provided VXLAN endpoints have at least one valid and active path to their EVPN Type-2 routes. - * Failure: If any of the provided VXLAN endpoints do not have at least one valid and active path to their EVPN Type-2 routes. + * Success: If all of the following conditions are met: + - All specified VXLAN endpoints are found in the BGP EVPN table. + - Each endpoint has at least one EVPN Type-2 route with a valid and active path. + * Failure: If any of the following occur: + - A VXLAN endpoint is not found in the BGP EVPN table. + - No EVPN Type-2 route with a valid and active path exists for an endpoint. Examples -------- @@ -849,14 +783,7 @@ class Input(AntaTest.Input): vxlan_endpoints: list[VxlanEndpoint] """List of VXLAN endpoints to verify.""" - - class VxlanEndpoint(BaseModel): - """Model for a VXLAN endpoint.""" - - address: IPv4Address | MacAddress - """IPv4 or MAC address of the VXLAN endpoint.""" - vni: Vni - """VNI of the VXLAN endpoint.""" + VxlanEndpoint: ClassVar[type[VxlanEndpoint]] = VxlanEndpoint def render(self, template: AntaTemplate) -> list[AntaCommand]: """Render the template for each VXLAN endpoint in the input list.""" @@ -866,19 +793,15 @@ def render(self, template: AntaTemplate) -> list[AntaCommand]: def test(self) -> None: """Main test function for VerifyEVPNType2Route.""" self.result.is_success() - no_evpn_routes = [] - bad_evpn_routes = [] - for command in self.instance_commands: - address = command.params.address - vni = command.params.vni + for command, endpoint in zip(self.instance_commands, self.inputs.vxlan_endpoints): # Verify that the VXLAN endpoint is in the BGP EVPN table evpn_routes = command.json_output["evpnRoutes"] if not evpn_routes: - no_evpn_routes.append((address, vni)) + self.result.is_failure(f"{endpoint} - No EVPN Type-2 route") continue - # Verify that at least one EVPN route has at least one active/valid path across all learned routes from all RDs combined + # Verify that at least one EVPN Type-2 route has at least one active and valid path across all learned routes from all RDs combined has_active_path = False for route_data in evpn_routes.values(): for path in route_data.get("evpnRoutePaths", []): @@ -887,21 +810,29 @@ def test(self) -> None: has_active_path = True break if not has_active_path: - bad_evpn_routes.extend(list(evpn_routes)) - - if no_evpn_routes: - self.result.is_failure(f"The following VXLAN endpoint do not have any EVPN Type-2 route: {no_evpn_routes}") - if bad_evpn_routes: - self.result.is_failure(f"The following EVPN Type-2 routes do not have at least one valid and active path: {bad_evpn_routes}") + self.result.is_failure(f"{endpoint} - No valid and active path") class VerifyBGPAdvCommunities(AntaTest): - """Verifies if the advertised communities of BGP peers are standard, extended, and large in the specified VRF. + """Verifies that advertised communities are standard, extended and large for BGP peers. + + This test performs the following checks for each specified peer: + + 1. Confirms that the specified VRF is configured. + 2. Verifies that the peer exists in the BGP configuration. + 3. Validates that all required community types are advertised: + - Standard communities + - Extended communities + - Large communities Expected Results ---------------- - * Success: The test will pass if the advertised communities of BGP peers are standard, extended, and large in the specified VRF. - * Failure: The test will fail if the advertised communities of BGP peers are not standard, extended, and large in the specified VRF. + * Success: If all of the following conditions are met: + - All specified peers are found in the BGP configuration. + - Each peer advertises standard, extended and large communities. + * Failure: If any of the following occur: + - A specified peer is not found in the BGP configuration. + - A peer does not advertise standard, extended or large communities. Examples -------- @@ -917,7 +848,6 @@ class VerifyBGPAdvCommunities(AntaTest): ``` """ - description = "Verifies the advertised communities of a BGP peer." categories: ClassVar[list[str]] = ["bgp"] commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show bgp neighbors vrf all", revision=3)] @@ -926,54 +856,46 @@ class Input(AntaTest.Input): 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`.""" + BgpPeer: ClassVar[type[BgpPeer]] = BgpPeer @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyBGPAdvCommunities.""" - failures: dict[str, Any] = {"bgp_peers": {}} - - # Iterate over each bgp peer - for bgp_peer in self.inputs.bgp_peers: - peer = str(bgp_peer.peer_address) - vrf = bgp_peer.vrf - failure: dict[str, dict[str, dict[str, Any]]] = {"bgp_peers": {peer: {vrf: {}}}} - - # Verify BGP peer - 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 - ): - failure["bgp_peers"][peer][vrf] = {"status": "Not configured"} - failures = deep_update(failures, failure) - continue + self.result.is_success() + + output = self.instance_commands[0].json_output + + for peer in self.inputs.bgp_peers: + peer_ip = str(peer.peer_address) + peer_list = get_value(output, f"vrfs.{peer.vrf}.peerList", default=[]) - # Verify BGP peer's advertised communities - bgp_output = bgp_output.get("advertisedCommunities") - if not bgp_output["standard"] or not bgp_output["extended"] or not bgp_output["large"]: - failure["bgp_peers"][peer][vrf] = {"advertised_communities": bgp_output} - failures = deep_update(failures, failure) + # Check if the peer is found + if (peer_data := get_item(peer_list, "peerAddress", peer_ip)) is None: + self.result.is_failure(f"{peer} - Not found") + continue - if not failures["bgp_peers"]: - self.result.is_success() - else: - self.result.is_failure(f"Following BGP peers are not configured or advertised communities are not standard, extended, and large:\n{failures}") + # Check BGP peer advertised communities + if not all(get_value(peer_data, f"advertisedCommunities.{community}") is True for community in ["standard", "extended", "large"]): + self.result.is_failure(f"{peer} - {format_data(peer_data['advertisedCommunities'])}") class VerifyBGPTimers(AntaTest): - """Verifies if the BGP peers are configured with the correct hold and keep-alive timers in the specified VRF. + """Verifies the timers of BGP peers. + + This test performs the following checks for each specified peer: + + 1. Confirms that the specified VRF is configured. + 2. Verifies that the peer exists in the BGP configuration. + 3. Confirms the BGP session hold time/keepalive timers match the expected value. 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. + * Success: If all of the following conditions are met: + - All specified peers are found in the BGP configuration. + - The hold time/keepalive timers match the expected value for each peer. + * Failure: If any of the following occur: + - A specified peer is not found in the BGP configuration. + - The hold time/keepalive timers do not match the expected value for a peer. Examples -------- @@ -993,7 +915,6 @@ class VerifyBGPTimers(AntaTest): ``` """ - description = "Verifies the timers of a BGP peer." categories: ClassVar[list[str]] = ["bgp"] commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show bgp neighbors vrf all", revision=3)] @@ -1002,59 +923,62 @@ class Input(AntaTest.Input): bgp_peers: list[BgpPeer] """List of BGP peers""" + BgpPeer: ClassVar[type[BgpPeer]] = BgpPeer - 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`.""" - hold_time: int = Field(ge=3, le=7200) - """BGP hold time in seconds.""" - keep_alive_time: int = Field(ge=0, le=3600) - """BGP keep-alive time in seconds.""" + @field_validator("bgp_peers") + @classmethod + def validate_bgp_peers(cls, bgp_peers: list[T]) -> list[T]: + """Validate that 'hold_time' or 'keep_alive_time' field is provided in each address family.""" + for peer in bgp_peers: + if peer.hold_time is None or peer.keep_alive_time is None: + msg = f"{peer} 'hold_time' or 'keep_alive_time' field missing in the input" + raise ValueError(msg) + return bgp_peers @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyBGPTimers.""" - failures: dict[str, Any] = {} - - # Iterate over each bgp peer - for bgp_peer in self.inputs.bgp_peers: - peer_address = str(bgp_peer.peer_address) - 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")) - or (bgp_output := get_item(bgp_output, "peerAddress", peer_address)) is None - ): - failures[peer_address] = {vrf: "Not configured"} - continue + self.result.is_success() - # Verify BGP peer's hold and keep alive timers - 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")}} + output = self.instance_commands[0].json_output - 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}") + for peer in self.inputs.bgp_peers: + peer_ip = str(peer.peer_address) + peer_list = get_value(output, f"vrfs.{peer.vrf}.peerList", default=[]) + + # Check if the peer is found + if (peer_data := get_item(peer_list, "peerAddress", peer_ip)) is None: + self.result.is_failure(f"{peer} - Not found") + continue + + # Check BGP peer timers + if peer_data["holdTime"] != peer.hold_time: + self.result.is_failure(f"{peer} - Hold time mismatch - Expected: {peer.hold_time}, Actual: {peer_data['holdTime']}") + if peer_data["keepaliveTime"] != peer.keep_alive_time: + self.result.is_failure(f"{peer} - Keepalive time mismatch - Expected: {peer.keep_alive_time}, Actual: {peer_data['keepaliveTime']}") 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. + This test performs the following checks for each specified peer: + + 1. Confirms that the specified VRF is configured. + 2. Verifies that the peer exists in the BGP configuration. + 3. Validates the BGP drop statistics: + - If specific drop statistics are provided, checks only those counters. + - If no specific drop statistics are provided, checks all available counters. + - Confirms that all checked counters have a value of zero. 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. + * Success: If all of the following conditions are met: + - All specified peers are found in the BGP configuration. + - All specified drop statistics counters (or all counters if none specified) are zero. + * Failure: If any of the following occur: + - A specified peer is not found in the BGP configuration. + - Any checked drop statistics counter has a non-zero value. + - A specified drop statistics counter does not exist. Examples -------- @@ -1072,77 +996,68 @@ class VerifyBGPPeerDropStats(AntaTest): """ categories: ClassVar[list[str]] = ["bgp"] - commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaTemplate(template="show bgp neighbors {peer} vrf {vrf}", revision=3)] + commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show bgp neighbors vrf all", 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] + BgpPeer: ClassVar[type[BgpPeer]] = BgpPeer @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyBGPPeerDropStats.""" - failures: dict[Any, Any] = {} + self.result.is_success() + + output = self.instance_commands[0].json_output - 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 + for peer in self.inputs.bgp_peers: + peer_ip = str(peer.peer_address) + drop_stats_input = peer.drop_stats + peer_list = get_value(output, f"vrfs.{peer.vrf}.peerList", default=[]) - # 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"} + # Check if the peer is found + if (peer_data := get_item(peer_list, "peerAddress", peer_ip)) is None: + self.result.is_failure(f"{peer} - Not found") continue - # Verify BGP peer's drop stats - drop_stats_output = peer_detail.get("dropStats", {}) + # Verify BGP peers' drop stats + drop_stats_output = peer_data["dropStats"] # In case drop stats not provided, It will check all drop statistics - if not drop_statistics: - drop_statistics = drop_stats_output + if not drop_stats_input: + drop_stats_input = 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}") + for drop_stat in drop_stats_input: + if (stat_value := drop_stats_output.get(drop_stat, 0)) != 0: + self.result.is_failure(f"{peer} - Non-zero NLRI drop statistics counter - {drop_stat}: {stat_value}") class VerifyBGPPeerUpdateErrors(AntaTest): """Verifies BGP update error counters for the provided BGP IPv4 peer(s). - By default, all update error counters will be checked for any non-zero values. - An optional list of specific update error counters can be provided for granular testing. + This test performs the following checks for each specified peer: + + 1. Confirms that the specified VRF is configured. + 2. Verifies that the peer exists in the BGP configuration. + 3. Validates the BGP update error counters: + - If specific update error counters are provided, checks only those counters. + - If no update error counters are provided, checks all available counters. + - Confirms that all checked counters have a value of zero. Note: For "disabledAfiSafi" error counter field, checking that it's not "None" versus 0. Expected Results ---------------- - * Success: The test will pass if the BGP peer's update error counter(s) are zero/None. - * Failure: The test will fail if the BGP peer's update error counter(s) are non-zero/not None/Not Found or - peer is not configured. + * Success: If all of the following conditions are met: + - All specified peers are found in the BGP configuration. + - All specified update error counters (or all counters if none specified) are zero. + * Failure: If any of the following occur: + - A specified peer is not found in the BGP configuration. + - Any checked update error counters has a non-zero value. + - A specified update error counters does not exist. Examples -------- @@ -1153,79 +1068,68 @@ class VerifyBGPPeerUpdateErrors(AntaTest): bgp_peers: - peer_address: 172.30.11.1 vrf: default - update_error_filter: + update_errors: - inUpdErrWithdraw ``` """ categories: ClassVar[list[str]] = ["bgp"] - commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaTemplate(template="show bgp neighbors {peer} vrf {vrf}", revision=3)] + commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show bgp neighbors vrf all", revision=3)] class Input(AntaTest.Input): """Input model for the VerifyBGPPeerUpdateErrors 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`.""" - update_errors: list[BgpUpdateError] | None = None - """Optional list of update error counters to be verified. If not provided, test will verifies all the update error counters.""" - - 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] + BgpPeer: ClassVar[type[BgpPeer]] = BgpPeer @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyBGPPeerUpdateErrors.""" - failures: dict[Any, Any] = {} + self.result.is_success() + + output = self.instance_commands[0].json_output - for command, input_entry in zip(self.instance_commands, self.inputs.bgp_peers): - peer = command.params.peer - vrf = command.params.vrf - update_error_counters = input_entry.update_errors + for peer in self.inputs.bgp_peers: + peer_ip = str(peer.peer_address) + update_errors_input = peer.update_errors + peer_list = get_value(output, f"vrfs.{peer.vrf}.peerList", default=[]) - # 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"} + # Check if the peer is found + if (peer_data := get_item(peer_list, "peerAddress", peer_ip)) is None: + self.result.is_failure(f"{peer} - Not found") continue # Getting the BGP peer's error counters output. - error_counters_output = peer_detail.get("peerInUpdateErrors", {}) + error_counters_output = peer_data.get("peerInUpdateErrors", {}) # In case update error counters not provided, It will check all the update error counters. - if not update_error_counters: - update_error_counters = error_counters_output - - # verifying the error counters. - error_counters_not_ok = { - ("disabledAfiSafi" if error_counter == "disabledAfiSafi" else error_counter): value - for error_counter in update_error_counters - if (value := error_counters_output.get(error_counter, "Not Found")) != "None" and value != 0 - } - if error_counters_not_ok: - failures[peer] = {vrf: error_counters_not_ok} + if not update_errors_input: + update_errors_input = error_counters_output - # 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 update error counters:\n{failures}") + # Verify BGP peer's update error counters + for error_counter in update_errors_input: + if (stat_value := error_counters_output.get(error_counter, "Not Found")) != 0 and stat_value != "None": + self.result.is_failure(f"{peer} - Non-zero update error counter - {error_counter}: {stat_value}") class VerifyBgpRouteMaps(AntaTest): """Verifies BGP inbound and outbound route-maps of BGP IPv4 peer(s). + This test performs the following checks for each specified peer: + + 1. Confirms that the specified VRF is configured. + 2. Verifies that the peer exists in the BGP configuration. + 3. Validates the correct BGP route maps are applied in the correct direction (inbound or outbound). + Expected Results ---------------- - * Success: The test will pass if the correct route maps are applied in the correct direction (inbound or outbound) for IPv4 BGP peers in the specified VRF. - * Failure: The test will fail if BGP peers are not configured or any neighbor has an incorrect or missing route map in either the inbound or outbound direction. + * Success: If all of the following conditions are met: + - All specified peers are found in the BGP configuration. + - All specified peers has correct BGP route maps are applied in the correct direction (inbound or outbound). + * Failure: If any of the following occur: + - A specified peer is not found in the BGP configuration. + - A incorrect or missing route map in either the inbound or outbound direction. Examples -------- @@ -1242,86 +1146,72 @@ class VerifyBgpRouteMaps(AntaTest): """ categories: ClassVar[list[str]] = ["bgp"] - commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaTemplate(template="show bgp neighbors {peer} vrf {vrf}", revision=3)] + commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show bgp neighbors vrf all", revision=3)] class Input(AntaTest.Input): """Input model for the VerifyBgpRouteMaps test.""" bgp_peers: list[BgpPeer] """List of BGP peers""" + BgpPeer: ClassVar[type[BgpPeer]] = BgpPeer - 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`.""" - inbound_route_map: str | None = None - """Inbound route map applied, defaults to None.""" - outbound_route_map: str | None = None - """Outbound route map applied, defaults to None.""" - - @model_validator(mode="after") - def validate_inputs(self) -> Self: - """Validate the inputs provided to the BgpPeer class. - - At least one of 'inbound' or 'outbound' route-map must be provided. - """ - if not (self.inbound_route_map or self.outbound_route_map): - msg = "At least one of 'inbound_route_map' or 'outbound_route_map' must be provided." + @field_validator("bgp_peers") + @classmethod + def validate_bgp_peers(cls, bgp_peers: list[T]) -> list[T]: + """Validate that 'peers' field is provided in each address family. + + At least one of 'inbound' or 'outbound' route-map must be provided. + """ + for peer in bgp_peers: + if not (peer.inbound_route_map or peer.outbound_route_map): + msg = f"{peer}; At least one of 'inbound_route_map' or 'outbound_route_map' must be provided." raise ValueError(msg) - return self - - 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] + return bgp_peers @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyBgpRouteMaps.""" - failures: dict[Any, Any] = {} - - for command, input_entry in zip(self.instance_commands, self.inputs.bgp_peers): - peer = str(input_entry.peer_address) - vrf = input_entry.vrf - inbound_route_map = input_entry.inbound_route_map - outbound_route_map = input_entry.outbound_route_map - failure: dict[Any, Any] = {vrf: {}} - - # 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"} + self.result.is_success() + + output = self.instance_commands[0].json_output + + for peer in self.inputs.bgp_peers: + peer_ip = str(peer.peer_address) + inbound_route_map = peer.inbound_route_map + outbound_route_map = peer.outbound_route_map + peer_list = get_value(output, f"vrfs.{peer.vrf}.peerList", default=[]) + + # Check if the peer is found + if (peer_data := get_item(peer_list, "peerAddress", peer_ip)) is None: + self.result.is_failure(f"{peer} - Not found") continue # Verify Inbound route-map - if inbound_route_map and (inbound_map := peer_detail.get("routeMapInbound", "Not Configured")) != inbound_route_map: - failure[vrf].update({"Inbound route-map": inbound_map}) + if inbound_route_map and (inbound_map := peer_data.get("routeMapInbound", "Not Configured")) != inbound_route_map: + self.result.is_failure(f"{peer} - Inbound route-map mismatch - Expected: {inbound_route_map}, Actual: {inbound_map}") # Verify Outbound route-map - if outbound_route_map and (outbound_map := peer_detail.get("routeMapOutbound", "Not Configured")) != outbound_route_map: - failure[vrf].update({"Outbound route-map": outbound_map}) + if outbound_route_map and (outbound_map := peer_data.get("routeMapOutbound", "Not Configured")) != outbound_route_map: + self.result.is_failure(f"{peer} - Outbound route-map mismatch - Expected: {outbound_route_map}, Actual: {outbound_map}") - if failure[vrf]: - failures[peer] = failure - # 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 has an incorrect or missing route map in either the inbound or outbound direction:\n{failures}" - ) +class VerifyBGPPeerRouteLimit(AntaTest): + """Verifies maximum routes and outbound route-maps of BGP IPv4 peer(s). + This test performs the following checks for each specified peer: -class VerifyBGPPeerRouteLimit(AntaTest): - """Verifies the maximum routes and optionally verifies the maximum routes warning limit for the provided BGP IPv4 peer(s). + 1. Confirms that the specified VRF is configured. + 2. Verifies that the peer exists in the BGP configuration. + 3. Confirms the Maximum routes and maximum routes warning limit, if provided match the expected value. Expected Results ---------------- - * Success: The test will pass if the BGP peer's maximum routes and, if provided, the maximum routes warning limit are equal to the given limits. - * Failure: The test will fail if the BGP peer's maximum routes do not match the given limit, or if the maximum routes warning limit is provided - and does not match the given limit, or if the peer is not configured. + * Success: If all of the following conditions are met: + - All specified peers are found in the BGP configuration. + - The maximum routese/maximum routes warning limit match the expected value for a peer. + * Failure: If any of the following occur: + - A specified peer is not found in the BGP configuration. + - The maximum routese/maximum routes warning limit do not match the expected value for a peer. Examples -------- @@ -1338,61 +1228,47 @@ class VerifyBGPPeerRouteLimit(AntaTest): """ categories: ClassVar[list[str]] = ["bgp"] - commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaTemplate(template="show bgp neighbors {peer} vrf {vrf}", revision=3)] + commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show bgp neighbors vrf all", revision=3)] class Input(AntaTest.Input): """Input model for the VerifyBGPPeerRouteLimit test.""" bgp_peers: list[BgpPeer] """List of BGP peers""" + BgpPeer: ClassVar[type[BgpPeer]] = BgpPeer - 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`.""" - maximum_routes: int = Field(ge=0, le=4294967294) - """The maximum allowable number of BGP routes, `0` means unlimited.""" - warning_limit: int = Field(default=0, ge=0, le=4294967294) - """Optional maximum routes warning limit. If not provided, it defaults to `0` meaning no warning limit.""" - - 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] + @field_validator("bgp_peers") + @classmethod + def validate_bgp_peers(cls, bgp_peers: list[T]) -> list[T]: + """Validate that 'peers' field is provided in each address family.""" + for peer in bgp_peers: + if peer.maximum_routes is None: + msg = f"{peer}; 'maximum_routes' field missing in the input" + raise ValueError(msg) + return bgp_peers @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyBGPPeerRouteLimit.""" - failures: dict[Any, Any] = {} - - for command, input_entry in zip(self.instance_commands, self.inputs.bgp_peers): - peer = str(input_entry.peer_address) - vrf = input_entry.vrf - maximum_routes = input_entry.maximum_routes - warning_limit = input_entry.warning_limit - failure: dict[Any, Any] = {} - - # 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"} + self.result.is_success() + + output = self.instance_commands[0].json_output + + for peer in self.inputs.bgp_peers: + peer_ip = str(peer.peer_address) + maximum_routes = peer.maximum_routes + warning_limit = peer.warning_limit + peer_list = get_value(output, f"vrfs.{peer.vrf}.peerList", default=[]) + + # Check if the peer is found + if (peer_data := get_item(peer_list, "peerAddress", peer_ip)) is None: + self.result.is_failure(f"{peer} - Not found") continue # Verify maximum routes configured. - if (actual_routes := peer_detail.get("maxTotalRoutes", "Not Found")) != maximum_routes: - failure["Maximum total routes"] = actual_routes + if (actual_routes := peer_data.get("maxTotalRoutes", "Not Found")) != maximum_routes: + self.result.is_failure(f"{peer} - Maximum routes mismatch - Expected: {maximum_routes}, Actual: {actual_routes}") # Verify warning limit if given. - if warning_limit and (actual_warning_limit := peer_detail.get("totalRoutesWarnLimit", "Not Found")) != warning_limit: - failure["Warning limit"] = actual_warning_limit - - # Updated failures if any. - if failure: - failures[peer] = {vrf: failure} - - # Check if any failures - if not failures: - self.result.is_success() - else: - self.result.is_failure(f"The following BGP peer(s) are not configured or maximum routes and maximum routes warning limit is not correct:\n{failures}") + if warning_limit and (actual_warning_limit := peer_data.get("totalRoutesWarnLimit", "Not Found")) != warning_limit: + self.result.is_failure(f"{peer} - Maximum route warning limit mismatch - Expected: {warning_limit}, Actual: {actual_warning_limit}") diff --git a/examples/tests.yaml b/examples/tests.yaml index 9f1aea993..e22acf49a 100644 --- a/examples/tests.yaml +++ b/examples/tests.yaml @@ -366,7 +366,7 @@ anta.tests.ptp: # Verifies the PTP interfaces state. anta.tests.routing.bgp: - VerifyBGPAdvCommunities: - # Verifies the advertised communities of a BGP peer. + # Verifies that advertised communities are standard, extended and large for BGP peers. bgp_peers: - peer_address: 172.30.11.17 vrf: default @@ -389,7 +389,7 @@ anta.tests.routing.bgp: received_routes: - 192.0.254.3/32 - VerifyBGPPeerASNCap: - # Verifies the four octet asn capabilities of a BGP peer. + # Verifies the four octet ASN capability of BGP peers. bgp_peers: - peer_address: 172.30.11.1 vrf: default @@ -419,14 +419,14 @@ anta.tests.routing.bgp: - inDropAsloop - prefixEvpnDroppedUnsupportedRouteType - VerifyBGPPeerMD5Auth: - # Verifies the MD5 authentication and state of a BGP peer. + # Verifies the MD5 authentication and state of IPv4 BGP peers in a specified VRF. bgp_peers: - peer_address: 172.30.11.1 vrf: default - peer_address: 172.30.11.5 vrf: default - VerifyBGPPeerMPCaps: - # Verifies the multiprotocol capabilities of a BGP peer. + # Verifies the multiprotocol capabilities of BGP peers. bgp_peers: - peer_address: 172.30.11.1 vrf: default @@ -434,14 +434,14 @@ anta.tests.routing.bgp: capabilities: - ipv4Unicast - VerifyBGPPeerRouteLimit: - # Verifies the maximum routes and optionally verifies the maximum routes warning limit for the provided BGP IPv4 peer(s). + # Verifies maximum routes and outbound route-maps of BGP IPv4 peer(s). bgp_peers: - peer_address: 172.30.11.1 vrf: default maximum_routes: 12000 warning_limit: 10000 - VerifyBGPPeerRouteRefreshCap: - # Verifies the route refresh capabilities of a BGP peer. + # Verifies the route refresh capabilities of a BGP peer in a specified VRF. bgp_peers: - peer_address: 172.30.11.1 vrf: default @@ -450,7 +450,7 @@ anta.tests.routing.bgp: bgp_peers: - peer_address: 172.30.11.1 vrf: default - update_error_filter: + update_errors: - inUpdErrWithdraw - VerifyBGPPeersHealth: # Verifies the health of BGP peers for given address families. @@ -478,7 +478,7 @@ anta.tests.routing.bgp: - 10.1.255.2 - 10.1.255.4 - VerifyBGPTimers: - # Verifies the timers of a BGP peer. + # Verifies the timers of BGP peers. bgp_peers: - peer_address: 172.30.11.1 vrf: default diff --git a/tests/units/anta_tests/routing/test_bgp.py b/tests/units/anta_tests/routing/test_bgp.py index 2ae5e8e1b..59a67191c 100644 --- a/tests/units/anta_tests/routing/test_bgp.py +++ b/tests/units/anta_tests/routing/test_bgp.py @@ -1168,234 +1168,10 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "expected": { "result": "failure", "messages": [ - "Following BGP peers are not found or routes are not exchanged properly:\n" - "{'bgp_peers': {'172.30.11.11': {'default': 'Not configured'}, '172.30.11.12': {'default': 'Not configured'}}}" - ], - }, - }, - { - "name": "failure-no-peer", - "test": VerifyBGPExchangedRoutes, - "eos_data": [ - {"vrfs": {}}, - { - "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": {}}, - { - "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, - }, - } - ], - }, - }, - } - } - }, - ], - "inputs": { - "bgp_peers": [ - { - "peer_address": "172.30.11.11", - "vrf": "MGMT", - "advertised_routes": ["192.0.254.3/32"], - "received_routes": ["192.0.255.3/32"], - }, - { - "peer_address": "172.30.11.5", - "vrf": "default", - "advertised_routes": ["192.0.254.3/32", "192.0.254.5/32"], - "received_routes": ["192.0.254.3/32", "192.0.255.4/32"], - }, - ] - }, - "expected": { - "result": "failure", - "messages": ["Following BGP peers are not found or routes are not exchanged properly:\n{'bgp_peers': {'172.30.11.11': {'MGMT': 'Not configured'}}}"], - }, - }, - { - "name": "failure-missing-routes", - "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": 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": True, - }, - } - ], - }, - "192.0.255.4/32": { - "bgpRoutePaths": [ - { - "routeType": { - "valid": True, - "active": True, - }, - } - ], - }, - }, - } - } - }, - ], - "inputs": { - "bgp_peers": [ - { - "peer_address": "172.30.11.1", - "vrf": "default", - "advertised_routes": ["192.0.254.3/32", "192.0.254.51/32"], - "received_routes": ["192.0.254.31/32", "192.0.255.4/32"], - }, - { - "peer_address": "172.30.11.5", - "vrf": "default", - "advertised_routes": ["192.0.254.31/32", "192.0.254.5/32"], - "received_routes": ["192.0.254.3/32", "192.0.255.41/32"], - }, - ] - }, - "expected": { - "result": "failure", - "messages": [ - "Following BGP peers are not found or routes are not exchanged properly:\n{'bgp_peers': " - "{'172.30.11.1': {'default': {'advertised_routes': {'192.0.254.51/32': 'Not found'}, 'received_routes': {'192.0.254.31/32': 'Not found'}}}, " - "'172.30.11.5': {'default': {'advertised_routes': {'192.0.254.31/32': 'Not found'}, 'received_routes': {'192.0.255.41/32': 'Not found'}}}}}" + "Peer: 172.30.11.11 VRF: default Advertised route: 192.0.254.3/32 - Not found", + "Peer: 172.30.11.11 VRF: default Received route: 192.0.255.3/32 - Not found", + "Peer: 172.30.11.12 VRF: default Advertised route: 192.0.254.31/32 - Not found", + "Peer: 172.30.11.12 VRF: default Received route: 192.0.255.31/32 - Not found", ], }, }, @@ -1535,11 +1311,14 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "expected": { "result": "failure", "messages": [ - "Following BGP peers are not found or routes are not exchanged properly:\n{'bgp_peers': " - "{'172.30.11.1': {'default': {'advertised_routes': {'192.0.254.3/32': {'valid': True, 'active': False}, '192.0.254.51/32': 'Not found'}, " - "'received_routes': {'192.0.254.31/32': 'Not found', '192.0.255.4/32': {'valid': False, 'active': False}}}}, " - "'172.30.11.5': {'default': {'advertised_routes': {'192.0.254.31/32': 'Not found', '192.0.254.5/32': {'valid': True, 'active': False}}, " - "'received_routes': {'192.0.254.3/32': {'valid': False, 'active': True}, '192.0.255.41/32': 'Not found'}}}}}" + "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.1 VRF: default Received route: 192.0.254.31/32 - Not found", + "Peer: 172.30.11.1 VRF: default Received route: 192.0.255.4/32 - Valid: False, Active: False", + "Peer: 172.30.11.5 VRF: default Advertised route: 192.0.254.31/32 - Not found", + "Peer: 172.30.11.5 VRF: default Advertised route: 192.0.254.5/32 - Valid: False, Active: True", + "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", ], }, }, @@ -1651,9 +1430,7 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo }, "expected": { "result": "failure", - "messages": [ - "Following BGP peer multiprotocol capabilities are not found or not ok:\n{'bgp_peers': {'172.30.11.1': {'MGMT': {'status': 'Not configured'}}}}" - ], + "messages": ["Peer: 172.30.11.1 VRF: MGMT - VRF not configured"], }, }, { @@ -1714,8 +1491,8 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "expected": { "result": "failure", "messages": [ - "Following BGP peer multiprotocol capabilities are not found or not ok:\n" - "{'bgp_peers': {'172.30.11.10': {'default': {'status': 'Not configured'}}, '172.30.11.1': {'MGMT': {'status': 'Not configured'}}}}" + "Peer: 172.30.11.10 VRF: default - Not found", + "Peer: 172.30.11.1 VRF: MGMT - Not found", ], }, }, @@ -1755,9 +1532,7 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo }, "expected": { "result": "failure", - "messages": [ - "Following BGP peer multiprotocol capabilities are not found or not ok:\n{'bgp_peers': {'172.30.11.1': {'default': {'l2VpnEvpn': 'not found'}}}}" - ], + "messages": ["Peer: 172.30.11.1 VRF: default - l2VpnEvpn not found"], }, }, { @@ -1850,13 +1625,15 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "expected": { "result": "failure", "messages": [ - "Following BGP peer multiprotocol capabilities are not found or not ok:\n" - "{'bgp_peers': {'172.30.11.1': {'default': {'ipv4Unicast': {'advertised': False, 'received': False, 'enabled': False}, " - "'ipv4MplsVpn': {'advertised': False, 'received': True, 'enabled': False}, 'l2VpnEvpn': 'not found'}}, " - "'172.30.11.10': {'MGMT': {'ipv4Unicast': 'not found', 'ipv4MplsVpn': {'advertised': False, 'received': False, 'enabled': True}, " - "'l2VpnEvpn': {'advertised': True, 'received': False, 'enabled': False}}}, " - "'172.30.11.11': {'MGMT': {'ipv4Unicast': {'advertised': False, 'received': False, 'enabled': False}, " - "'ipv4MplsVpn': {'advertised': False, 'received': False, 'enabled': False}, 'l2VpnEvpn': 'not found'}}}}" + "Peer: 172.30.11.1 VRF: default - ipv4Unicast not negotiated - Advertised: False, Received: False, Enabled: False", + "Peer: 172.30.11.1 VRF: default - ipv4MplsVpn not negotiated - Advertised: False, Received: True, Enabled: False", + "Peer: 172.30.11.1 VRF: default - l2VpnEvpn not found", + "Peer: 172.30.11.10 VRF: MGMT - ipv4Unicast not found", + "Peer: 172.30.11.10 VRF: MGMT - ipv4MplsVpn not negotiated - Advertised: False, Received: False, Enabled: True", + "Peer: 172.30.11.10 VRF: MGMT - l2VpnEvpn not negotiated - Advertised: True, Received: False, Enabled: False", + "Peer: 172.30.11.11 VRF: MGMT - ipv4Unicast not negotiated - Advertised: False, Received: False, Enabled: False", + "Peer: 172.30.11.11 VRF: MGMT - ipv4MplsVpn not negotiated - Advertised: False, Received: False, Enabled: False", + "Peer: 172.30.11.11 VRF: MGMT - l2VpnEvpn not found", ], }, }, @@ -1999,10 +1776,8 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "expected": { "result": "failure", "messages": [ - "Following BGP peer multiprotocol capabilities are not found or not ok:\n{'bgp_peers': {'172.30.11.1': " - "{'default': {'status': 'Expected only `ipv4Unicast` capabilities should be listed but found `ipv4Unicast, ipv4MplsLabels` instead.'}}," - " '172.30.11.10': {'MGMT': {'status': 'Expected only `ipv4MplsVpn, l2VpnEvpn` capabilities should be listed but found `ipv4Unicast, " - "ipv4MplsVpn` instead.'}}}}" + "Peer: 172.30.11.1 VRF: default - Mismatch - Expected: ipv4Unicast Actual: ipv4Unicast, ipv4MplsLabels", + "Peer: 172.30.11.10 VRF: MGMT - Mismatch - Expected: ipv4MplsVpn, l2VpnEvpn Actual: ipv4Unicast, ipv4MplsVpn", ], }, }, @@ -2057,63 +1832,6 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo }, "expected": {"result": "success"}, }, - { - "name": "failure-no-vrf", - "test": VerifyBGPPeerASNCap, - "eos_data": [ - { - "vrfs": { - "default": { - "peerList": [ - { - "peerAddress": "172.30.11.1", - "neighborCapabilities": { - "fourOctetAsnCap": { - "advertised": True, - "received": True, - "enabled": True, - }, - }, - } - ] - } - }, - "MGMT": { - "peerList": [ - { - "peerAddress": "172.30.11.10", - "neighborCapabilities": { - "fourOctetAsnCap": { - "advertised": True, - "received": True, - "enabled": True, - }, - }, - } - ] - }, - } - ], - "inputs": { - "bgp_peers": [ - { - "peer_address": "172.30.11.1", - "vrf": "MGMT", - }, - { - "peer_address": "172.30.11.10", - "vrf": "default", - }, - ] - }, - "expected": { - "result": "failure", - "messages": [ - "Following BGP peer four octet asn capabilities are not found or not ok:\n" - "{'bgp_peers': {'172.30.11.1': {'MGMT': {'status': 'Not configured'}}, '172.30.11.10': {'default': {'status': 'Not configured'}}}}" - ], - }, - }, { "name": "failure-no-peer", "test": VerifyBGPPeerASNCap, @@ -2149,9 +1867,7 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo }, "expected": { "result": "failure", - "messages": [ - "Following BGP peer four octet asn capabilities are not found or not ok:\n{'bgp_peers': {'172.30.11.10': {'default': {'status': 'Not configured'}}}}" - ], + "messages": ["Peer: 172.30.11.10 VRF: default - Not found"], }, }, { @@ -2204,8 +1920,8 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "expected": { "result": "failure", "messages": [ - "Following BGP peer four octet asn capabilities are not found or not ok:\n" - "{'bgp_peers': {'172.30.11.1': {'default': {'fourOctetAsnCap': 'not found'}}, '172.30.11.10': {'MGMT': {'fourOctetAsnCap': 'not found'}}}}" + "Peer: 172.30.11.1 VRF: default - 4-octet ASN capability not found", + "Peer: 172.30.11.10 VRF: MGMT - 4-octet ASN capability not found", ], }, }, @@ -2255,9 +1971,8 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "expected": { "result": "failure", "messages": [ - "Following BGP peer four octet asn capabilities are not found or not ok:\n" - "{'bgp_peers': {'172.30.11.1': {'default': {'fourOctetAsnCap': {'advertised': False, 'received': False, 'enabled': False}}}, " - "'172.30.11.10': {'MGMT': {'fourOctetAsnCap': {'advertised': True, 'received': False, 'enabled': True}}}}}" + "Peer: 172.30.11.1 VRF: default - 4-octet ASN capability not negotiated - Advertised: False, Received: False, Enabled: False", + "Peer: 172.30.11.10 VRF: MGMT - 4-octet ASN capability not negotiated - Advertised: True, Received: False, Enabled: True", ], }, }, @@ -2312,25 +2027,6 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo }, "expected": {"result": "success"}, }, - { - "name": "failure-no-vrf", - "test": VerifyBGPPeerRouteRefreshCap, - "eos_data": [{"vrfs": {}}], - "inputs": { - "bgp_peers": [ - { - "peer_address": "172.30.11.1", - "vrf": "MGMT", - } - ] - }, - "expected": { - "result": "failure", - "messages": [ - "Following BGP peer route refresh capabilities are not found or not ok:\n{'bgp_peers': {'172.30.11.1': {'MGMT': {'status': 'Not configured'}}}}" - ], - }, - }, { "name": "failure-no-peer", "test": VerifyBGPPeerRouteRefreshCap, @@ -2387,8 +2083,8 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "expected": { "result": "failure", "messages": [ - "Following BGP peer route refresh capabilities are not found or not ok:\n" - "{'bgp_peers': {'172.30.11.12': {'default': {'status': 'Not configured'}}, '172.30.11.1': {'CS': {'status': 'Not configured'}}}}" + "Peer: 172.30.11.12 VRF: default - Not found", + "Peer: 172.30.11.1 VRF: CS - Not found", ], }, }, @@ -2442,8 +2138,8 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "expected": { "result": "failure", "messages": [ - "Following BGP peer route refresh capabilities are not found or not ok:\n" - "{'bgp_peers': {'172.30.11.1': {'default': {'routeRefreshCap': 'not found'}}, '172.30.11.11': {'CS': {'routeRefreshCap': 'not found'}}}}" + "Peer: 172.30.11.1 VRF: default - Route refresh capability not found", + "Peer: 172.30.11.11 VRF: CS - Route refresh capability not found", ], }, }, @@ -2493,8 +2189,7 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "expected": { "result": "failure", "messages": [ - "Following BGP peer route refresh capabilities are not found or not ok:\n" - "{'bgp_peers': {'172.30.11.1': {'default': {'routeRefreshCap': {'advertised': False, 'received': False, 'enabled': False}}}}}" + "Peer: 172.30.11.1 VRF: default - Route refresh capability not negotiated - Advertised: False, Received: False, Enabled: False", ], }, }, @@ -2539,40 +2234,6 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo }, "expected": {"result": "success"}, }, - { - "name": "failure-no-vrf", - "test": VerifyBGPPeerMD5Auth, - "eos_data": [ - { - "vrfs": { - "default": { - "peerList": [ - { - "peerAddress": "172.30.11.10", - "state": "Established", - "md5AuthEnabled": True, - } - ] - }, - } - } - ], - "inputs": { - "bgp_peers": [ - { - "peer_address": "172.30.11.1", - "vrf": "MGMT", - } - ] - }, - "expected": { - "result": "failure", - "messages": [ - "Following BGP peers are not configured, not established or MD5 authentication is not enabled:\n" - "{'bgp_peers': {'172.30.11.1': {'MGMT': {'status': 'Not configured'}}}}" - ], - }, - }, { "name": "failure-no-peer", "test": VerifyBGPPeerMD5Auth, @@ -2607,16 +2268,16 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "vrf": "default", }, { - "peer_address": "172.30.11.11", - "vrf": "default", + "peer_address": "172.30.11.12", + "vrf": "CS", }, ] }, "expected": { "result": "failure", "messages": [ - "Following BGP peers are not configured, not established or MD5 authentication is not enabled:\n" - "{'bgp_peers': {'172.30.11.10': {'default': {'status': 'Not configured'}}, '172.30.11.11': {'default': {'status': 'Not configured'}}}}" + "Peer: 172.30.11.10 VRF: default - Not found", + "Peer: 172.30.11.12 VRF: CS - Not found", ], }, }, @@ -2640,7 +2301,7 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo { "peerAddress": "172.30.11.10", "state": "Idle", - "md5AuthEnabled": False, + "md5AuthEnabled": True, } ] }, @@ -2662,9 +2323,8 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "expected": { "result": "failure", "messages": [ - "Following BGP peers are not configured, not established or MD5 authentication is not enabled:\n" - "{'bgp_peers': {'172.30.11.1': {'default': {'state': 'Idle', 'md5_auth_enabled': True}}, " - "'172.30.11.10': {'MGMT': {'state': 'Idle', 'md5_auth_enabled': False}}}}" + "Peer: 172.30.11.1 VRF: default - Session state is not established - State: Idle", + "Peer: 172.30.11.10 VRF: MGMT - Session state is not established - State: Idle", ], }, }, @@ -2714,9 +2374,8 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "expected": { "result": "failure", "messages": [ - "Following BGP peers are not configured, not established or MD5 authentication is not enabled:\n" - "{'bgp_peers': {'172.30.11.1': {'default': {'state': 'Established', 'md5_auth_enabled': None}}, " - "'172.30.11.11': {'MGMT': {'state': 'Established', 'md5_auth_enabled': False}}}}" + "Peer: 172.30.11.1 VRF: default - Session does not have MD5 authentication enabled", + "Peer: 172.30.11.11 VRF: MGMT - Session does not have MD5 authentication enabled", ], }, }, @@ -2876,141 +2535,19 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo { "routeType": { "active": True, - "valid": True, - "ecmp": True, - "ecmpContributor": True, - "ecmpHead": True, - }, - }, - { - "routeType": { - "active": False, - "valid": True, - "ecmp": True, - "ecmpContributor": True, - "ecmpHead": False, - }, - }, - ] - }, - "RD: 10.1.0.6:500 mac-ip 10020 aac1.ab4e.bec2 192.168.20.102": { - "evpnRoutePaths": [ - { - "routeType": { - "active": True, - "valid": True, - }, - }, - ] - }, - }, - }, - ], - "inputs": {"vxlan_endpoints": [{"address": "192.168.20.102", "vni": 10020}]}, - "expected": {"result": "success"}, - }, - { - "name": "success-multiple-routes-multiple-paths-mac", - "test": VerifyEVPNType2Route, - "eos_data": [ - { - "vrf": "default", - "routerId": "10.1.0.3", - "asn": 65120, - "evpnRoutes": { - "RD: 10.1.0.5:500 mac-ip 10020 aac1.ab4e.bec2": { - "evpnRoutePaths": [ - { - "routeType": { - "active": True, - "valid": True, - "ecmp": True, - "ecmpContributor": True, - "ecmpHead": True, - }, - }, - { - "routeType": { - "active": False, - "valid": True, - "ecmp": True, - "ecmpContributor": True, - "ecmpHead": False, - }, - }, - ] - }, - "RD: 10.1.0.6:500 mac-ip 10020 aac1.ab4e.bec2": { - "evpnRoutePaths": [ - { - "routeType": { - "active": True, - "valid": True, - }, - }, - ] - }, - }, - }, - ], - "inputs": {"vxlan_endpoints": [{"address": "aac1.ab4e.bec2", "vni": 10020}]}, - "expected": {"result": "success"}, - }, - { - "name": "failure-no-routes", - "test": VerifyEVPNType2Route, - "eos_data": [{"vrf": "default", "routerId": "10.1.0.3", "asn": 65120, "evpnRoutes": {}}], - "inputs": {"vxlan_endpoints": [{"address": "192.168.20.102", "vni": 10020}]}, - "expected": { - "result": "failure", - "messages": ["The following VXLAN endpoint do not have any EVPN Type-2 route: [('192.168.20.102', 10020)]"], - }, - }, - { - "name": "failure-path-not-active", - "test": VerifyEVPNType2Route, - "eos_data": [ - { - "vrf": "default", - "routerId": "10.1.0.3", - "asn": 65120, - "evpnRoutes": { - "RD: 10.1.0.5:500 mac-ip 10020 aac1.ab4e.bec2 192.168.20.102": { - "evpnRoutePaths": [ - { - "routeType": { - "active": False, - "valid": True, - }, - }, - ] - }, - }, - }, - ], - "inputs": {"vxlan_endpoints": [{"address": "192.168.20.102", "vni": 10020}]}, - "expected": { - "result": "failure", - "messages": [ - "The following EVPN Type-2 routes do not have at least one valid and active path: ['RD: 10.1.0.5:500 mac-ip 10020 aac1.ab4e.bec2 192.168.20.102']" - ], - }, - }, - { - "name": "failure-multiple-routes-not-active", - "test": VerifyEVPNType2Route, - "eos_data": [ - { - "vrf": "default", - "routerId": "10.1.0.3", - "asn": 65120, - "evpnRoutes": { - "RD: 10.1.0.5:500 mac-ip 10020 aac1.ab4e.bec2 192.168.20.102": { - "evpnRoutePaths": [ + "valid": True, + "ecmp": True, + "ecmpContributor": True, + "ecmpHead": True, + }, + }, { "routeType": { "active": False, "valid": True, + "ecmp": True, + "ecmpContributor": True, + "ecmpHead": False, }, }, ] @@ -3019,8 +2556,8 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "evpnRoutePaths": [ { "routeType": { - "active": False, - "valid": False, + "active": True, + "valid": True, }, }, ] @@ -3029,17 +2566,10 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo }, ], "inputs": {"vxlan_endpoints": [{"address": "192.168.20.102", "vni": 10020}]}, - "expected": { - "result": "failure", - "messages": [ - "The following EVPN Type-2 routes do not have at least one valid and active path: " - "['RD: 10.1.0.5:500 mac-ip 10020 aac1.ab4e.bec2 192.168.20.102', " - "'RD: 10.1.0.6:500 mac-ip 10020 aac1.ab4e.bec2 192.168.20.102']" - ], - }, + "expected": {"result": "success"}, }, { - "name": "success-multiple-path-and-have-one-active/valid", + "name": "success-multiple-routes-multiple-paths-mac", "test": VerifyEVPNType2Route, "eos_data": [ { @@ -3047,34 +2577,34 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "routerId": "10.1.0.3", "asn": 65120, "evpnRoutes": { - "RD: 10.1.0.5:500 mac-ip 10020 aac1.ab4e.bec2 192.168.20.102": { + "RD: 10.1.0.5:500 mac-ip 10020 aac1.ab4e.bec2": { "evpnRoutePaths": [ { "routeType": { "active": True, "valid": True, + "ecmp": True, + "ecmpContributor": True, + "ecmpHead": True, }, }, { "routeType": { "active": False, "valid": True, + "ecmp": True, + "ecmpContributor": True, + "ecmpHead": False, }, }, ] }, - "RD: 10.1.0.6:500 mac-ip 10020 aac1.ab4e.bec2 192.168.20.102": { + "RD: 10.1.0.6:500 mac-ip 10020 aac1.ab4e.bec2": { "evpnRoutePaths": [ { "routeType": { - "active": False, - "valid": False, - }, - }, - { - "routeType": { - "active": False, - "valid": False, + "active": True, + "valid": True, }, }, ] @@ -3082,13 +2612,21 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo }, }, ], + "inputs": {"vxlan_endpoints": [{"address": "aac1.ab4e.bec2", "vni": 10020}]}, + "expected": {"result": "success"}, + }, + { + "name": "failure-no-routes", + "test": VerifyEVPNType2Route, + "eos_data": [{"vrf": "default", "routerId": "10.1.0.3", "asn": 65120, "evpnRoutes": {}}], "inputs": {"vxlan_endpoints": [{"address": "192.168.20.102", "vni": 10020}]}, "expected": { - "result": "success", + "result": "failure", + "messages": ["Address: 192.168.20.102 VNI: 10020 - No EVPN Type-2 route"], }, }, { - "name": "failure-multiple-endpoints", + "name": "failure-path-not-active", "test": VerifyEVPNType2Route, "eos_data": [ { @@ -3101,19 +2639,30 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo { "routeType": { "active": False, - "valid": False, + "valid": True, }, }, ] }, }, }, + ], + "inputs": {"vxlan_endpoints": [{"address": "192.168.20.102", "vni": 10020}]}, + "expected": { + "result": "failure", + "messages": ["Address: 192.168.20.102 VNI: 10020 - No valid and active path"], + }, + }, + { + "name": "failure-multiple-endpoints", + "test": VerifyEVPNType2Route, + "eos_data": [ { "vrf": "default", "routerId": "10.1.0.3", "asn": 65120, "evpnRoutes": { - "RD: 10.1.0.5:500 mac-ip 10010 aac1.ab5d.b41e": { + "RD: 10.1.0.5:500 mac-ip 10020 aac1.ab4e.bec2 192.168.20.102": { "evpnRoutePaths": [ { "routeType": { @@ -3125,33 +2674,12 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo }, }, }, - ], - "inputs": { - "vxlan_endpoints": [ - {"address": "192.168.20.102", "vni": 10020}, - {"address": "aac1.ab5d.b41e", "vni": 10010}, - ] - }, - "expected": { - "result": "failure", - "messages": [ - "The following EVPN Type-2 routes do not have at least one valid and active path: " - "['RD: 10.1.0.5:500 mac-ip 10020 aac1.ab4e.bec2 192.168.20.102', " - "'RD: 10.1.0.5:500 mac-ip 10010 aac1.ab5d.b41e']" - ], - }, - }, - { - "name": "failure-multiple-endpoints-one-no-routes", - "test": VerifyEVPNType2Route, - "eos_data": [ - {"vrf": "default", "routerId": "10.1.0.3", "asn": 65120, "evpnRoutes": {}}, { "vrf": "default", "routerId": "10.1.0.3", "asn": 65120, "evpnRoutes": { - "RD: 10.1.0.5:500 mac-ip 10010 aac1.ab5d.b41e 192.168.10.101": { + "RD: 10.1.0.5:500 mac-ip 10010 aac1.ab5d.b41e": { "evpnRoutePaths": [ { "routeType": { @@ -3166,35 +2694,13 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo ], "inputs": { "vxlan_endpoints": [ - {"address": "aac1.ab4e.bec2", "vni": 10020}, - {"address": "192.168.10.101", "vni": 10010}, - ] - }, - "expected": { - "result": "failure", - "messages": [ - "The following VXLAN endpoint do not have any EVPN Type-2 route: [('aa:c1:ab:4e:be:c2', 10020)]", - "The following EVPN Type-2 routes do not have at least one valid and active path: " - "['RD: 10.1.0.5:500 mac-ip 10010 aac1.ab5d.b41e 192.168.10.101']", - ], - }, - }, - { - "name": "failure-multiple-endpoints-no-routes", - "test": VerifyEVPNType2Route, - "eos_data": [ - {"vrf": "default", "routerId": "10.1.0.3", "asn": 65120, "evpnRoutes": {}}, - {"vrf": "default", "routerId": "10.1.0.3", "asn": 65120, "evpnRoutes": {}}, - ], - "inputs": { - "vxlan_endpoints": [ - {"address": "aac1.ab4e.bec2", "vni": 10020}, - {"address": "192.168.10.101", "vni": 10010}, + {"address": "192.168.20.102", "vni": 10020}, + {"address": "aac1.ab5d.b41e", "vni": 10010}, ] }, "expected": { "result": "failure", - "messages": ["The following VXLAN endpoint do not have any EVPN Type-2 route: [('aa:c1:ab:4e:be:c2', 10020), ('192.168.10.101', 10010)]"], + "messages": ["Address: 192.168.20.102 VNI: 10020 - No valid and active path", "Address: aa:c1:ab:5d:b4:1e VNI: 10010 - No valid and active path"], }, }, { @@ -3243,43 +2749,6 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo }, "expected": {"result": "success"}, }, - { - "name": "failure-no-vrf", - "test": VerifyBGPAdvCommunities, - "eos_data": [ - { - "vrfs": { - "default": { - "peerList": [ - { - "peerAddress": "172.30.11.1", - "advertisedCommunities": { - "standard": True, - "extended": True, - "large": True, - }, - } - ] - }, - } - } - ], - "inputs": { - "bgp_peers": [ - { - "peer_address": "172.30.11.17", - "vrf": "MGMT", - } - ] - }, - "expected": { - "result": "failure", - "messages": [ - "Following BGP peers are not configured or advertised communities are not standard, extended, and large:\n" - "{'bgp_peers': {'172.30.11.17': {'MGMT': {'status': 'Not configured'}}}}" - ], - }, - }, { "name": "failure-no-peer", "test": VerifyBGPAdvCommunities, @@ -3328,8 +2797,8 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "expected": { "result": "failure", "messages": [ - "Following BGP peers are not configured or advertised communities are not standard, extended, and large:\n" - "{'bgp_peers': {'172.30.11.10': {'default': {'status': 'Not configured'}}, '172.30.11.12': {'MGMT': {'status': 'Not configured'}}}}" + "Peer: 172.30.11.10 VRF: default - Not found", + "Peer: 172.30.11.12 VRF: MGMT - Not found", ], }, }, @@ -3381,9 +2850,8 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "expected": { "result": "failure", "messages": [ - "Following BGP peers are not configured or advertised communities are not standard, extended, and large:\n" - "{'bgp_peers': {'172.30.11.1': {'default': {'advertised_communities': {'standard': False, 'extended': False, 'large': False}}}, " - "'172.30.11.10': {'CS': {'advertised_communities': {'standard': True, 'extended': True, 'large': False}}}}}" + "Peer: 172.30.11.1 VRF: default - Standard: False, Extended: False, Large: False", + "Peer: 172.30.11.10 VRF: CS - Standard: True, Extended: True, Large: False", ], }, }, @@ -3438,15 +2906,7 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "eos_data": [ { "vrfs": { - "default": { - "peerList": [ - { - "peerAddress": "172.30.11.1", - "holdTime": 180, - "keepaliveTime": 60, - } - ] - }, + "default": {"peerList": []}, "MGMT": {"peerList": []}, } } @@ -3461,7 +2921,7 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo }, { "peer_address": "172.30.11.11", - "vrf": "MGMT", + "vrf": "default", "hold_time": 180, "keep_alive_time": 60, }, @@ -3470,8 +2930,8 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "expected": { "result": "failure", "messages": [ - "Following BGP peers are not configured or hold and keep-alive timers are not correct:\n" - "{'172.30.11.1': {'MGMT': 'Not configured'}, '172.30.11.11': {'MGMT': 'Not configured'}}" + "Peer: 172.30.11.1 VRF: MGMT - Not found", + "Peer: 172.30.11.11 VRF: default - Not found", ], }, }, @@ -3521,9 +2981,9 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "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': 160, 'keep_alive_time': 60}}, " - "'172.30.11.11': {'MGMT': {'hold_time': 120, 'keep_alive_time': 40}}}" + "Peer: 172.30.11.1 VRF: default - Hold time mismatch - Expected: 180, Actual: 160", + "Peer: 172.30.11.11 VRF: MGMT - Hold time mismatch - Expected: 180, Actual: 120", + "Peer: 172.30.11.11 VRF: MGMT - Keepalive time mismatch - Expected: 60, Actual: 40", ], }, }, @@ -3551,10 +3011,6 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo } ] }, - }, - }, - { - "vrfs": { "MGMT": { "peerList": [ { @@ -3591,10 +3047,7 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo { "name": "failure-not-found", "test": VerifyBGPPeerDropStats, - "eos_data": [ - {"vrfs": {}}, - {"vrfs": {}}, - ], + "eos_data": [{"vrfs": {}}], "inputs": { "bgp_peers": [ { @@ -3608,8 +3061,8 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "expected": { "result": "failure", "messages": [ - "The following BGP peers are not configured or have non-zero NLRI drop statistics counters:\n" - "{'10.100.0.8': {'default': 'Not configured'}, '10.100.0.9': {'MGMT': 'Not configured'}}" + "Peer: 10.100.0.8 VRF: default - Not found", + "Peer: 10.100.0.9 VRF: MGMT - Not found", ], }, }, @@ -3637,10 +3090,6 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo } ] }, - }, - }, - { - "vrfs": { "MGMT": { "peerList": [ { @@ -3675,9 +3124,10 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "expected": { "result": "failure", "messages": [ - "The following BGP peers are not configured or have non-zero NLRI drop statistics counters:\n" - "{'10.100.0.8': {'default': {'prefixDroppedMartianV4': 1, 'prefixDroppedMaxRouteLimitViolatedV4': 1}}, " - "'10.100.0.9': {'MGMT': {'inDropOrigId': 1, 'inDropNhLocal': 1}}}" + "Peer: 10.100.0.8 VRF: default - Non-zero NLRI drop statistics counter - prefixDroppedMartianV4: 1", + "Peer: 10.100.0.8 VRF: default - Non-zero NLRI drop statistics counter - prefixDroppedMaxRouteLimitViolatedV4: 1", + "Peer: 10.100.0.9 VRF: MGMT - Non-zero NLRI drop statistics counter - inDropOrigId: 1", + "Peer: 10.100.0.9 VRF: MGMT - Non-zero NLRI drop statistics counter - inDropNhLocal: 1", ], }, }, @@ -3705,10 +3155,6 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo } ] }, - }, - }, - { - "vrfs": { "MGMT": { "peerList": [ { @@ -3762,10 +3208,6 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo } ] }, - }, - }, - { - "vrfs": { "MGMT": { "peerList": [ { @@ -3796,49 +3238,14 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "expected": { "result": "failure", "messages": [ - "The following BGP peers are not configured or have non-zero NLRI drop statistics counters:\n" - "{'10.100.0.8': {'default': {'inDropAsloop': 3, 'inDropOrigId': 1, 'inDropNhLocal': 1, " - "'prefixDroppedMartianV4': 1, 'prefixDroppedMaxRouteLimitViolatedV4': 1}}, " - "'10.100.0.9': {'MGMT': {'inDropAsloop': 2, 'inDropOrigId': 1, 'inDropNhLocal': 1}}}" - ], - }, - }, - { - "name": "failure-drop-stat-not-found", - "test": VerifyBGPPeerDropStats, - "eos_data": [ - { - "vrfs": { - "default": { - "peerList": [ - { - "peerAddress": "10.100.0.8", - "dropStats": { - "inDropAsloop": 3, - "inDropClusterIdLoop": 0, - "inDropMalformedMpbgp": 0, - "inDropOrigId": 1, - "inDropNhLocal": 1, - "inDropNhAfV6": 0, - "prefixDroppedMaxRouteLimitViolatedV4": 1, - "prefixDroppedMartianV6": 0, - }, - } - ] - }, - }, - }, - ], - "inputs": { - "bgp_peers": [ - {"peer_address": "10.100.0.8", "vrf": "default", "drop_stats": ["inDropAsloop", "inDropOrigId", "inDropNhLocal", "prefixDroppedMartianV4"]} - ] - }, - "expected": { - "result": "failure", - "messages": [ - "The following BGP peers are not configured or have non-zero NLRI drop statistics counters:\n" - "{'10.100.0.8': {'default': {'inDropAsloop': 3, 'inDropOrigId': 1, 'inDropNhLocal': 1, 'prefixDroppedMartianV4': 'Not Found'}}}" + "Peer: 10.100.0.8 VRF: default - Non-zero NLRI drop statistics counter - inDropAsloop: 3", + "Peer: 10.100.0.8 VRF: default - Non-zero NLRI drop statistics counter - inDropOrigId: 1", + "Peer: 10.100.0.8 VRF: default - Non-zero NLRI drop statistics counter - inDropNhLocal: 1", + "Peer: 10.100.0.8 VRF: default - Non-zero NLRI drop statistics counter - prefixDroppedMartianV4: 1", + "Peer: 10.100.0.8 VRF: default - Non-zero NLRI drop statistics counter - prefixDroppedMaxRouteLimitViolatedV4: 1", + "Peer: 10.100.0.9 VRF: MGMT - Non-zero NLRI drop statistics counter - inDropAsloop: 2", + "Peer: 10.100.0.9 VRF: MGMT - Non-zero NLRI drop statistics counter - inDropOrigId: 1", + "Peer: 10.100.0.9 VRF: MGMT - Non-zero NLRI drop statistics counter - inDropNhLocal: 1", ], }, }, @@ -3862,10 +3269,6 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo } ] }, - }, - }, - { - "vrfs": { "MGMT": { "peerList": [ { @@ -3896,7 +3299,6 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "test": VerifyBGPPeerUpdateErrors, "eos_data": [ {"vrfs": {}}, - {"vrfs": {}}, ], "inputs": { "bgp_peers": [ @@ -3907,8 +3309,8 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "expected": { "result": "failure", "messages": [ - "The following BGP peers are not configured or have non-zero update error counters:\n" - "{'10.100.0.8': {'default': 'Not configured'}, '10.100.0.9': {'MGMT': 'Not configured'}}" + "Peer: 10.100.0.8 VRF: default - Not found", + "Peer: 10.100.0.9 VRF: MGMT - Not found", ], }, }, @@ -3932,10 +3334,6 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo } ] }, - }, - }, - { - "vrfs": { "MGMT": { "peerList": [ { @@ -3962,9 +3360,8 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "expected": { "result": "failure", "messages": [ - "The following BGP peers are not configured or have non-zero update error counters:\n" - "{'10.100.0.8': {'default': {'disabledAfiSafi': 'ipv4Unicast'}}, " - "'10.100.0.9': {'MGMT': {'inUpdErrWithdraw': 1}}}" + "Peer: 10.100.0.8 VRF: default - Non-zero update error counter - disabledAfiSafi: ipv4Unicast", + "Peer: 10.100.0.9 VRF: MGMT - Non-zero update error counter - inUpdErrWithdraw: 1", ], }, }, @@ -3988,10 +3385,6 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo } ] }, - }, - }, - { - "vrfs": { "MGMT": { "peerList": [ { @@ -4037,10 +3430,6 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo } ] }, - }, - }, - { - "vrfs": { "MGMT": { "peerList": [ { @@ -4071,9 +3460,10 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "expected": { "result": "failure", "messages": [ - "The following BGP peers are not configured or have non-zero update error counters:\n" - "{'10.100.0.8': {'default': {'inUpdErrWithdraw': 1, 'disabledAfiSafi': 'ipv4Unicast'}}, " - "'10.100.0.9': {'MGMT': {'inUpdErrWithdraw': 1, 'inUpdErrDisableAfiSafi': 1}}}" + "Peer: 10.100.0.8 VRF: default - Non-zero update error counter - inUpdErrWithdraw: 1", + "Peer: 10.100.0.8 VRF: default - Non-zero update error counter - disabledAfiSafi: ipv4Unicast", + "Peer: 10.100.0.9 VRF: MGMT - Non-zero update error counter - inUpdErrWithdraw: 1", + "Peer: 10.100.0.9 VRF: MGMT - Non-zero update error counter - inUpdErrDisableAfiSafi: 1", ], }, }, @@ -4096,10 +3486,6 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo } ] }, - }, - }, - { - "vrfs": { "MGMT": { "peerList": [ { @@ -4129,9 +3515,10 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "expected": { "result": "failure", "messages": [ - "The following BGP peers are not configured or have non-zero update error counters:\n" - "{'10.100.0.8': {'default': {'inUpdErrWithdraw': 'Not Found', 'disabledAfiSafi': 'ipv4Unicast'}}, " - "'10.100.0.9': {'MGMT': {'inUpdErrWithdraw': 1, 'inUpdErrDisableAfiSafi': 'Not Found'}}}" + "Peer: 10.100.0.8 VRF: default - Non-zero update error counter - inUpdErrWithdraw: Not Found", + "Peer: 10.100.0.8 VRF: default - Non-zero update error counter - disabledAfiSafi: ipv4Unicast", + "Peer: 10.100.0.9 VRF: MGMT - Non-zero update error counter - inUpdErrWithdraw: 1", + "Peer: 10.100.0.9 VRF: MGMT - Non-zero update error counter - inUpdErrDisableAfiSafi: Not Found", ], }, }, @@ -4150,10 +3537,6 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo } ] }, - }, - }, - { - "vrfs": { "MGMT": { "peerList": [ { @@ -4189,10 +3572,6 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo } ] }, - }, - }, - { - "vrfs": { "MGMT": { "peerList": [ { @@ -4214,9 +3593,10 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "expected": { "result": "failure", "messages": [ - "The following BGP peers are not configured or has an incorrect or missing route map in either the inbound or outbound direction:\n" - "{'10.100.0.8': {'default': {'Inbound route-map': 'RM-MLAG-PEER', 'Outbound route-map': 'RM-MLAG-PEER'}}, " - "'10.100.0.10': {'MGMT': {'Inbound route-map': 'RM-MLAG-PEER', 'Outbound route-map': 'RM-MLAG-PEER'}}}" + "Peer: 10.100.0.8 VRF: default - Inbound route-map mismatch - Expected: RM-MLAG-PEER-IN, Actual: RM-MLAG-PEER", + "Peer: 10.100.0.8 VRF: default - Outbound route-map mismatch - Expected: RM-MLAG-PEER-OUT, Actual: RM-MLAG-PEER", + "Peer: 10.100.0.10 VRF: MGMT - Inbound route-map mismatch - Expected: RM-MLAG-PEER-IN, Actual: RM-MLAG-PEER", + "Peer: 10.100.0.10 VRF: MGMT - Outbound route-map mismatch - Expected: RM-MLAG-PEER-OUT, Actual: RM-MLAG-PEER", ], }, }, @@ -4235,10 +3615,6 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo } ] }, - }, - }, - { - "vrfs": { "MGMT": { "peerList": [ { @@ -4260,8 +3636,8 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "expected": { "result": "failure", "messages": [ - "The following BGP peers are not configured or has an incorrect or missing route map in either the inbound or outbound direction:\n" - "{'10.100.0.8': {'default': {'Inbound route-map': 'RM-MLAG-PEER'}}, '10.100.0.10': {'MGMT': {'Inbound route-map': 'RM-MLAG-PEER'}}}" + "Peer: 10.100.0.8 VRF: default - Inbound route-map mismatch - Expected: RM-MLAG-PEER-IN, Actual: RM-MLAG-PEER", + "Peer: 10.100.0.10 VRF: MGMT - Inbound route-map mismatch - Expected: RM-MLAG-PEER-IN, Actual: RM-MLAG-PEER", ], }, }, @@ -4278,10 +3654,6 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo } ] }, - }, - }, - { - "vrfs": { "MGMT": { "peerList": [ { @@ -4301,9 +3673,10 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "expected": { "result": "failure", "messages": [ - "The following BGP peers are not configured or has an incorrect or missing route map in either the inbound or outbound direction:\n" - "{'10.100.0.8': {'default': {'Inbound route-map': 'Not Configured', 'Outbound route-map': 'Not Configured'}}, " - "'10.100.0.10': {'MGMT': {'Inbound route-map': 'Not Configured', 'Outbound route-map': 'Not Configured'}}}" + "Peer: 10.100.0.8 VRF: default - Inbound route-map mismatch - Expected: RM-MLAG-PEER-IN, Actual: Not Configured", + "Peer: 10.100.0.8 VRF: default - Outbound route-map mismatch - Expected: RM-MLAG-PEER-OUT, Actual: Not Configured", + "Peer: 10.100.0.10 VRF: MGMT - Inbound route-map mismatch - Expected: RM-MLAG-PEER-IN, Actual: Not Configured", + "Peer: 10.100.0.10 VRF: MGMT - Outbound route-map mismatch - Expected: RM-MLAG-PEER-OUT, Actual: Not Configured", ], }, }, @@ -4314,10 +3687,6 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo { "vrfs": { "default": {"peerList": []}, - }, - }, - { - "vrfs": { "MGMT": {"peerList": []}, }, }, @@ -4331,8 +3700,8 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "expected": { "result": "failure", "messages": [ - "The following BGP peers are not configured or has an incorrect or missing route map in either the inbound or outbound direction:\n" - "{'10.100.0.8': {'default': 'Not configured'}, '10.100.0.10': {'MGMT': 'Not configured'}}" + "Peer: 10.100.0.8 VRF: default - Not found", + "Peer: 10.100.0.10 VRF: MGMT - Not found", ], }, }, @@ -4351,10 +3720,6 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo } ] }, - }, - }, - { - "vrfs": { "MGMT": { "peerList": [ { @@ -4382,10 +3747,6 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo { "vrfs": { "default": {}, - }, - }, - { - "vrfs": { "MGMT": {}, }, }, @@ -4399,8 +3760,8 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "expected": { "result": "failure", "messages": [ - "The following BGP peer(s) are not configured or maximum routes and maximum routes warning limit is not correct:\n" - "{'10.100.0.8': {'default': 'Not configured'}, '10.100.0.9': {'MGMT': 'Not configured'}}" + "Peer: 10.100.0.8 VRF: default - Not found", + "Peer: 10.100.0.9 VRF: MGMT - Not found", ], }, }, @@ -4419,10 +3780,6 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo } ] }, - }, - }, - { - "vrfs": { "MGMT": { "peerList": [ { @@ -4444,9 +3801,10 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "expected": { "result": "failure", "messages": [ - "The following BGP peer(s) are not configured or maximum routes and maximum routes warning limit is not correct:\n" - "{'10.100.0.8': {'default': {'Maximum total routes': 13000, 'Warning limit': 11000}}, " - "'10.100.0.9': {'MGMT': {'Maximum total routes': 11000, 'Warning limit': 10000}}}" + "Peer: 10.100.0.8 VRF: default - Maximum routes mismatch - Expected: 12000, Actual: 13000", + "Peer: 10.100.0.8 VRF: default - Maximum route warning limit mismatch - Expected: 10000, Actual: 11000", + "Peer: 10.100.0.9 VRF: MGMT - Maximum routes mismatch - Expected: 10000, Actual: 11000", + "Peer: 10.100.0.9 VRF: MGMT - Maximum route warning limit mismatch - Expected: 9000, Actual: 10000", ], }, }, @@ -4464,10 +3822,6 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo } ] }, - }, - }, - { - "vrfs": { "MGMT": { "peerList": [ { @@ -4487,9 +3841,9 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "expected": { "result": "failure", "messages": [ - "The following BGP peer(s) are not configured or maximum routes and maximum routes warning limit is not correct:\n" - "{'10.100.0.8': {'default': {'Warning limit': 'Not Found'}}, " - "'10.100.0.9': {'MGMT': {'Maximum total routes': 'Not Found', 'Warning limit': 'Not Found'}}}" + "Peer: 10.100.0.8 VRF: default - Maximum route warning limit mismatch - Expected: 10000, Actual: Not Found", + "Peer: 10.100.0.9 VRF: MGMT - Maximum routes mismatch - Expected: 10000, Actual: Not Found", + "Peer: 10.100.0.9 VRF: MGMT - Maximum route warning limit mismatch - Expected: 9000, Actual: Not Found", ], }, }, diff --git a/tests/units/input_models/routing/test_bgp.py b/tests/units/input_models/routing/test_bgp.py index aabc8f293..66c37af61 100644 --- a/tests/units/input_models/routing/test_bgp.py +++ b/tests/units/input_models/routing/test_bgp.py @@ -11,8 +11,16 @@ import pytest from pydantic import ValidationError -from anta.input_models.routing.bgp import BgpAddressFamily -from anta.tests.routing.bgp import VerifyBGPPeerCount, VerifyBGPSpecificPeers +from anta.input_models.routing.bgp import BgpAddressFamily, BgpPeer +from anta.tests.routing.bgp import ( + VerifyBGPExchangedRoutes, + VerifyBGPPeerCount, + VerifyBGPPeerMPCaps, + VerifyBGPPeerRouteLimit, + VerifyBgpRouteMaps, + VerifyBGPSpecificPeers, + VerifyBGPTimers, +) if TYPE_CHECKING: from anta.custom_types import Afi, Safi @@ -96,3 +104,135 @@ def test_invalid(self, address_families: list[BgpAddressFamily]) -> None: """Test VerifyBGPSpecificPeers.Input invalid inputs.""" with pytest.raises(ValidationError): VerifyBGPSpecificPeers.Input(address_families=address_families) + + +class TestVerifyBGPExchangedRoutesInput: + """Test anta.tests.routing.bgp.VerifyBGPExchangedRoutes.Input.""" + + @pytest.mark.parametrize( + ("bgp_peers"), + [ + pytest.param( + [{"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", + ), + ], + ) + def test_valid(self, bgp_peers: list[BgpPeer]) -> None: + """Test VerifyBGPExchangedRoutes.Input valid inputs.""" + VerifyBGPExchangedRoutes.Input(bgp_peers=bgp_peers) + + @pytest.mark.parametrize( + ("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: + """Test VerifyBGPExchangedRoutes.Input invalid inputs.""" + with pytest.raises(ValidationError): + VerifyBGPExchangedRoutes.Input(bgp_peers=bgp_peers) + + +class TestVerifyBGPPeerMPCapsInput: + """Test anta.tests.routing.bgp.VerifyBGPPeerMPCaps.Input.""" + + @pytest.mark.parametrize( + ("bgp_peers"), + [ + pytest.param([{"peer_address": "172.30.255.5", "vrf": "default", "capabilities": ["ipv4Unicast"]}], id="valid"), + ], + ) + def test_valid(self, bgp_peers: list[BgpPeer]) -> None: + """Test VerifyBGPPeerMPCaps.Input valid inputs.""" + VerifyBGPPeerMPCaps.Input(bgp_peers=bgp_peers) + + @pytest.mark.parametrize( + ("bgp_peers"), + [ + pytest.param([{"peer_address": "172.30.255.5", "vrf": "default"}], id="invalid"), + ], + ) + def test_invalid(self, bgp_peers: list[BgpPeer]) -> None: + """Test VerifyBGPPeerMPCaps.Input invalid inputs.""" + with pytest.raises(ValidationError): + VerifyBGPPeerMPCaps.Input(bgp_peers=bgp_peers) + + +class TestVerifyBGPTimersInput: + """Test anta.tests.routing.bgp.VerifyBGPTimers.Input.""" + + @pytest.mark.parametrize( + ("bgp_peers"), + [ + pytest.param([{"peer_address": "172.30.255.5", "vrf": "default", "hold_time": 180, "keep_alive_time": 60}], id="valid"), + ], + ) + def test_valid(self, bgp_peers: list[BgpPeer]) -> None: + """Test VerifyBGPTimers.Input valid inputs.""" + VerifyBGPTimers.Input(bgp_peers=bgp_peers) + + @pytest.mark.parametrize( + ("bgp_peers"), + [ + pytest.param([{"peer_address": "172.30.255.5", "vrf": "default"}], id="invalid"), + pytest.param([{"peer_address": "172.30.255.5", "vrf": "default", "hold_time": 180}], id="invalid_keep_alive"), + pytest.param([{"peer_address": "172.30.255.5", "vrf": "default", "keep_alive_time": 180}], id="invalid_hold_time"), + ], + ) + def test_invalid(self, bgp_peers: list[BgpPeer]) -> None: + """Test VerifyBGPTimers.Input invalid inputs.""" + with pytest.raises(ValidationError): + VerifyBGPTimers.Input(bgp_peers=bgp_peers) + + +class TestVerifyBgpRouteMapsInput: + """Test anta.tests.routing.bgp.VerifyBgpRouteMaps.Input.""" + + @pytest.mark.parametrize( + ("bgp_peers"), + [ + pytest.param([{"peer_address": "172.30.255.5", "vrf": "default", "inbound_route_map": "Test", "outbound_route_map": "Test"}], id="valid"), + ], + ) + def test_valid(self, bgp_peers: list[BgpPeer]) -> None: + """Test VerifyBgpRouteMaps.Input valid inputs.""" + VerifyBgpRouteMaps.Input(bgp_peers=bgp_peers) + + @pytest.mark.parametrize( + ("bgp_peers"), + [ + pytest.param([{"peer_address": "172.30.255.5", "vrf": "default"}], id="invalid"), + ], + ) + def test_invalid(self, bgp_peers: list[BgpPeer]) -> None: + """Test VerifyBgpRouteMaps.Input invalid inputs.""" + with pytest.raises(ValidationError): + VerifyBgpRouteMaps.Input(bgp_peers=bgp_peers) + + +class TestVerifyBGPPeerRouteLimitInput: + """Test anta.tests.routing.bgp.VerifyBGPPeerRouteLimit.Input.""" + + @pytest.mark.parametrize( + ("bgp_peers"), + [ + pytest.param([{"peer_address": "172.30.255.5", "vrf": "default", "maximum_routes": 10000}], id="valid"), + ], + ) + def test_valid(self, bgp_peers: list[BgpPeer]) -> None: + """Test VerifyBGPPeerRouteLimit.Input valid inputs.""" + VerifyBGPPeerRouteLimit.Input(bgp_peers=bgp_peers) + + @pytest.mark.parametrize( + ("bgp_peers"), + [ + pytest.param([{"peer_address": "172.30.255.5", "vrf": "default"}], id="invalid"), + ], + ) + def test_invalid(self, bgp_peers: list[BgpPeer]) -> None: + """Test VerifyBGPPeerRouteLimit.Input invalid inputs.""" + with pytest.raises(ValidationError): + VerifyBGPPeerRouteLimit.Input(bgp_peers=bgp_peers) From 737bbf0a3b3ff396317ab3badd20e89116e25901 Mon Sep 17 00:00:00 2001 From: Carl Baillargeon Date: Tue, 24 Dec 2024 11:07:06 -0500 Subject: [PATCH 3/4] chore(anta): Update RESPX (#982) --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index ac2ca3363..d52d4876e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -74,7 +74,7 @@ dev = [ "pytest-httpx>=0.30.0", "pytest-metadata>=3.0.0", "pytest>=7.4.0", - "respx @ git+https://github.com/ndhansen/respx", # until https://github.com/lundberg/respx/pull/278 is merged + "respx>=0.22.0", "ruff>=0.5.4,<0.9.0", "tox>=4.10.0,<5.0.0", "types-PyYAML", From 2a0bf6c9b561d0f9af451de8c85e51213a8459b6 Mon Sep 17 00:00:00 2001 From: Guillaume Mulocher Date: Tue, 24 Dec 2024 17:14:05 +0100 Subject: [PATCH 4/4] bump: Version 1.1.0 -> 1.2.0 (#983) --- docs/contribution.md | 2 +- docs/requirements-and-installation.md | 2 +- pyproject.toml | 4 ++-- 3 files changed, 4 insertions(+), 4 deletions(-) diff --git a/docs/contribution.md b/docs/contribution.md index 03923f695..50aed4466 100644 --- a/docs/contribution.md +++ b/docs/contribution.md @@ -29,7 +29,7 @@ $ pip install -e .[dev,cli] $ pip list -e Package Version Editable project location ------- ------- ------------------------- -anta 1.1.0 /mnt/lab/projects/anta +anta 1.2.0 /mnt/lab/projects/anta ``` Then, [`tox`](https://tox.wiki/) is configured with few environments to run CI locally: diff --git a/docs/requirements-and-installation.md b/docs/requirements-and-installation.md index 22faf7c12..e9fbbc7e1 100644 --- a/docs/requirements-and-installation.md +++ b/docs/requirements-and-installation.md @@ -84,7 +84,7 @@ which anta ```bash # Check ANTA version anta --version -anta, version v1.1.0 +anta, version v1.2.0 ``` ## EOS Requirements diff --git a/pyproject.toml b/pyproject.toml index d52d4876e..1e85b01f0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,7 +5,7 @@ build-backend = "setuptools.build_meta" [project] name = "anta" -version = "v1.1.0" +version = "v1.2.0" readme = "docs/README.md" authors = [{ name = "Arista Networks ANTA maintainers", email = "anta-dev@arista.com" }] maintainers = [ @@ -120,7 +120,7 @@ namespaces = false # Version ################################ [tool.bumpver] -current_version = "1.1.0" +current_version = "1.2.0" version_pattern = "MAJOR.MINOR.PATCH" commit_message = "bump: Version {old_version} -> {new_version}" commit = true