diff --git a/anta/custom_types.py b/anta/custom_types.py index ccd0b5f6e..aa36d2fc7 100644 --- a/anta/custom_types.py +++ b/anta/custom_types.py @@ -24,6 +24,13 @@ """Match hostname like `my-hostname`, `my-hostname-1`, `my-hostname-1-2`.""" +# Regular expression for BGP redistributed routes +REGEX_IPV4_UNICAST = r"ipv4[-_ ]?unicast$" +REGEX_IPV4_MULTICAST = r"ipv4[-_ ]?multicast$" +REGEX_IPV6_UNICAST = r"ipv6[-_ ]?unicast$" +REGEX_IPV6_MULTICAST = r"ipv6[-_ ]?multicast$" + + def aaa_group_prefix(v: str) -> str: """Prefix the AAA method with 'group' if it is known.""" built_in_methods = ["local", "none", "logging"] @@ -92,10 +99,10 @@ def bgp_multiprotocol_capabilities_abbreviations(value: str) -> str: patterns = { f"{r'dynamic[-_ ]?path[-_ ]?selection$'}": "dps", f"{r'dps$'}": "dps", - f"{r'ipv4[-_ ]?unicast$'}": "ipv4Unicast", - f"{r'ipv6[-_ ]?unicast$'}": "ipv6Unicast", - f"{r'ipv4[-_ ]?multicast$'}": "ipv4Multicast", - f"{r'ipv6[-_ ]?multicast$'}": "ipv6Multicast", + f"{REGEX_IPV4_UNICAST}": "ipv4Unicast", + f"{REGEX_IPV6_UNICAST}": "ipv6Unicast", + f"{REGEX_IPV4_MULTICAST}": "ipv4Multicast", + f"{REGEX_IPV6_MULTICAST}": "ipv6Multicast", f"{r'ipv4[-_ ]?labeled[-_ ]?Unicast$'}": "ipv4MplsLabels", f"{r'ipv4[-_ ]?mpls[-_ ]?labels$'}": "ipv4MplsLabels", f"{r'ipv6[-_ ]?labeled[-_ ]?Unicast$'}": "ipv6MplsLabels", @@ -132,6 +139,54 @@ def validate_regex(value: str) -> str: return value +def bgp_redistributed_route_proto_abbreviations(value: str) -> str: + """Abbreviations for different BGP redistributed route protocols. + + Handles different separators (hyphen, underscore, space) and case sensitivity. + + Examples + -------- + ```python + >>> bgp_redistributed_route_proto_abbreviations("IPv4 Unicast") + 'v4u' + >>> bgp_redistributed_route_proto_abbreviations("IPv4-multicast") + 'v4m' + >>> bgp_redistributed_route_proto_abbreviations("IPv6_multicast") + 'v6m' + >>> bgp_redistributed_route_proto_abbreviations("ipv6unicast") + 'v6u' + ``` + """ + patterns = {REGEX_IPV4_UNICAST: "v4u", REGEX_IPV4_MULTICAST: "v4m", REGEX_IPV6_UNICAST: "v6u", REGEX_IPV6_MULTICAST: "v6m"} + + for pattern, replacement in patterns.items(): + match = re.match(pattern, value, re.IGNORECASE) + if match: + return replacement + + return value + + +def update_bgp_redistributed_proto_user(value: str) -> str: + """Update BGP redistributed route `User` proto with EOS SDK. + + Examples + -------- + ```python + >>> update_bgp_redistributed_proto_user("User") + 'EOS SDK' + >>> update_bgp_redistributed_proto_user("Bgp") + 'Bgp' + >>> update_bgp_redistributed_proto_user("RIP") + 'RIP' + ``` + """ + if value == "User": + value = "EOS SDK" + + return value + + # AntaTest.Input types AAAAuthMethod = Annotated[str, AfterValidator(aaa_group_prefix)] Vlan = Annotated[int, Field(ge=0, le=4094)] @@ -319,3 +374,23 @@ def snmp_v3_prefix(auth_type: Literal["auth", "priv", "noauth"]) -> str: "inVersionErrs", "inBadCommunityNames", "inBadCommunityUses", "inParseErrs", "outTooBigErrs", "outNoSuchNameErrs", "outBadValueErrs", "outGeneralErrs" ] SnmpVersionV3AuthType = Annotated[Literal["auth", "priv", "noauth"], AfterValidator(snmp_v3_prefix)] +RedistributedProtocol = Annotated[ + Literal[ + "AttachedHost", + "Bgp", + "Connected", + "Dynamic", + "IS-IS", + "OSPF Internal", + "OSPF External", + "OSPF Nssa-External", + "OSPFv3 Internal", + "OSPFv3 External", + "OSPFv3 Nssa-External", + "RIP", + "Static", + "User", + ], + AfterValidator(update_bgp_redistributed_proto_user), +] +RedistributedAfiSafi = Annotated[Literal["v4u", "v4m", "v6u", "v6m"], BeforeValidator(bgp_redistributed_route_proto_abbreviations)] diff --git a/anta/input_models/routing/bgp.py b/anta/input_models/routing/bgp.py index 945b0305c..5c9226ec1 100644 --- a/anta/input_models/routing/bgp.py +++ b/anta/input_models/routing/bgp.py @@ -12,7 +12,7 @@ from pydantic import BaseModel, ConfigDict, Field, PositiveInt, model_validator from pydantic_extra_types.mac_address import MacAddress -from anta.custom_types import Afi, BgpDropStats, BgpUpdateError, MultiProtocolCaps, Safi, Vni +from anta.custom_types import Afi, BgpDropStats, BgpUpdateError, MultiProtocolCaps, RedistributedAfiSafi, RedistributedProtocol, Safi, Vni if TYPE_CHECKING: import sys @@ -68,8 +68,7 @@ class BgpAddressFamily(BaseModel): check_peer_state: bool = False """Flag to check if the peers are established with negotiated AFI/SAFI. Defaults to `False`. - Can be enabled in the `VerifyBGPPeerCount` tests. - """ + Can be enabled in the `VerifyBGPPeerCount` tests.""" @model_validator(mode="after") def validate_inputs(self) -> Self: @@ -256,3 +255,73 @@ def __str__(self) -> str: - Next-hop: 192.168.66.101 Origin: Igp """ return f"Next-hop: {self.nexthop} Origin: {self.origin}" + + +class BgpVrf(BaseModel): + """Model representing a VRF in a BGP instance.""" + + vrf: str = "default" + """VRF context.""" + address_families: list[AddressFamilyConfig] + """List of address family configuration.""" + + def __str__(self) -> str: + """Return a human-readable string representation of the BgpVrf for reporting. + + Examples + -------- + - VRF: default + """ + return f"VRF: {self.vrf}" + + +class RedistributedRouteConfig(BaseModel): + """Model representing a BGP redistributed route configuration.""" + + proto: RedistributedProtocol + """The redistributed protocol.""" + include_leaked: bool = False + """Flag to include leaked routes of the redistributed protocol while redistributing.""" + route_map: str | None = None + """Optional route map applied to the redistribution.""" + + @model_validator(mode="after") + def validate_inputs(self) -> Self: + """Validate that 'include_leaked' is not set when the redistributed protocol is AttachedHost, User, Dynamic, or RIP.""" + if self.include_leaked and self.proto in ["AttachedHost", "EOS SDK", "Dynamic", "RIP"]: + msg = f"'include_leaked' field is not supported for redistributed protocol '{self.proto}'" + raise ValueError(msg) + return self + + def __str__(self) -> str: + """Return a human-readable string representation of the RedistributedRouteConfig for reporting. + + Examples + -------- + - Proto: Connected, Include Leaked: True, Route Map: RM-CONN-2-BGP + """ + base_string = f"Proto: {self.proto}" + if self.include_leaked: + base_string += f", Include Leaked: {self.include_leaked}" + if self.route_map: + base_string += f", Route Map: {self.route_map}" + return base_string + + +class AddressFamilyConfig(BaseModel): + """Model representing a BGP address family configuration.""" + + afi_safi: RedistributedAfiSafi + """AFI/SAFI abbreviation per EOS.""" + redistributed_routes: list[RedistributedRouteConfig] + """List of redistributed route configuration.""" + + def __str__(self) -> str: + """Return a human-readable string representation of the AddressFamilyConfig for reporting. + + Examples + -------- + - AFI-SAFI: IPv4 Unicast + """ + mappings = {"v4u": "IPv4 Unicast", "v4m": "IPv4 Multicast", "v6u": "IPv6 Unicast", "v6m": "IPv6 Multicast"} + return f"AFI-SAFI: {mappings[self.afi_safi]}" diff --git a/anta/tests/routing/bgp.py b/anta/tests/routing/bgp.py index b2677340f..7522c2549 100644 --- a/anta/tests/routing/bgp.py +++ b/anta/tests/routing/bgp.py @@ -11,13 +11,15 @@ from pydantic import field_validator -from anta.input_models.routing.bgp import BgpAddressFamily, BgpAfi, BgpNeighbor, BgpPeer, BgpRoute, VxlanEndpoint +from anta.input_models.routing.bgp import BgpAddressFamily, BgpAfi, BgpNeighbor, BgpPeer, BgpRoute, BgpVrf, VxlanEndpoint from anta.models import AntaCommand, AntaTemplate, AntaTest from anta.tools import format_data, get_item, get_value # 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) +# TODO: Refactor to reduce the number of lines in this module later + def _check_bgp_neighbor_capability(capability_status: dict[str, bool]) -> bool: """Check if a BGP neighbor capability is advertised, received, and enabled. @@ -1797,3 +1799,97 @@ def test(self) -> None: # Verify BGP and RIB nexthops are same. if len(bgp_nexthops) != len(route_entry["vias"]): self.result.is_failure(f"{route} - Nexthops count mismatch - BGP: {len(bgp_nexthops)}, RIB: {len(route_entry['vias'])}") + + +class VerifyBGPRedistribution(AntaTest): + """Verifies BGP redistribution. + + This test performs the following checks for each specified VRF in the BGP instance: + + 1. Ensures that the expected address-family is configured on the device. + 2. Confirms that the redistributed route protocol, include leaked and route map match the expected values. + + + Expected Results + ---------------- + * Success: If all of the following conditions are met: + - The expected address-family is configured on the device. + - The redistributed route protocol, include leaked and route map align with the expected values for the route. + * Failure: If any of the following occur: + - The expected address-family is not configured on device. + - The redistributed route protocol, include leaked or route map does not match the expected values. + + Examples + -------- + ```yaml + anta.tests.routing: + bgp: + - VerifyBGPRedistribution: + vrfs: + - vrf: default + address_families: + - afi_safi: ipv4Unicast + redistributed_routes: + - proto: Connected + include_leaked: True + route_map: RM-CONN-2-BGP + - proto: Static + include_leaked: True + route_map: RM-CONN-2-BGP + - afi_safi: IPv6 Unicast + redistributed_routes: + - proto: User # Converted to EOS SDK + route_map: RM-CONN-2-BGP + - proto: Static + include_leaked: True + route_map: RM-CONN-2-BGP + ``` + """ + + categories: ClassVar[list[str]] = ["bgp"] + commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show bgp instance vrf all", revision=4)] + + class Input(AntaTest.Input): + """Input model for the VerifyBGPRedistribution test.""" + + vrfs: list[BgpVrf] + """List of VRFs in the BGP instance.""" + + def _validate_redistribute_route(self, vrf_data: str, addr_family: str, afi_safi_configs: list[dict[str, Any]], route_info: dict[str, Any]) -> list[Any]: + """Validate the redstributed route details for a given address family.""" + failure_msg = [] + # If the redistributed route protocol does not match the expected value, test fails. + if not (actual_route := get_item(afi_safi_configs.get("redistributedRoutes"), "proto", route_info.proto)): + failure_msg.append(f"{vrf_data}, {addr_family}, Proto: {route_info.proto} - Not configured") + return failure_msg + + # If includes leaked field applicable, and it does not matches the expected value, test fails. + if (act_include_leaked := actual_route.get("includeLeaked", False)) != route_info.include_leaked: + failure_msg.append(f"{vrf_data}, {addr_family}, {route_info} - Include leaked mismatch - Actual: {act_include_leaked}") + + # If route map is required and it is not matching the expected value, test fails. + if all([route_info.route_map, (act_route_map := actual_route.get("routeMap", "Not Found")) != route_info.route_map]): + failure_msg.append(f"{vrf_data}, {addr_family}, {route_info} - Route map mismatch - Actual: {act_route_map}") + return failure_msg + + @AntaTest.anta_test + def test(self) -> None: + """Main test function for VerifyBGPRedistribution.""" + self.result.is_success() + command_output = self.instance_commands[0].json_output + + for vrf_data in self.inputs.vrfs: + # If the specified VRF details are not found, test fails. + if not (instance_details := get_value(command_output, f"vrfs.{vrf_data.vrf}")): + self.result.is_failure(f"{vrf_data} - Not configured") + continue + for address_family in vrf_data.address_families: + # If the AFI-SAFI configuration details are not found, test fails. + if not (afi_safi_configs := get_value(instance_details, f"afiSafiConfig.{address_family.afi_safi}")): + self.result.is_failure(f"{vrf_data}, {address_family} - Not redistributed") + continue + + for route_info in address_family.redistributed_routes: + failure_msg = self._validate_redistribute_route(str(vrf_data), str(address_family), afi_safi_configs, route_info) + for msg in failure_msg: + self.result.is_failure(msg) diff --git a/examples/tests.yaml b/examples/tests.yaml index e4f08a937..f5fd3ebd0 100644 --- a/examples/tests.yaml +++ b/examples/tests.yaml @@ -513,6 +513,26 @@ anta.tests.routing.bgp: - VerifyBGPPeersHealthRibd: # Verifies the health of all the BGP IPv4 peer(s). check_tcp_queues: True + - VerifyBGPRedistribution: + # Verifies BGP redistribution. + vrfs: + - vrf: default + address_families: + - afi_safi: ipv4Unicast + redistributed_routes: + - proto: Connected + include_leaked: True + route_map: RM-CONN-2-BGP + - proto: Static + include_leaked: True + route_map: RM-CONN-2-BGP + - afi_safi: IPv6 Unicast + redistributed_routes: + - proto: User # Converted to EOS SDK + route_map: RM-CONN-2-BGP + - proto: Static + include_leaked: True + route_map: RM-CONN-2-BGP - VerifyBGPRouteECMP: # Verifies BGP IPv4 route ECMP paths. route_entries: diff --git a/tests/units/anta_tests/routing/test_bgp.py b/tests/units/anta_tests/routing/test_bgp.py index 418c2d0e1..ee26155f3 100644 --- a/tests/units/anta_tests/routing/test_bgp.py +++ b/tests/units/anta_tests/routing/test_bgp.py @@ -28,6 +28,7 @@ VerifyBGPPeersHealth, VerifyBGPPeersHealthRibd, VerifyBGPPeerUpdateErrors, + VerifyBGPRedistribution, VerifyBGPRouteECMP, VerifyBgpRouteMaps, VerifyBGPRoutePaths, @@ -5788,4 +5789,333 @@ def test_check_bgp_neighbor_capability(input_dict: dict[str, bool], expected: bo "inputs": {"route_entries": [{"prefix": "10.111.134.0/24", "vrf": "default", "ecmp_count": 2}]}, "expected": {"result": "failure", "messages": ["Prefix: 10.111.134.0/24 VRF: default - Nexthops count mismatch - BGP: 2, RIB: 1"]}, }, + { + "name": "success", + "test": VerifyBGPRedistribution, + "eos_data": [ + { + "vrfs": { + "default": { + "afiSafiConfig": { + "v4u": { + "redistributedRoutes": [ + {"proto": "Connected", "includeLeaked": True, "routeMap": "RM-CONN-2-BGP"}, + {"proto": "Static", "includeLeaked": True, "routeMap": "RM-CONN-2-BGP"}, + ] + }, + "v6m": { + "redistributedRoutes": [ + {"proto": "Dynamic", "routeMap": "RM-CONN-2-BGP"}, + {"proto": "IS-IS", "includeLeaked": True, "routeMap": "RM-CONN-2-BGP"}, + ] + }, + } + }, + "test": { + "afiSafiConfig": { + "v4u": { + "redistributedRoutes": [ + {"proto": "EOS SDK", "routeMap": "RM-CONN-2-BGP"}, + {"proto": "OSPF Internal", "includeLeaked": True, "routeMap": "RM-CONN-2-BGP"}, + ] + }, + "v6m": { + "redistributedRoutes": [ + {"proto": "RIP", "routeMap": "RM-CONN-2-BGP"}, + {"proto": "Bgp", "includeLeaked": True, "routeMap": "RM-CONN-2-BGP"}, + ] + }, + } + }, + } + } + ], + "inputs": { + "vrfs": [ + { + "vrf": "default", + "address_families": [ + { + "afi_safi": "ipv4Unicast", + "redistributed_routes": [ + {"proto": "Connected", "include_leaked": True, "route_map": "RM-CONN-2-BGP"}, + {"proto": "Static", "include_leaked": True, "route_map": "RM-CONN-2-BGP"}, + ], + }, + { + "afi_safi": "IPv6 multicast", + "redistributed_routes": [ + {"proto": "Dynamic", "route_map": "RM-CONN-2-BGP"}, + {"proto": "IS-IS", "include_leaked": True, "route_map": "RM-CONN-2-BGP"}, + ], + }, + ], + }, + { + "vrf": "test", + "address_families": [ + { + "afi_safi": "ipv4 Unicast", + "redistributed_routes": [ + {"proto": "User", "route_map": "RM-CONN-2-BGP"}, + {"proto": "OSPF Internal", "include_leaked": True, "route_map": "RM-CONN-2-BGP"}, + ], + }, + { + "afi_safi": "IPv6Multicast", + "redistributed_routes": [ + {"proto": "RIP", "route_map": "RM-CONN-2-BGP"}, + {"proto": "Bgp", "include_leaked": True, "route_map": "RM-CONN-2-BGP"}, + ], + }, + ], + }, + ] + }, + "expected": {"result": "success"}, + }, + { + "name": "failure-vrf-not-found", + "test": VerifyBGPRedistribution, + "eos_data": [ + { + "vrfs": { + "default": {"afiSafiConfig": {"v6m": {"redistributedRoutes": [{"proto": "Connected", "routeMap": "RM-CONN-2-BGP"}]}}}, + "tenant": {"afiSafiConfig": {"v4u": {"redistributedRoutes": [{"proto": "Connected"}]}}}, + } + } + ], + "inputs": { + "vrfs": [ + { + "vrf": "default", + "address_families": [ + { + "afi_safi": "ipv6 Multicast", + "redistributed_routes": [{"proto": "Connected", "include_leaked": False, "route_map": "RM-CONN-2-BGP"}], + }, + ], + }, + { + "vrf": "test", + "address_families": [ + { + "afi_safi": "ipv6 Multicast", + "redistributed_routes": [ + {"proto": "Connected", "include_leaked": True, "route_map": "RM-CONN-2-BGP"}, + ], + }, + ], + }, + ] + }, + "expected": {"result": "failure", "messages": ["VRF: test - Not configured"]}, + }, + { + "name": "failure-afi-safi-config-not-found", + "test": VerifyBGPRedistribution, + "eos_data": [ + { + "vrfs": { + "default": {"afiSafiConfig": {"v6m": {}}}, + "test": {"afiSafiConfig": {"v4u": {"redistributedRoutes": [{"proto": "Connected", "routeMap": "RM-CONN-2-BGP"}]}}}, + } + } + ], + "inputs": { + "vrfs": [ + { + "vrf": "default", + "address_families": [ + { + "afi_safi": "ipv6 Multicast", + "redistributed_routes": [ + {"proto": "Connected", "include_leaked": True, "route_map": "RM-CONN-2-BGP"}, + {"proto": "Static", "include_leaked": True, "route_map": "RM-CONN-2-BGP"}, + ], + }, + ], + }, + ] + }, + "expected": {"result": "failure", "messages": ["VRF: default, AFI-SAFI: IPv6 Multicast - Not redistributed"]}, + }, + { + "name": "failure-expected-proto-not-found", + "test": VerifyBGPRedistribution, + "eos_data": [ + { + "vrfs": { + "default": { + "afiSafiConfig": { + "v4m": {"redistributedRoutes": [{"proto": "RIP", "routeMap": "RM-CONN-2-BGP"}, {"proto": "IS-IS", "routeMap": "RM-MLAG-PEER-IN"}]} + } + }, + "test": { + "afiSafiConfig": { + "v6u": { + "redistributedRoutes": [{"proto": "Static", "routeMap": "RM-CONN-2-BGP"}], + } + } + }, + } + } + ], + "inputs": { + "vrfs": [ + { + "vrf": "default", + "address_families": [ + { + "afi_safi": "ipv4 multicast", + "redistributed_routes": [ + {"proto": "OSPFv3 External", "include_leaked": True, "route_map": "RM-CONN-2-BGP"}, + {"proto": "OSPFv3 Nssa-External", "include_leaked": True, "route_map": "RM-CONN-2-BGP"}, + ], + } + ], + }, + { + "vrf": "test", + "address_families": [ + { + "afi_safi": "IPv6Unicast", + "redistributed_routes": [ + {"proto": "RIP", "route_map": "RM-CONN-2-BGP"}, + {"proto": "Bgp", "include_leaked": True, "route_map": "RM-CONN-2-BGP"}, + ], + }, + ], + }, + ] + }, + "expected": { + "result": "failure", + "messages": [ + "VRF: default, AFI-SAFI: IPv4 Multicast, Proto: OSPFv3 External - Not configured", + "VRF: default, AFI-SAFI: IPv4 Multicast, Proto: OSPFv3 Nssa-External - Not configured", + "VRF: test, AFI-SAFI: IPv6 Unicast, Proto: RIP - Not configured", + "VRF: test, AFI-SAFI: IPv6 Unicast, Proto: Bgp - Not configured", + ], + }, + }, + { + "name": "failure-route-map-not-found", + "test": VerifyBGPRedistribution, + "eos_data": [ + { + "vrfs": { + "default": {"afiSafiConfig": {"v4u": {"redistributedRoutes": [{"proto": "Connected", "routeMap": "RM-CONN-10-BGP"}, {"proto": "Static"}]}}}, + "test": { + "afiSafiConfig": { + "v6u": { + "redistributedRoutes": [{"proto": "EOS SDK", "routeMap": "RM-MLAG-PEER-IN"}, {"proto": "OSPF Internal"}], + } + } + }, + } + } + ], + "inputs": { + "vrfs": [ + { + "vrf": "default", + "address_families": [ + { + "afi_safi": "ipv4 UNicast", + "redistributed_routes": [ + {"proto": "Connected", "route_map": "RM-CONN-2-BGP"}, + {"proto": "Static", "route_map": "RM-CONN-2-BGP"}, + ], + }, + ], + }, + { + "vrf": "test", + "address_families": [ + { + "afi_safi": "ipv6-Unicast", + "redistributed_routes": [ + {"proto": "User", "route_map": "RM-CONN-2-BGP"}, + {"proto": "OSPF Internal", "route_map": "RM-CONN-2-BGP"}, + ], + }, + ], + }, + ] + }, + "expected": { + "result": "failure", + "messages": [ + "VRF: default, AFI-SAFI: IPv4 Unicast, Proto: Connected, Route Map: RM-CONN-2-BGP - Route map mismatch - Actual: RM-CONN-10-BGP", + "VRF: default, AFI-SAFI: IPv4 Unicast, Proto: Static, Route Map: RM-CONN-2-BGP - Route map mismatch - Actual: Not Found", + "VRF: test, AFI-SAFI: IPv6 Unicast, Proto: EOS SDK, Route Map: RM-CONN-2-BGP - Route map mismatch - Actual: RM-MLAG-PEER-IN", + "VRF: test, AFI-SAFI: IPv6 Unicast, Proto: OSPF Internal, Route Map: RM-CONN-2-BGP - Route map mismatch - Actual: Not Found", + ], + }, + }, + { + "name": "failure-incorrect-value-include-leaked", + "test": VerifyBGPRedistribution, + "eos_data": [ + { + "vrfs": { + "default": { + "afiSafiConfig": { + "v4m": { + "redistributedRoutes": [ + {"proto": "Dynamic", "routeMap": "RM-CONN-2-BGP"}, + {"proto": "IS-IS", "includeLeaked": False, "routeMap": "RM-CONN-2-BGP"}, + ] + }, + } + }, + "test": { + "afiSafiConfig": { + "v6u": { + "redistributedRoutes": [ + {"proto": "RIP", "routeMap": "RM-CONN-2-BGP"}, + {"proto": "Bgp", "includeLeaked": True, "routeMap": "RM-CONN-2-BGP"}, + ] + }, + } + }, + } + } + ], + "inputs": { + "vrfs": [ + { + "vrf": "default", + "address_families": [ + { + "afi_safi": "ipv4-multicast", + "redistributed_routes": [ + {"proto": "IS-IS", "include_leaked": True, "route_map": "RM-CONN-2-BGP"}, + ], + }, + ], + }, + { + "vrf": "test", + "address_families": [ + { + "afi_safi": "IPv6_unicast", + "redistributed_routes": [ + {"proto": "RIP", "route_map": "RM-CONN-2-BGP"}, + {"proto": "Bgp", "include_leaked": False, "route_map": "RM-CONN-2-BGP"}, + ], + }, + ], + }, + ] + }, + "expected": { + "result": "failure", + "messages": [ + "VRF: default, AFI-SAFI: IPv4 Multicast, Proto: IS-IS, Include Leaked: True, Route Map: RM-CONN-2-BGP - Include leaked mismatch - Actual: False", + "VRF: test, AFI-SAFI: IPv6 Unicast, Proto: Bgp, Route Map: RM-CONN-2-BGP - Include leaked mismatch - Actual: True", + ], + }, + }, ] diff --git a/tests/units/input_models/routing/test_bgp.py b/tests/units/input_models/routing/test_bgp.py index d8d64507f..0ff859ff6 100644 --- a/tests/units/input_models/routing/test_bgp.py +++ b/tests/units/input_models/routing/test_bgp.py @@ -6,12 +6,12 @@ # pylint: disable=C0302 from __future__ import annotations -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any import pytest from pydantic import ValidationError -from anta.input_models.routing.bgp import BgpAddressFamily, BgpPeer, BgpRoute +from anta.input_models.routing.bgp import AddressFamilyConfig, BgpAddressFamily, BgpPeer, BgpRoute, RedistributedRouteConfig from anta.tests.routing.bgp import ( VerifyBGPExchangedRoutes, VerifyBGPNlriAcceptance, @@ -27,7 +27,7 @@ ) if TYPE_CHECKING: - from anta.custom_types import Afi, Safi + from anta.custom_types import Afi, RedistributedAfiSafi, RedistributedProtocol, Safi class TestBgpAddressFamily: @@ -348,3 +348,90 @@ def test_invalid(self, route_entries: list[BgpRoute]) -> None: """Test VerifyBGPRoutePaths.Input invalid inputs.""" with pytest.raises(ValidationError): VerifyBGPRoutePaths.Input(route_entries=route_entries) + + +class TestVerifyBGPRedistributedRoute: + """Test anta.input_models.routing.bgp.RedistributedRouteConfig.""" + + @pytest.mark.parametrize( + ("proto", "include_leaked"), + [ + pytest.param("Connected", True, id="proto-valid"), + pytest.param("Static", False, id="proto-valid-leaked-false"), + pytest.param("User", False, id="proto-User"), + ], + ) + def test_validate_inputs(self, proto: RedistributedProtocol, include_leaked: bool) -> None: + """Test RedistributedRouteConfig valid inputs.""" + RedistributedRouteConfig(proto=proto, include_leaked=include_leaked) + + @pytest.mark.parametrize( + ("proto", "include_leaked"), + [ + pytest.param("Dynamic", True, id="proto-valid"), + pytest.param("User", True, id="proto-valid-leaked-false"), + ], + ) + def test_invalid(self, proto: RedistributedProtocol, include_leaked: bool) -> None: + """Test RedistributedRouteConfig invalid inputs.""" + with pytest.raises(ValidationError): + RedistributedRouteConfig(proto=proto, include_leaked=include_leaked) + + @pytest.mark.parametrize( + ("proto", "include_leaked", "route_map", "expected"), + [ + pytest.param("Connected", True, "RM-CONN-2-BGP", "Proto: Connected, Include Leaked: True, Route Map: RM-CONN-2-BGP", id="check-all-params"), + pytest.param("Static", False, None, "Proto: Static", id="check-proto-include_leaked-false"), + pytest.param("User", False, "RM-CONN-2-BGP", "Proto: EOS SDK, Route Map: RM-CONN-2-BGP", id="check-proto-route_map"), + pytest.param("Dynamic", False, None, "Proto: Dynamic", id="check-proto-only"), + ], + ) + def test_valid_str(self, proto: RedistributedProtocol, include_leaked: bool, route_map: str | None, expected: str) -> None: + """Test RedistributedRouteConfig __str__.""" + assert str(RedistributedRouteConfig(proto=proto, include_leaked=include_leaked, route_map=route_map)) == expected + + +class TestVerifyBGPAddressFamilyConfig: + """Test anta.input_models.routing.bgp.AddressFamilyConfig.""" + + @pytest.mark.parametrize( + ("afi_safi", "redistributed_routes"), + [ + pytest.param("ipv4Unicast", [{"proto": "OSPFv3 External", "include_leaked": True, "route_map": "RM-CONN-2-BGP"}], id="afisafi-ipv4-unicast"), + pytest.param("ipv6 Multicast", [{"proto": "OSPF Internal", "include_leaked": True, "route_map": "RM-CONN-2-BGP"}], id="afisafi-ipv6-multicast"), + pytest.param("ipv4-Multicast", [{"proto": "IS-IS", "include_leaked": False, "route_map": "RM-CONN-2-BGP"}], id="afisafi-ipv4-multicast"), + pytest.param("ipv6_Unicast", [{"proto": "AttachedHost", "route_map": "RM-CONN-2-BGP"}], id="afisafi-ipv6-unicast"), + ], + ) + def test_valid(self, afi_safi: RedistributedAfiSafi, redistributed_routes: list[Any]) -> None: + """Test AddressFamilyConfig valid inputs.""" + AddressFamilyConfig(afi_safi=afi_safi, redistributed_routes=redistributed_routes) + + @pytest.mark.parametrize( + ("afi_safi", "redistributed_routes"), + [ + pytest.param("evpn", [{"proto": "OSPFv3 Nssa-External", "include_leaked": True, "route_map": "RM-CONN-2-BGP"}], id="invalid-address-family"), + pytest.param("ipv6 sr-te", [{"proto": "RIP", "route_map": "RM-CONN-2-BGP"}], id="ipv6-invalid-address-family"), + pytest.param("iipv6_Unicast", [{"proto": "Bgp", "include_leaked": True, "route_map": "RM-CONN-2-BGP"}], id="ipv6-unicast-invalid-address-family"), + pytest.param("ipv6_Unicastt", [{"proto": "Static", "include_leaked": True, "route_map": "RM-CONN-2-BGP"}], id="ipv6-unicast-invalid-address-family"), + ], + ) + def test_invalid(self, afi_safi: RedistributedAfiSafi, redistributed_routes: list[Any]) -> None: + """Test AddressFamilyConfig invalid inputs.""" + with pytest.raises(ValidationError): + AddressFamilyConfig(afi_safi=afi_safi, redistributed_routes=redistributed_routes) + + @pytest.mark.parametrize( + ("afi_safi", "redistributed_routes", "expected"), + [ + pytest.param( + "v4u", [{"proto": "OSPFv3 Nssa-External", "include_leaked": True, "route_map": "RM-CONN-2-BGP"}], "AFI-SAFI: IPv4 Unicast", id="valid-ipv4-unicast" + ), + pytest.param("v4m", [{"proto": "RIP", "route_map": "RM-CONN-2-BGP"}], "AFI-SAFI: IPv4 Multicast", id="valid-ipv4-multicast"), + pytest.param("v6u", [{"proto": "Bgp", "include_leaked": True, "route_map": "RM-CONN-2-BGP"}], "AFI-SAFI: IPv6 Unicast", id="valid-ipv6-unicast"), + pytest.param("v6m", [{"proto": "Static", "include_leaked": True, "route_map": "RM-CONN-2-BGP"}], "AFI-SAFI: IPv6 Multicast", id="valid-ipv6-multicast"), + ], + ) + def test_valid_str(self, afi_safi: RedistributedAfiSafi, redistributed_routes: list[Any], expected: str) -> None: + """Test AddressFamilyConfig __str__.""" + assert str(AddressFamilyConfig(afi_safi=afi_safi, redistributed_routes=redistributed_routes)) == expected