feat(anta): Added testcase to verify the BGP Redistributed Routes

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"\bipv4[\s]?unicast\b"
+REGEX_IPV4_MULTICAST = r"\bipv4[\s]?multicast\b"
+REGEX_IPV6_UNICAST = r"\bipv6[\s]?unicast\b"
+REGEX_IPV6_MULTICAST = r"\bipv6[\s]?multicast\b"
+
+
 def aaa_group_prefix(v: str) -> str:
     """Prefix the AAA method with 'group' if it is known."""
     built_in_methods = ["local", "none", "logging"]
@@ -132,6 +139,40 @@ def validate_regex(value: str) -> str:
     return value
 
 
+def bgp_redistributed_route_proto_abbreviations(value: str) -> str:
+    """Abbreviations for different BGP redistributed route protocols.
+
+    Examples
+    --------
+    - IPv4 Unicast
+    - ipv4Unicast
+    - IPv4 Multicast
+    - ipv4Multicast
+
+    """
+    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.search(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
+    --------
+    - User
+    """
+    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 +360,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),
+]
+RedisrbutedAfiSafi = Annotated[str, BeforeValidator(bgp_redistributed_route_proto_abbreviations)]

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, RedisrbutedAfiSafi, RedistributedProtocol, Safi, Vni
 
 if TYPE_CHECKING:
     import sys
@@ -68,8 +68,10 @@ 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."""
+
+    route_map: str | None = None
+    """Specify redistributed route protocol route map. Required field in the `VerifyBGPRedistribution` test."""
 
     @model_validator(mode="after")
     def validate_inputs(self) -> Self:
@@ -256,3 +258,83 @@ 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 BGP vrfs."""
+
+    vrf: str = "default"
+    """VRF for the BGP instance. Defaults to `default`."""
+    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 RedistributedRoute(BaseModel):
+    """Model representing BGP redistributed route."""
+
+    proto: RedistributedProtocol
+    """The redistributed route protocol."""
+    include_leaked: bool | None = None
+    """Flag to include leaked BGP routes in the advertisement"""
+    route_map: str | None = None
+    """The route map of the redistributed routes."""
+
+    @model_validator(mode="after")
+    def validate_include_leaked_support(self) -> Self:
+        """Validate the input provided for included leaked field, included _leaked this field is not supported for proto AttachedHost, User, Dynamic, RIP."""
+        if self.include_leaked and self.proto in ["AttachedHost", "EOS SDK", "Dynamic", "RIP"]:
+            msg = f"{self.include_leaked}, field is not supported for redistributed route protocol `{self.proto}`"
+            raise ValueError(msg)
+        return self
+
+    def __str__(self) -> str:
+        """Return a human-readable string representation of the RedistributedRoute for reporting.
+
+        Examples
+        --------
+        - Proto: Connected, Included Leaked: False, Route Map: RM-CONN-2-BGP
+        """
+        base_string = f"Proto: {self.proto}"
+        if self.include_leaked is not None:
+            base_string += f", Included Leaked: {self.include_leaked}"
+        if self.route_map is not None:
+            base_string += f", Route Map: {self.route_map}"
+        return base_string
+
+
+class AddressFamilyConfig(BaseModel):
+    """Model representing BGP address family configs."""
+
+    afi_safi: RedisrbutedAfiSafi
+    """BGP redistributed route supported address families"""
+    redistributed_routes: list[RedistributedRoute]
+    """A list of redistributed route"""
+
+    @model_validator(mode="after")
+    def validate_inputs(self) -> Self:
+        """Validate the inputs provided to the AddressFamilyConfig class.
+
+        address families must be `ipv4` or `ipv6` only, and sub address families can be `unicast` or `multicast`.
+        """
+        if self.afi_safi not in ["v4u", "v4m", "v6u", "v6m"]:
+            msg = f"Redistributed route protocol is not supported for address family `{self.afi_safi}`"
+            raise ValueError(msg)
+        return self
+
+    def __str__(self) -> str:
+        """Return a human-readable string representation of the AddressFamilyConfig for reporting.
+
+        Examples
+        --------
+        - AFI-SAFI: v4u
+        """
+        return f"AFI-SAFI: {self.afi_safi}"

anta/tests/routing/bgp.py

@@ -11,13 +11,16 @@
 
 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)
 
+# pylint: disable=C0302
+# 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 +1800,89 @@ 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 redistributed routes protocol and route-map.
+
+    This test performs the following checks for each specified route:
+
+      1. Ensures that the expected address-family is configured on the device.
+      2. Confirms that the redistributed route protocol, included leaked and route map match the expected values for a route.
+
+    Note: For "User" proto field, checking that it's "EOS SDK" versus User.
+
+    Expected Results
+    ----------------
+    * Success: If all of the following conditions are met:
+        - The expected address-family is configured on the device.
+        - The redistributed route protocol, included 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 protocolor, included leaked or route map does not match the expected value for a route.
+
+    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: Dynamic
+                        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 address families."""
+
+    def _validate_redistribute_route_details(self, vrf_data: str, address_family: dict[str, Any], afi_safi_configs: list[dict[str, Any]]) -> None:
+        """Validate the redstributed route details for a given address family."""
+        for route_info in address_family.redistributed_routes:
+            # 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)):
+                self.result.is_failure(f"{vrf_data}, {address_family}, {route_info} - Not configured")
+                continue
+
+            # If includes leaked field applicable, and it does not matches the expected value, test fails.
+            if all([route_info.include_leaked is not None, (act_include_leaked := actual_route.get("includeLeaked", "Not Found")) != route_info.include_leaked]):
+                self.result.is_failure(f"{vrf_data}, {address_family}, {route_info} - Value for included 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]):
+                self.result.is_failure(f"{vrf_data}, {address_family}, {route_info} - Route map mismatch - Actual: {act_route_map}")
+
+    @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:
+            for address_family in vrf_data.address_families:
+                # If the specified VRF, AFI-SAFI details are not found, test fails.
+                if not (afi_safi_configs := get_value(command_output, f"vrfs.{vrf_data.vrf}.afiSafiConfig.{address_family.afi_safi}")):
+                    self.result.is_failure(f"{vrf_data}, {address_family} - Not configured")
+                    continue
+                self._validate_redistribute_route_details(str(vrf_data), address_family, afi_safi_configs)

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 redistributed routes protocol and route-map.
+      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: Dynamic
+                  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: