diff options
Diffstat (limited to 'anta/tests/routing/bgp.py')
| -rw-r--r-- | anta/tests/routing/bgp.py | 857 |
1 files changed, 541 insertions, 316 deletions
diff --git a/anta/tests/routing/bgp.py b/anta/tests/routing/bgp.py index 334ac3b..3b99a02 100644 --- a/anta/tests/routing/bgp.py +++ b/anta/tests/routing/bgp.py @@ -1,41 +1,40 @@ # Copyright (c) 2023-2024 Arista Networks, Inc. # Use of this source code is governed by the Apache License 2.0 # that can be found in the LICENSE file. -""" -BGP test functions -""" +"""Module related to BGP tests.""" + # Mypy does not understand AntaTest.Input typing # mypy: disable-error-code=attr-defined from __future__ import annotations from ipaddress import IPv4Address, IPv4Network, IPv6Address -from typing import Any, List, Optional, Union, cast +from typing import Any, ClassVar -from pydantic import BaseModel, Field, PositiveInt, model_validator, utils +from pydantic import BaseModel, Field, PositiveInt, model_validator +from pydantic.v1.utils import deep_update from pydantic_extra_types.mac_address import MacAddress from anta.custom_types import Afi, MultiProtocolCaps, Safi, Vni from anta.models import AntaCommand, AntaTemplate, AntaTest -from anta.tools.get_item import get_item -from anta.tools.get_value import get_value - -# Need to keep List for pydantic in python 3.8 +from anta.tools import get_item, get_value -def _add_bgp_failures(failures: dict[tuple[str, Union[str, None]], dict[str, Any]], afi: Afi, safi: Optional[Safi], vrf: str, issue: Any) -> None: - """ - Add a BGP failure entry to the given `failures` dictionary. +def _add_bgp_failures(failures: dict[tuple[str, str | None], dict[str, Any]], afi: Afi, safi: Safi | None, vrf: str, issue: str | dict[str, Any]) -> None: + """Add a BGP failure entry to the given `failures` dictionary. Note: This function modifies `failures` in-place. - Parameters: - failures (dict): The dictionary to which the failure will be added. - afi (Afi): The address family identifier. - vrf (str): The VRF name. - safi (Safi, optional): The subsequent address family identifier. - issue (Any): A description of the issue. Can be of any type. + Args: + ---- + failures: The dictionary to which the failure will be added. + afi: The address family identifier. + vrf: The VRF name. + safi: The subsequent address family identifier. + issue: A description of the issue. Can be of any type. - The `failures` dictionnary will have the following structure: + Example: + ------- + The `failures` dictionary will have the following structure: { ('afi1', 'safi1'): { 'afi': 'afi1', @@ -52,40 +51,43 @@ def _add_bgp_failures(failures: dict[tuple[str, Union[str, None]], dict[str, Any } } } + """ key = (afi, safi) - if safi: - failure_entry = failures.setdefault(key, {"afi": afi, "safi": safi, "vrfs": {}}) - else: - failure_entry = failures.setdefault(key, {"afi": afi, "vrfs": {}}) + failure_entry = failures.setdefault(key, {"afi": afi, "safi": safi, "vrfs": {}}) if safi else failures.setdefault(key, {"afi": afi, "vrfs": {}}) failure_entry["vrfs"][vrf] = issue -def _check_peer_issues(peer_data: Optional[dict[str, Any]]) -> dict[str, Any]: - """ - Check for issues in BGP peer data. +def _check_peer_issues(peer_data: dict[str, Any] | None) -> dict[str, Any]: + """Check for issues in BGP peer data. - Parameters: - peer_data (dict, optional): The BGP peer data dictionary nested in the `show bgp <afi> <safi> summary` command. + Args: + ---- + peer_data: The BGP peer data dictionary nested in the `show bgp <afi> <safi> summary` command. - Returns: + Returns + ------- dict: Dictionary with keys indicating issues or an empty dictionary if no issues. + Raises + ------ + ValueError: If any of the required keys ("peerState", "inMsgQueue", "outMsgQueue") are missing in `peer_data`, i.e. invalid BGP peer data. + Example: + ------- {"peerNotFound": True} {"peerState": "Idle", "inMsgQueue": 2, "outMsgQueue": 0} {} - Raises: - ValueError: If any of the required keys ("peerState", "inMsgQueue", "outMsgQueue") are missing in `peer_data`, i.e. invalid BGP peer data. """ if peer_data is None: return {"peerNotFound": True} if any(key not in peer_data for key in ["peerState", "inMsgQueue", "outMsgQueue"]): - raise ValueError("Provided BGP peer data is invalid.") + msg = "Provided BGP peer data is invalid." + raise ValueError(msg) if peer_data["peerState"] != "Established" or peer_data["inMsgQueue"] != 0 or peer_data["outMsgQueue"] != 0: return {"peerState": peer_data["peerState"], "inMsgQueue": peer_data["inMsgQueue"], "outMsgQueue": peer_data["outMsgQueue"]} @@ -96,80 +98,110 @@ def _check_peer_issues(peer_data: Optional[dict[str, Any]]) -> dict[str, Any]: 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]]]]]: - """ - Identifies missing BGP routes and invalid or inactive route entries. + """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 (list[str]): The list of expected routes. - bgp_output (dict[str, Any]): The BGP output from the device. - peer (str): The IP address of the BGP peer. - vrf (str): The name of the VRF for which the routes need to be verified. - route_type (str, optional): The type of BGP routes. Defaults to 'advertised_routes'. + Args: + ---- + 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: + 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: - route = str(route) - failure = {"bgp_peers": {peer: {vrf: {route_type: {route: Any}}}}} + str_route = str(route) + failure = {"bgp_peers": {peer: {vrf: {route_type: {str_route: Any}}}}} # Check if the route is missing in the BGP output - if route not in 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][route] = "Not found" - failure_routes = utils.deep_update(failure_routes, failure) + 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[route]["bgpRoutePaths"][0]["routeType"]["valid"] - is_valid = bgp_output[route]["bgpRoutePaths"][0]["routeType"]["active"] + 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][route] = {"valid": is_valid, "active": is_active} - failure_routes = utils.deep_update(failure_routes, failure) + 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 class VerifyBGPPeerCount(AntaTest): - """ - This test verifies the count of BGP peers for a given address family. + """Verifies the count of BGP peers for a given address family. + + It supports multiple types of Address Families Identifiers (AFI) and Subsequent Address Family Identifiers (SAFI). + + For SR-TE SAFI, the EOS command supports sr-te first then ipv4/ipv6 (AFI) which is handled automatically in this test. - It supports multiple types of address families (AFI) and subsequent service families (SAFI). Please refer to the Input class attributes below for details. - Expected Results: - * success: If the count of BGP peers matches the expected count for each address family and VRF. - * failure: If the count of BGP peers does not match the expected count, or if BGP is not configured for an expected VRF or address family. + Expected Results + ---------------- + * Success: If the count of BGP peers matches the expected count for each address family and VRF. + * Failure: If the count of BGP peers does not match the expected count, or if BGP is not configured for an expected VRF or address family. + + Examples + -------- + ```yaml + anta.tests.routing: + bgp: + - VerifyBGPPeerCount: + address_families: + - afi: "evpn" + num_peers: 2 + - afi: "ipv4" + safi: "unicast" + vrf: "PROD" + num_peers: 2 + - afi: "ipv4" + safi: "unicast" + vrf: "default" + num_peers: 3 + - afi: "ipv4" + safi: "multicast" + vrf: "DEV" + num_peers: 3 + ``` """ name = "VerifyBGPPeerCount" description = "Verifies the count of BGP peers." - categories = ["bgp"] - commands = [ - AntaTemplate(template="show bgp {afi} {safi} summary vrf {vrf}"), - AntaTemplate(template="show bgp {afi} summary"), + categories: ClassVar[list[str]] = ["bgp"] + commands: ClassVar[list[AntaCommand | AntaTemplate]] = [ + AntaTemplate(template="show bgp {afi} {safi} summary vrf {vrf}", revision=3), + AntaTemplate(template="show bgp {afi} summary", revision=3), ] - class Input(AntaTest.Input): # pylint: disable=missing-class-docstring - address_families: List[BgpAfi] - """ - List of BGP address families (BgpAfi) - """ + class Input(AntaTest.Input): + """Input model for the VerifyBGPPeerCount test.""" + + address_families: list[BgpAfi] + """List of BGP address families (BgpAfi).""" + + class BgpAfi(BaseModel): + """Model for a BGP address family (AFI) and subsequent address family (SAFI).""" - class BgpAfi(BaseModel): # pylint: disable=missing-class-docstring afi: Afi - """BGP address family (AFI)""" - safi: Optional[Safi] = None + """BGP address family (AFI).""" + safi: Safi | None = None """Optional BGP subsequent service family (SAFI). If the input `afi` is `ipv4` or `ipv6`, a valid `safi` must be provided. @@ -181,12 +213,11 @@ class VerifyBGPPeerCount(AntaTest): If the input `afi` is not `ipv4` or `ipv6`, e.g. `evpn`, `vrf` must be `default`. """ num_peers: PositiveInt - """Number of expected BGP peer(s)""" + """Number of expected BGP peer(s).""" @model_validator(mode="after") def validate_inputs(self: BaseModel) -> BaseModel: - """ - Validate the inputs provided to the BgpAfi class. + """Validate the inputs provided to the BgpAfi class. If afi is either ipv4 or ipv6, safi must be provided. @@ -194,36 +225,54 @@ class VerifyBGPPeerCount(AntaTest): """ if self.afi in ["ipv4", "ipv6"]: if self.safi is None: - raise ValueError("'safi' must be provided when afi is ipv4 or ipv6") + msg = "'safi' must be provided when afi is ipv4 or ipv6" + raise ValueError(msg) elif self.safi is not None: - raise ValueError("'safi' must not be provided when afi is not ipv4 or ipv6") + msg = "'safi' must not be provided when afi is not ipv4 or ipv6" + raise ValueError(msg) elif self.vrf != "default": - raise ValueError("'vrf' must be default when afi is not ipv4 or ipv6") + msg = "'vrf' must be default when afi is not ipv4 or ipv6" + raise ValueError(msg) return self def render(self, template: AntaTemplate) -> list[AntaCommand]: + """Render the template for each BGP address family in the input list.""" commands = [] for afi in self.inputs.address_families: - if template == VerifyBGPPeerCount.commands[0] and afi.afi in ["ipv4", "ipv6"]: - commands.append(template.render(afi=afi.afi, safi=afi.safi, vrf=afi.vrf, num_peers=afi.num_peers)) + if template == VerifyBGPPeerCount.commands[0] and afi.afi in ["ipv4", "ipv6"] and afi.safi != "sr-te": + commands.append(template.render(afi=afi.afi, safi=afi.safi, vrf=afi.vrf)) + + # For SR-TE SAFI, the EOS command supports sr-te first then ipv4/ipv6 + elif template == VerifyBGPPeerCount.commands[0] and afi.afi in ["ipv4", "ipv6"] and afi.safi == "sr-te": + commands.append(template.render(afi=afi.safi, safi=afi.afi, vrf=afi.vrf)) elif template == VerifyBGPPeerCount.commands[1] and afi.afi not in ["ipv4", "ipv6"]: - commands.append(template.render(afi=afi.afi, vrf=afi.vrf, num_peers=afi.num_peers)) + commands.append(template.render(afi=afi.afi)) return commands @AntaTest.anta_test def test(self) -> None: + """Main test function for VerifyBGPPeerCount.""" self.result.is_success() failures: dict[tuple[str, Any], dict[str, Any]] = {} for command in self.instance_commands: + num_peers = None peer_count = 0 command_output = command.json_output - afi = cast(Afi, command.params.get("afi")) - safi = cast(Optional[Safi], command.params.get("safi")) - afi_vrf = cast(str, command.params.get("vrf")) - num_peers = cast(PositiveInt, command.params.get("num_peers")) + afi = command.params.afi + safi = command.params.safi + afi_vrf = command.params.vrf or "default" + + # Swapping AFI and SAFI in case of SR-TE + if afi == "sr-te": + afi, safi = safi, afi + + for input_entry in self.inputs.address_families: + if input_entry.afi == afi and input_entry.safi == safi and input_entry.vrf == afi_vrf: + num_peers = input_entry.num_peers + break if not (vrfs := command_output.get("vrfs")): _add_bgp_failures(failures=failures, afi=afi, safi=safi, vrf=afi_vrf, issue="Not Configured") @@ -243,37 +292,58 @@ class VerifyBGPPeerCount(AntaTest): class VerifyBGPPeersHealth(AntaTest): - """ - This test verifies the health of BGP peers. + """Verifies the health of BGP peers. It will validate that all BGP sessions are established and all message queues for these BGP sessions are empty for a given address family. - It supports multiple types of address families (AFI) and subsequent service families (SAFI). + It supports multiple types of Address Families Identifiers (AFI) and Subsequent Address Family Identifiers (SAFI). + + For SR-TE SAFI, the EOS command supports sr-te first then ipv4/ipv6 (AFI) which is handled automatically in this test. + Please refer to the Input class attributes below for details. - Expected Results: - * success: If all BGP sessions are established and all messages queues are empty for each address family and VRF. - * failure: If there are issues with any of the BGP sessions, or if BGP is not configured for an expected VRF or address family. + Expected Results + ---------------- + * Success: If all BGP sessions are established and all messages queues are empty for each address family and VRF. + * Failure: If there are issues with any of the BGP sessions, or if BGP is not configured for an expected VRF or address family. + + Examples + -------- + ```yaml + anta.tests.routing: + bgp: + - VerifyBGPPeersHealth: + address_families: + - afi: "evpn" + - afi: "ipv4" + safi: "unicast" + vrf: "default" + - afi: "ipv6" + safi: "unicast" + vrf: "DEV" + ``` """ name = "VerifyBGPPeersHealth" description = "Verifies the health of BGP peers" - categories = ["bgp"] - commands = [ - AntaTemplate(template="show bgp {afi} {safi} summary vrf {vrf}"), - AntaTemplate(template="show bgp {afi} summary"), + categories: ClassVar[list[str]] = ["bgp"] + commands: ClassVar[list[AntaCommand | AntaTemplate]] = [ + AntaTemplate(template="show bgp {afi} {safi} summary vrf {vrf}", revision=3), + AntaTemplate(template="show bgp {afi} summary", revision=3), ] - class Input(AntaTest.Input): # pylint: disable=missing-class-docstring - address_families: List[BgpAfi] - """ - List of BGP address families (BgpAfi) - """ + class Input(AntaTest.Input): + """Input model for the VerifyBGPPeersHealth test.""" + + address_families: list[BgpAfi] + """List of BGP address families (BgpAfi).""" + + class BgpAfi(BaseModel): + """Model for a BGP address family (AFI) and subsequent address family (SAFI).""" - class BgpAfi(BaseModel): # pylint: disable=missing-class-docstring afi: Afi - """BGP address family (AFI)""" - safi: Optional[Safi] = None + """BGP address family (AFI).""" + safi: Safi | None = None """Optional BGP subsequent service family (SAFI). If the input `afi` is `ipv4` or `ipv6`, a valid `safi` must be provided. @@ -287,8 +357,7 @@ class VerifyBGPPeersHealth(AntaTest): @model_validator(mode="after") def validate_inputs(self: BaseModel) -> BaseModel: - """ - Validate the inputs provided to the BgpAfi class. + """Validate the inputs provided to the BgpAfi class. If afi is either ipv4 or ipv6, safi must be provided. @@ -296,24 +365,33 @@ class VerifyBGPPeersHealth(AntaTest): """ if self.afi in ["ipv4", "ipv6"]: if self.safi is None: - raise ValueError("'safi' must be provided when afi is ipv4 or ipv6") + msg = "'safi' must be provided when afi is ipv4 or ipv6" + raise ValueError(msg) elif self.safi is not None: - raise ValueError("'safi' must not be provided when afi is not ipv4 or ipv6") + msg = "'safi' must not be provided when afi is not ipv4 or ipv6" + raise ValueError(msg) elif self.vrf != "default": - raise ValueError("'vrf' must be default when afi is not ipv4 or ipv6") + msg = "'vrf' must be default when afi is not ipv4 or ipv6" + raise ValueError(msg) return self def render(self, template: AntaTemplate) -> list[AntaCommand]: + """Render the template for each BGP address family in the input list.""" commands = [] for afi in self.inputs.address_families: - if template == VerifyBGPPeersHealth.commands[0] and afi.afi in ["ipv4", "ipv6"]: + if template == VerifyBGPPeersHealth.commands[0] and afi.afi in ["ipv4", "ipv6"] and afi.safi != "sr-te": commands.append(template.render(afi=afi.afi, safi=afi.safi, vrf=afi.vrf)) + + # For SR-TE SAFI, the EOS command supports sr-te first then ipv4/ipv6 + elif template == VerifyBGPPeersHealth.commands[0] and afi.afi in ["ipv4", "ipv6"] and afi.safi == "sr-te": + commands.append(template.render(afi=afi.safi, safi=afi.afi, vrf=afi.vrf)) elif template == VerifyBGPPeersHealth.commands[1] and afi.afi not in ["ipv4", "ipv6"]: - commands.append(template.render(afi=afi.afi, vrf=afi.vrf)) + commands.append(template.render(afi=afi.afi)) return commands @AntaTest.anta_test def test(self) -> None: + """Main test function for VerifyBGPPeersHealth.""" self.result.is_success() failures: dict[tuple[str, Any], dict[str, Any]] = {} @@ -321,9 +399,13 @@ class VerifyBGPPeersHealth(AntaTest): for command in self.instance_commands: command_output = command.json_output - afi = cast(Afi, command.params.get("afi")) - safi = cast(Optional[Safi], command.params.get("safi")) - afi_vrf = cast(str, command.params.get("vrf")) + afi = command.params.afi + safi = command.params.safi + + # Swapping AFI and SAFI in case of SR-TE + if afi == "sr-te": + afi, safi = safi, afi + afi_vrf = command.params.vrf or "default" if not (vrfs := command_output.get("vrfs")): _add_bgp_failures(failures=failures, afi=afi, safi=safi, vrf=afi_vrf, issue="Not Configured") @@ -349,37 +431,62 @@ class VerifyBGPPeersHealth(AntaTest): class VerifyBGPSpecificPeers(AntaTest): - """ - This test verifies the health of specific BGP peer(s). + """Verifies the health of specific BGP peer(s). It will validate that the BGP session is established and all message queues for this BGP session are empty for the given peer(s). - It supports multiple types of address families (AFI) and subsequent service families (SAFI). + It supports multiple types of Address Families Identifiers (AFI) and Subsequent Address Family Identifiers (SAFI). + + For SR-TE SAFI, the EOS command supports sr-te first then ipv4/ipv6 (AFI) which is handled automatically in this test. + Please refer to the Input class attributes below for details. - Expected Results: - * success: If the BGP session is established and all messages queues are empty for each given peer. - * failure: If the BGP session has issues or is not configured, or if BGP is not configured for an expected VRF or address family. + Expected Results + ---------------- + * Success: If the BGP session is established and all messages queues are empty for each given peer. + * Failure: If the BGP session has issues or is not configured, or if BGP is not configured for an expected VRF or address family. + + Examples + -------- + ```yaml + anta.tests.routing: + bgp: + - VerifyBGPSpecificPeers: + address_families: + - afi: "evpn" + peers: + - 10.1.0.1 + - 10.1.0.2 + - afi: "ipv4" + safi: "unicast" + peers: + - 10.1.254.1 + - 10.1.255.0 + - 10.1.255.2 + - 10.1.255.4 + ``` """ name = "VerifyBGPSpecificPeers" description = "Verifies the health of specific BGP peer(s)." - categories = ["bgp"] - commands = [ - AntaTemplate(template="show bgp {afi} {safi} summary vrf {vrf}"), - AntaTemplate(template="show bgp {afi} summary"), + categories: ClassVar[list[str]] = ["bgp"] + commands: ClassVar[list[AntaCommand | AntaTemplate]] = [ + AntaTemplate(template="show bgp {afi} {safi} summary vrf {vrf}", revision=3), + AntaTemplate(template="show bgp {afi} summary", revision=3), ] - class Input(AntaTest.Input): # pylint: disable=missing-class-docstring - address_families: List[BgpAfi] - """ - List of BGP address families (BgpAfi) - """ + class Input(AntaTest.Input): + """Input model for the VerifyBGPSpecificPeers test.""" + + address_families: list[BgpAfi] + """List of BGP address families (BgpAfi).""" + + class BgpAfi(BaseModel): + """Model for a BGP address family (AFI) and subsequent address family (SAFI).""" - class BgpAfi(BaseModel): # pylint: disable=missing-class-docstring afi: Afi - """BGP address family (AFI)""" - safi: Optional[Safi] = None + """BGP address family (AFI).""" + safi: Safi | None = None """Optional BGP subsequent service family (SAFI). If the input `afi` is `ipv4` or `ipv6`, a valid `safi` must be provided. @@ -392,13 +499,12 @@ class VerifyBGPSpecificPeers(AntaTest): If the input `afi` is not `ipv4` or `ipv6`, e.g. `evpn`, `vrf` must be `default`. """ - peers: List[Union[IPv4Address, IPv6Address]] - """List of BGP IPv4 or IPv6 peer""" + peers: list[IPv4Address | IPv6Address] + """List of BGP IPv4 or IPv6 peer.""" @model_validator(mode="after") def validate_inputs(self: BaseModel) -> BaseModel: - """ - Validate the inputs provided to the BgpAfi class. + """Validate the inputs provided to the BgpAfi class. If afi is either ipv4 or ipv6, safi must be provided and vrf must NOT be all. @@ -406,26 +512,37 @@ class VerifyBGPSpecificPeers(AntaTest): """ if self.afi in ["ipv4", "ipv6"]: if self.safi is None: - raise ValueError("'safi' must be provided when afi is ipv4 or ipv6") + msg = "'safi' must be provided when afi is ipv4 or ipv6" + raise ValueError(msg) if self.vrf == "all": - raise ValueError("'all' is not supported in this test. Use VerifyBGPPeersHealth test instead.") + msg = "'all' is not supported in this test. Use VerifyBGPPeersHealth test instead." + raise ValueError(msg) elif self.safi is not None: - raise ValueError("'safi' must not be provided when afi is not ipv4 or ipv6") + msg = "'safi' must not be provided when afi is not ipv4 or ipv6" + raise ValueError(msg) elif self.vrf != "default": - raise ValueError("'vrf' must be default when afi is not ipv4 or ipv6") + msg = "'vrf' must be default when afi is not ipv4 or ipv6" + raise ValueError(msg) return self def render(self, template: AntaTemplate) -> list[AntaCommand]: + """Render the template for each BGP address family in the input list.""" commands = [] + for afi in self.inputs.address_families: - if template == VerifyBGPSpecificPeers.commands[0] and afi.afi in ["ipv4", "ipv6"]: - commands.append(template.render(afi=afi.afi, safi=afi.safi, vrf=afi.vrf, peers=afi.peers)) + if template == VerifyBGPSpecificPeers.commands[0] and afi.afi in ["ipv4", "ipv6"] and afi.safi != "sr-te": + commands.append(template.render(afi=afi.afi, safi=afi.safi, vrf=afi.vrf)) + + # For SR-TE SAFI, the EOS command supports sr-te first then ipv4/ipv6 + elif template == VerifyBGPSpecificPeers.commands[0] and afi.afi in ["ipv4", "ipv6"] and afi.safi == "sr-te": + commands.append(template.render(afi=afi.safi, safi=afi.afi, vrf=afi.vrf)) elif template == VerifyBGPSpecificPeers.commands[1] and afi.afi not in ["ipv4", "ipv6"]: - commands.append(template.render(afi=afi.afi, vrf=afi.vrf, peers=afi.peers)) + commands.append(template.render(afi=afi.afi)) return commands @AntaTest.anta_test def test(self) -> None: + """Main test function for VerifyBGPSpecificPeers.""" self.result.is_success() failures: dict[tuple[str, Any], dict[str, Any]] = {} @@ -433,10 +550,18 @@ class VerifyBGPSpecificPeers(AntaTest): for command in self.instance_commands: command_output = command.json_output - afi = cast(Afi, command.params.get("afi")) - safi = cast(Optional[Safi], command.params.get("safi")) - afi_vrf = cast(str, command.params.get("vrf")) - afi_peers = cast(List[Union[IPv4Address, IPv6Address]], command.params.get("peers", [])) + afi = command.params.afi + safi = command.params.safi + afi_vrf = command.params.vrf or "default" + + # Swapping AFI and SAFI in case of SR-TE + if afi == "sr-te": + afi, safi = safi, afi + + for input_entry in self.inputs.address_families: + if input_entry.afi == afi and input_entry.safi == safi and input_entry.vrf == afi_vrf: + afi_peers = input_entry.peers + break if not (vrfs := command_output.get("vrfs")): _add_bgp_failures(failures=failures, afi=afi, safi=safi, vrf=afi_vrf, issue="Not Configured") @@ -458,63 +583,82 @@ class VerifyBGPSpecificPeers(AntaTest): class VerifyBGPExchangedRoutes(AntaTest): - """ - Verifies if the BGP peers have correctly advertised and received routes. + """Verifies if the BGP peers have correctly advertised and received routes. + The route type should be 'valid' and 'active' for a specified VRF. - 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'. + 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'. + + Examples + -------- + ```yaml + anta.tests.routing: + bgp: + - VerifyBGPExchangedRoutes: + bgp_peers: + - peer_address: 172.30.255.5 + vrf: default + advertised_routes: + - 192.0.254.5/32 + received_routes: + - 192.0.255.4/32 + - peer_address: 172.30.255.1 + vrf: default + advertised_routes: + - 192.0.255.1/32 + - 192.0.254.5/32 + received_routes: + - 192.0.254.3/32 + ``` """ name = "VerifyBGPExchangedRoutes" - description = "Verifies if BGP peers have correctly advertised/received routes with type as valid and active for a specified VRF." - categories = ["bgp"] - commands = [ - AntaTemplate(template="show bgp neighbors {peer} advertised-routes vrf {vrf}"), - AntaTemplate(template="show bgp neighbors {peer} routes vrf {vrf}"), + description = "Verifies the advertised and received routes of BGP peers." + categories: ClassVar[list[str]] = ["bgp"] + commands: ClassVar[list[AntaCommand | AntaTemplate]] = [ + AntaTemplate(template="show bgp neighbors {peer} advertised-routes vrf {vrf}", revision=3), + AntaTemplate(template="show bgp neighbors {peer} routes vrf {vrf}", revision=3), ] class Input(AntaTest.Input): - """ - Input parameters of the testcase. - """ + """Input model for the VerifyBGPExchangedRoutes test.""" - bgp_peers: List[BgpNeighbors] - """List of BGP peers""" + bgp_peers: list[BgpNeighbor] + """List of BGP neighbors.""" - class BgpNeighbors(BaseModel): - """ - This class defines the details of a BGP peer. - """ + class BgpNeighbor(BaseModel): + """Model for a BGP neighbor.""" peer_address: IPv4Address - """IPv4 address of a BGP peer""" + """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 of a BGP peer.""" - received_routes: List[IPv4Network] - """List of received routes of a BGP peer.""" + advertised_routes: list[IPv4Network] + """List of advertised routes in CIDR format.""" + received_routes: list[IPv4Network] + """List of received routes in CIDR format.""" def render(self, template: AntaTemplate) -> list[AntaCommand]: - """Renders the template with the provided inputs. Returns a list of commands to be executed.""" - - return [ - template.render(peer=bgp_peer.peer_address, vrf=bgp_peer.vrf, advertised_routes=bgp_peer.advertised_routes, received_routes=bgp_peer.received_routes) - for bgp_peer in self.inputs.bgp_peers - ] + """Render the template for each BGP neighbor 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 = str(command.params["peer"]) - vrf = command.params["vrf"] - advertised_routes = command.params["advertised_routes"] - received_routes = command.params["received_routes"] + 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 @@ -530,7 +674,7 @@ class VerifyBGPExchangedRoutes(AntaTest): # Validate received routes else: failure_routes = _add_bgp_routes_failure(received_routes, bgp_routes, peer, vrf, route_type="received_routes") - failures = utils.deep_update(failures, failure_routes) + failures = deep_update(failures, failure_routes) if not failures["bgp_peers"]: self.result.is_success() @@ -539,40 +683,51 @@ class VerifyBGPExchangedRoutes(AntaTest): class VerifyBGPPeerMPCaps(AntaTest): - """ - Verifies the multiprotocol capabilities of a BGP peer in a specified VRF. - 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. + """Verifies the multiprotocol capabilities of a BGP peer in a specified VRF. + + 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. + + Examples + -------- + ```yaml + anta.tests.routing: + bgp: + - VerifyBGPPeerMPCaps: + bgp_peers: + - peer_address: 172.30.11.1 + vrf: default + capabilities: + - ipv4Unicast + ``` """ name = "VerifyBGPPeerMPCaps" - description = "Verifies the multiprotocol capabilities of a BGP peer in a specified VRF" - categories = ["bgp"] - commands = [AntaCommand(command="show bgp neighbors vrf all")] + 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)] class Input(AntaTest.Input): - """ - Input parameters of the testcase. - """ + """Input model for the VerifyBGPPeerMPCaps test.""" - bgp_peers: List[BgpPeers] + bgp_peers: list[BgpPeer] """List of BGP peers""" - class BgpPeers(BaseModel): - """ - This class defines the details of a BGP peer. - """ + class BgpPeer(BaseModel): + """Model for a BGP peer.""" peer_address: IPv4Address - """IPv4 address of a BGP peer""" + """IPv4 address of a BGP peer.""" vrf: str = "default" """Optional VRF for BGP peer. If not provided, it defaults to `default`.""" - capabilities: List[MultiProtocolCaps] - """Multiprotocol capabilities""" + capabilities: list[MultiProtocolCaps] + """List of multiprotocol capabilities to be verified.""" @AntaTest.anta_test def test(self) -> None: + """Main test function for VerifyBGPPeerMPCaps.""" failures: dict[str, Any] = {"bgp_peers": {}} # Iterate over each bgp peer @@ -588,7 +743,7 @@ class VerifyBGPPeerMPCaps(AntaTest): or (bgp_output := get_item(bgp_output, "peerAddress", peer)) is None ): failure["bgp_peers"][peer][vrf] = {"status": "Not configured"} - failures = utils.deep_update(failures, failure) + failures = deep_update(failures, failure) continue # Check each capability @@ -599,12 +754,12 @@ class VerifyBGPPeerMPCaps(AntaTest): # Check if capabilities are missing if not capability_output: failure["bgp_peers"][peer][vrf][capability] = "not found" - failures = utils.deep_update(failures, failure) + failures = deep_update(failures, failure) # 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 = utils.deep_update(failures, failure) + failures = deep_update(failures, failure) # Check if there are any failures if not failures["bgp_peers"]: @@ -614,38 +769,47 @@ class VerifyBGPPeerMPCaps(AntaTest): class VerifyBGPPeerASNCap(AntaTest): - """ - Verifies the four octet asn capabilities of a BGP peer in a specified VRF. - 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. + """Verifies the four octet asn capabilities of a BGP peer in a specified VRF. + + 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. + + Examples + -------- + ```yaml + anta.tests.routing: + bgp: + - VerifyBGPPeerASNCap: + bgp_peers: + - peer_address: 172.30.11.1 + vrf: default + ``` """ name = "VerifyBGPPeerASNCap" - description = "Verifies the four octet asn capabilities of a BGP peer in a specified VRF." - categories = ["bgp"] - commands = [AntaCommand(command="show bgp neighbors vrf all")] + 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)] class Input(AntaTest.Input): - """ - Input parameters of the testcase. - """ + """Input model for the VerifyBGPPeerASNCap test.""" - bgp_peers: List[BgpPeers] - """List of BGP peers""" + bgp_peers: list[BgpPeer] + """List of BGP peers.""" - class BgpPeers(BaseModel): - """ - This class defines the details of a BGP peer. - """ + class BgpPeer(BaseModel): + """Model for a BGP peer.""" peer_address: IPv4Address - """IPv4 address of a BGP peer""" + """IPv4 address of a BGP peer.""" vrf: str = "default" """Optional VRF for BGP peer. If not provided, it defaults to `default`.""" @AntaTest.anta_test def test(self) -> None: + """Main test function for VerifyBGPPeerASNCap.""" failures: dict[str, Any] = {"bgp_peers": {}} # Iterate over each bgp peer @@ -660,7 +824,7 @@ class VerifyBGPPeerASNCap(AntaTest): or (bgp_output := get_item(bgp_output, "peerAddress", peer)) is None ): failure["bgp_peers"][peer][vrf] = {"status": "Not configured"} - failures = utils.deep_update(failures, failure) + failures = deep_update(failures, failure) continue bgp_output = get_value(bgp_output, "neighborCapabilities.fourOctetAsnCap") @@ -668,12 +832,12 @@ class VerifyBGPPeerASNCap(AntaTest): # Check if four octet asn capabilities are found if not bgp_output: failure["bgp_peers"][peer][vrf] = {"fourOctetAsnCap": "not found"} - failures = utils.deep_update(failures, failure) + failures = deep_update(failures, failure) # 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 = utils.deep_update(failures, failure) + failures = deep_update(failures, failure) # Check if there are any failures if not failures["bgp_peers"]: @@ -683,38 +847,47 @@ class VerifyBGPPeerASNCap(AntaTest): class VerifyBGPPeerRouteRefreshCap(AntaTest): - """ - Verifies the route refresh capabilities of a BGP peer in a specified VRF. - 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. + """Verifies the route refresh capabilities of a BGP peer in a specified VRF. + + 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. + + Examples + -------- + ```yaml + anta.tests.routing: + bgp: + - VerifyBGPPeerRouteRefreshCap: + bgp_peers: + - peer_address: 172.30.11.1 + vrf: default + ``` """ name = "VerifyBGPPeerRouteRefreshCap" - description = "Verifies the route refresh capabilities of a BGP peer in a specified VRF." - categories = ["bgp"] - commands = [AntaCommand(command="show bgp neighbors vrf all")] + 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)] class Input(AntaTest.Input): - """ - Input parameters of the testcase. - """ + """Input model for the VerifyBGPPeerRouteRefreshCap test.""" - bgp_peers: List[BgpPeers] + bgp_peers: list[BgpPeer] """List of BGP peers""" - class BgpPeers(BaseModel): - """ - This class defines the details of a BGP peer. - """ + class BgpPeer(BaseModel): + """Model for a BGP peer.""" peer_address: IPv4Address - """IPv4 address of a BGP peer""" + """IPv4 address of a BGP peer.""" vrf: str = "default" """Optional VRF for BGP peer. If not provided, it defaults to `default`.""" @AntaTest.anta_test def test(self) -> None: + """Main test function for VerifyBGPPeerRouteRefreshCap.""" failures: dict[str, Any] = {"bgp_peers": {}} # Iterate over each bgp peer @@ -729,7 +902,7 @@ class VerifyBGPPeerRouteRefreshCap(AntaTest): or (bgp_output := get_item(bgp_output, "peerAddress", peer)) is None ): failure["bgp_peers"][peer][vrf] = {"status": "Not configured"} - failures = utils.deep_update(failures, failure) + failures = deep_update(failures, failure) continue bgp_output = get_value(bgp_output, "neighborCapabilities.routeRefreshCap") @@ -737,12 +910,12 @@ class VerifyBGPPeerRouteRefreshCap(AntaTest): # Check if route refresh capabilities are found if not bgp_output: failure["bgp_peers"][peer][vrf] = {"routeRefreshCap": "not found"} - failures = utils.deep_update(failures, failure) + failures = deep_update(failures, failure) # 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 = utils.deep_update(failures, failure) + failures = deep_update(failures, failure) # Check if there are any failures if not failures["bgp_peers"]: @@ -752,30 +925,40 @@ class VerifyBGPPeerRouteRefreshCap(AntaTest): class VerifyBGPPeerMD5Auth(AntaTest): - """ - Verifies the MD5 authentication and state of IPv4 BGP peers in a specified VRF. - 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. + """Verifies the MD5 authentication and state of IPv4 BGP peers in a specified VRF. + + 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. + + Examples + -------- + ```yaml + anta.tests.routing: + bgp: + - VerifyBGPPeerMD5Auth: + bgp_peers: + - peer_address: 172.30.11.1 + vrf: default + - peer_address: 172.30.11.5 + vrf: default + ``` """ name = "VerifyBGPPeerMD5Auth" - description = "Verifies the MD5 authentication and state of IPv4 BGP peers in a specified VRF" - categories = ["routing", "bgp"] - commands = [AntaCommand(command="show bgp neighbors vrf all")] + 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)] class Input(AntaTest.Input): - """ - Input parameters of the test case. - """ + """Input model for the VerifyBGPPeerMD5Auth test.""" - bgp_peers: List[BgpPeers] - """List of IPv4 BGP peers""" + bgp_peers: list[BgpPeer] + """List of IPv4 BGP peers.""" - class BgpPeers(BaseModel): - """ - This class defines the details of an IPv4 BGP peer. - """ + class BgpPeer(BaseModel): + """Model for a BGP peer.""" peer_address: IPv4Address """IPv4 address of BGP peer.""" @@ -784,6 +967,7 @@ class VerifyBGPPeerMD5Auth(AntaTest): @AntaTest.anta_test def test(self) -> None: + """Main test function for VerifyBGPPeerMD5Auth.""" failures: dict[str, Any] = {"bgp_peers": {}} # Iterate over each command @@ -798,7 +982,7 @@ class VerifyBGPPeerMD5Auth(AntaTest): or (bgp_output := get_item(bgp_output, "peerAddress", peer)) is None ): failure["bgp_peers"][peer][vrf] = {"status": "Not configured"} - failures = utils.deep_update(failures, failure) + failures = deep_update(failures, failure) continue # Check if BGP peer state and authentication @@ -806,7 +990,7 @@ class VerifyBGPPeerMD5Auth(AntaTest): 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 = utils.deep_update(failures, failure) + failures = deep_update(failures, failure) # Check if there are any failures if not failures["bgp_peers"]: @@ -816,45 +1000,60 @@ class VerifyBGPPeerMD5Auth(AntaTest): class VerifyEVPNType2Route(AntaTest): - """ - This test verifies the EVPN Type-2 routes for a given IPv4 or MAC address and VNI. - - 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. + """Verifies the EVPN Type-2 routes for a given IPv4 or MAC address and VNI. + + 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. + + Examples + -------- + ```yaml + anta.tests.routing: + bgp: + - VerifyEVPNType2Route: + vxlan_endpoints: + - address: 192.168.20.102 + vni: 10020 + - address: aac1.ab5d.b41e + vni: 10010 + ``` """ name = "VerifyEVPNType2Route" description = "Verifies the EVPN Type-2 routes for a given IPv4 or MAC address and VNI." - categories = ["routing", "bgp"] - commands = [AntaTemplate(template="show bgp evpn route-type mac-ip {address} vni {vni}")] + categories: ClassVar[list[str]] = ["bgp"] + commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaTemplate(template="show bgp evpn route-type mac-ip {address} vni {vni}", revision=2)] class Input(AntaTest.Input): - """Inputs for the VerifyEVPNType2Route test.""" + """Input model for the VerifyEVPNType2Route test.""" - vxlan_endpoints: List[VxlanEndpoint] - """List of VXLAN endpoints to verify""" + vxlan_endpoints: list[VxlanEndpoint] + """List of VXLAN endpoints to verify.""" class VxlanEndpoint(BaseModel): - """VXLAN endpoint input model.""" + """Model for a VXLAN endpoint.""" - address: Union[IPv4Address, MacAddress] - """IPv4 or MAC address of the VXLAN endpoint""" + address: IPv4Address | MacAddress + """IPv4 or MAC address of the VXLAN endpoint.""" vni: Vni - """VNI of the VXLAN endpoint""" + """VNI of the VXLAN endpoint.""" def render(self, template: AntaTemplate) -> list[AntaCommand]: - return [template.render(address=endpoint.address, vni=endpoint.vni) for endpoint in self.inputs.vxlan_endpoints] + """Render the template for each VXLAN endpoint in the input list.""" + return [template.render(address=str(endpoint.address), vni=endpoint.vni) for endpoint in self.inputs.vxlan_endpoints] @AntaTest.anta_test 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 = str(command.params["address"]) - vni = command.params["vni"] + address = command.params.address + vni = command.params.vni # Verify that the VXLAN endpoint is in the BGP EVPN table evpn_routes = command.json_output["evpnRoutes"] if not evpn_routes: @@ -878,30 +1077,40 @@ class VerifyEVPNType2Route(AntaTest): class VerifyBGPAdvCommunities(AntaTest): - """ - Verifies if the advertised communities of BGP peers are standard, extended, and large in the specified VRF. - 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. + """Verifies if the advertised communities of BGP peers are standard, extended, and large in the specified VRF. + + 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. + + Examples + -------- + ```yaml + anta.tests.routing: + bgp: + - VerifyBGPAdvCommunities: + bgp_peers: + - peer_address: 172.30.11.17 + vrf: default + - peer_address: 172.30.11.21 + vrf: default + ``` """ name = "VerifyBGPAdvCommunities" - description = "Verifies if the advertised communities of BGP peers are standard, extended, and large in the specified VRF." - categories = ["routing", "bgp"] - commands = [AntaCommand(command="show bgp neighbors vrf all")] + 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)] class Input(AntaTest.Input): - """ - Input parameters for the test. - """ + """Input model for the VerifyBGPAdvCommunities test.""" - bgp_peers: List[BgpPeers] - """List of BGP peers""" + bgp_peers: list[BgpPeer] + """List of BGP peers.""" - class BgpPeers(BaseModel): - """ - This class defines the details of a BGP peer. - """ + class BgpPeer(BaseModel): + """Model for a BGP peer.""" peer_address: IPv4Address """IPv4 address of a BGP peer.""" @@ -910,6 +1119,7 @@ class VerifyBGPAdvCommunities(AntaTest): @AntaTest.anta_test def test(self) -> None: + """Main test function for VerifyBGPAdvCommunities.""" failures: dict[str, Any] = {"bgp_peers": {}} # Iterate over each bgp peer @@ -924,14 +1134,14 @@ class VerifyBGPAdvCommunities(AntaTest): or (bgp_output := get_item(bgp_output, "peerAddress", peer)) is None ): failure["bgp_peers"][peer][vrf] = {"status": "Not configured"} - failures = utils.deep_update(failures, failure) + failures = deep_update(failures, failure) continue # 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 = utils.deep_update(failures, failure) + failures = deep_update(failures, failure) if not failures["bgp_peers"]: self.result.is_success() @@ -940,42 +1150,57 @@ class VerifyBGPAdvCommunities(AntaTest): class VerifyBGPTimers(AntaTest): - """ - Verifies if the BGP peers are configured with the correct hold and keep-alive timers in the specified VRF. - Expected results: - * success: The test will pass if the hold and keep-alive timers are correct for BGP peers in the specified VRF. - * failure: The test will fail if BGP peers are not found or hold and keep-alive timers are not correct in the specified VRF. + """Verifies if the BGP peers are configured with the correct hold and keep-alive timers in the specified VRF. + + Expected Results + ---------------- + * Success: The test will pass if the hold and keep-alive timers are correct for BGP peers in the specified VRF. + * Failure: The test will fail if BGP peers are not found or hold and keep-alive timers are not correct in the specified VRF. + + Examples + -------- + ```yaml + anta.tests.routing: + bgp: + - VerifyBGPTimers: + bgp_peers: + - peer_address: 172.30.11.1 + vrf: default + hold_time: 180 + keep_alive_time: 60 + - peer_address: 172.30.11.5 + vrf: default + hold_time: 180 + keep_alive_time: 60 + ``` """ name = "VerifyBGPTimers" - description = "Verifies if the BGP peers are configured with the correct hold and keep alive timers in the specified VRF." - categories = ["routing", "bgp"] - commands = [AntaCommand(command="show bgp neighbors vrf all")] + 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)] class Input(AntaTest.Input): - """ - Input parameters for the test. - """ + """Input model for the VerifyBGPTimers test.""" - bgp_peers: List[BgpPeers] + bgp_peers: list[BgpPeer] """List of BGP peers""" - class BgpPeers(BaseModel): - """ - This class defines the details of a BGP peer. - """ + class BgpPeer(BaseModel): + """Model for a BGP peer.""" peer_address: IPv4Address - """IPv4 address of a BGP peer""" + """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""" + """BGP hold time in seconds.""" keep_alive_time: int = Field(ge=0, le=3600) - """BGP keep-alive time in seconds""" + """BGP keep-alive time in seconds.""" @AntaTest.anta_test def test(self) -> None: + """Main test function for VerifyBGPTimers.""" failures: dict[str, Any] = {} # Iterate over each bgp peer |