feat(anta.tests): Adding the test case to verify BGP ECMP paths (#507)

* refactor ECMP test

* Issue05: Updated test logic, added unit test for test and input model

* Issue-505: Added unit tests for VerifyBGPRoutePaths input model

* Issue_505: Updated unit test and test logic to support direct connected route

* Issue-505: Updated test with handling directly connected nexthop with active ecmp head

* Issue-505: Updated unit test and test logic

* Issue-505, update test verification route command and logic for nexthop verification

---------

Co-authored-by: Carl Baillargeon <[email protected]>
Co-authored-by: Guillaume Mulocher <[email protected]>
Co-authored-by: Geetanjali.mane <[email protected]>

anta/input_models/routing/bgp.py

@@ -224,8 +224,10 @@ class BgpRoute(BaseModel):
     """The IPv4 network address."""
     vrf: str = "default"
     """Optional VRF for the BGP peer. Defaults to `default`."""
-    paths: list[BgpRoutePath]
-    """A list of paths for the BGP route."""
+    paths: list[BgpRoutePath] | None = None
+    """A list of paths for the BGP route. Required field in the `VerifyBGPRoutePaths` test."""
+    ecmp_count: int | None = None
+    """The expected number of ECMP paths for the BGP route. Required field in the `VerifyBGPRouteECMP` test."""
 
     def __str__(self) -> str:
         """Return a human-readable string representation of the BgpRoute for reporting.

anta/tests/routing/bgp.py

@@ -3,7 +3,7 @@
 # that can be found in the LICENSE file.
 """Module related to BGP tests."""
 
-# Mypy does not understand AntaTest.Input typing
+# pylint: disable=too-many-lines
 # mypy: disable-error-code=attr-defined
 from __future__ import annotations
 
@@ -1666,6 +1666,16 @@ class Input(AntaTest.Input):
         route_entries: list[BgpRoute]
         """List of BGP IPv4 route(s)."""
 
+        @field_validator("route_entries")
+        @classmethod
+        def validate_route_entries(cls, route_entries: list[BgpRoute]) -> list[BgpRoute]:
+            """Validate that 'paths' field is provided in each BGP route."""
+            for route in route_entries:
+                if route.paths is None:
+                    msg = f"{route} 'paths' field missing in the input"
+                    raise ValueError(msg)
+            return route_entries
+
     @AntaTest.anta_test
     def test(self) -> None:
         """Main test function for VerifyBGPRoutePaths."""
@@ -1687,3 +1697,94 @@ def test(self) -> None:
 
                 if (actual_origin := get_value(route_path, "routeType.origin")) != origin:
                     self.result.is_failure(f"{route} {path} - Origin mismatch - Actual: {actual_origin}")
+
+
+class VerifyBGPRouteECMP(AntaTest):
+    """Verifies BGP IPv4 route ECMP paths.
+
+    This test performs the following checks for each specified BGP route entry:
+
+      1. Route exists in BGP table.
+      2. First path is a valid and active ECMP head.
+      3. Correct number of valid ECMP contributors follow the head path.
+      4. Route is installed in RIB with same amount of next-hops.
+
+    Expected Results
+    ----------------
+    * Success: The test will pass if all specified routes exist in both BGP and RIB tables with correct amount of ECMP paths.
+    * Failure: The test will fail if:
+        - A specified route is not found in BGP table.
+        - A valid and active ECMP head is not found.
+        - ECMP contributors count does not match the expected value.
+        - Route is not installed in RIB table.
+        - BGP and RIB nexthops count do not match.
+
+    Examples
+    --------
+    ```yaml
+    anta.tests.routing:
+      bgp:
+        - VerifyBGPRouteECMP:
+            route_entries:
+                - prefix: 10.100.0.128/31
+                  vrf: default
+                  ecmp_count: 2
+    ```
+    """
+
+    categories: ClassVar[list[str]] = ["bgp"]
+    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [
+        AntaCommand(command="show ip bgp vrf all", revision=3),
+        AntaCommand(command="show ip route vrf all bgp", revision=4),
+    ]
+
+    class Input(AntaTest.Input):
+        """Input model for the VerifyBGPRouteECMP test."""
+
+        route_entries: list[BgpRoute]
+        """List of BGP IPv4 route(s)."""
+
+        @field_validator("route_entries")
+        @classmethod
+        def validate_route_entries(cls, route_entries: list[BgpRoute]) -> list[BgpRoute]:
+            """Validate that 'ecmp_count' field is provided in each BGP route."""
+            for route in route_entries:
+                if route.ecmp_count is None:
+                    msg = f"{route} 'ecmp_count' field missing in the input"
+                    raise ValueError(msg)
+            return route_entries
+
+    @AntaTest.anta_test
+    def test(self) -> None:
+        """Main test function for VerifyBGPRouteECMP."""
+        self.result.is_success()
+
+        for route in self.inputs.route_entries:
+            # Verify if the prefix exists in BGP table.
+            if not (bgp_route_entry := get_value(self.instance_commands[0].json_output, f"vrfs..{route.vrf}..bgpRouteEntries..{route.prefix}", separator="..")):
+                self.result.is_failure(f"{route} - prefix not found in BGP table")
+                continue
+
+            route_paths = iter(bgp_route_entry["bgpRoutePaths"])
+            head = next(route_paths, None)
+            # Verify if the active ECMP head exists.
+            if head is None or not all(head["routeType"][key] for key in ["valid", "active", "ecmpHead"]):
+                self.result.is_failure(f"{route} - valid and active ECMP head not found")
+                continue
+
+            bgp_nexthops = {head["nextHop"]}
+            bgp_nexthops.update([path["nextHop"] for path in route_paths if all(path["routeType"][key] for key in ["valid", "ecmp", "ecmpContributor"])])
+
+            # Verify ECMP count is correct.
+            if len(bgp_nexthops) != route.ecmp_count:
+                self.result.is_failure(f"{route} - ECMP count mismatch - Expected: {route.ecmp_count}, Actual: {len(bgp_nexthops)}")
+                continue
+
+            # Verify if the prefix exists in routing table.
+            if not (route_entry := get_value(self.instance_commands[1].json_output, f"vrfs..{route.vrf}..routes..{route.prefix}", separator="..")):
+                self.result.is_failure(f"{route} - prefix not found in routing table")
+                continue
+
+            # 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'])}")

examples/tests.yaml

@@ -515,6 +515,12 @@ anta.tests.routing.bgp:
   - VerifyBGPPeersHealthRibd:
       # Verifies the health of all the BGP IPv4 peer(s).
       check_tcp_queues: True
+  - VerifyBGPRouteECMP:
+      # Verifies BGP IPv4 route ECMP paths.
+      route_entries:
+          - prefix: 10.100.0.128/31
+            vrf: default
+            ecmp_count: 2
   - VerifyBGPRoutePaths:
       # Verifies BGP IPv4 route paths.
       route_entries: