tests : bgp-default-originate ecmp scenerio is added

Signed-off-by: ARShreenidhi <rshreenidhi@vmware.com>

1 files changed, 396 insertions, 0 deletions

diff --git a/tests/topotests/bgp_default_originate/test_bgp_default_originate_2links.py b/tests/topotests/bgp_default_originate/test_bgp_default_originate_2links.py
index c8cdc7ec5..4d7f436ea 100644
--- a/tests/topotests/bgp_default_originate/test_bgp_default_originate_2links.py
+++ b/tests/topotests/bgp_default_originate/test_bgp_default_originate_2links.py
@@ -22,6 +22,7 @@
 Following tests are covered.
 1. Verify default-originate route with default static and network command
 2. Verify default-originate route with aggregate summary command
+3. Verfiy  default-originate behaviour in ecmp
 """
 import os
 import sys
@@ -48,7 +49,10 @@ from lib.bgp import (
 from lib.common_config import (
     verify_fib_routes,
     step,
+    create_prefix_lists,
     run_frr_cmd,
+    create_route_maps,
+    shutdown_bringup_interface,
     get_frr_ipv6_linklocal,
     start_topology,
     apply_raw_config,
@@ -296,6 +300,78 @@ def verify_the_uptime(time_stamp_before, time_stamp_after, incremented=None):
         return True
 
 
+def get_best_path_route_in_FIB(tgen, topo, dut, network):
+    """
+    API to verify the best route in FIB and return the ipv4 and ipv6 nexthop for the given route
+    command
+    =======
+    show ip route
+    show ipv6 route
+    params
+    ======
+    dut : device under test :
+    network ; route (ip) to which the best route to be retrieved
+    Returns
+    ========
+    on success : return dict with next hops for the best hop
+    on failure : return error message with boolean False
+    """
+    is_ipv4_best_path_found = False
+    is_ipv6_best_path_found = False
+    rnode = tgen.routers()[dut]
+    ipv4_show_bgp_json = run_frr_cmd(rnode, "sh ip bgp  json ", isjson=True)
+    ipv6_show_bgp_json = run_frr_cmd(
+        rnode, "sh ip bgp ipv6 unicast  json ", isjson=True
+    )
+    output_dict = {"ipv4": None, "ipv6": None}
+    ipv4_nxt_hop_count = len(ipv4_show_bgp_json["routes"][network["ipv4"]])
+    for index in range(ipv4_nxt_hop_count):
+        if "bestpath" in ipv4_show_bgp_json["routes"][network["ipv4"]][index].keys():
+            best_path_ip = ipv4_show_bgp_json["routes"][network["ipv4"]][index][
+                "nexthops"
+            ][0]["ip"]
+            output_dict["ipv4"] = best_path_ip
+            logger.info(
+                "[DUT [{}]] Best path for the route {} is  {} ".format(
+                    dut, network["ipv4"], best_path_ip
+                )
+            )
+            is_ipv4_best_path_found = True
+        else:
+            logger.error("ERROR....! No Best Path  Found in BGP RIB.... FAILED")
+
+    ipv6_nxt_hop_count = len(ipv6_show_bgp_json["routes"][network["ipv6"]])
+    for index in range(ipv6_nxt_hop_count):
+        if "bestpath" in ipv6_show_bgp_json["routes"][network["ipv6"]][index].keys():
+            ip_add_count = len(
+                ipv6_show_bgp_json["routes"][network["ipv6"]][index]["nexthops"]
+            )
+            for i_index in range(ip_add_count):
+                if (
+                    "global"
+                    in ipv6_show_bgp_json["routes"][network["ipv6"]][index]["nexthops"][
+                        i_index
+                    ]["scope"]
+                ):
+                    best_path_ip = ipv6_show_bgp_json["routes"][network["ipv6"]][index][
+                        "nexthops"
+                    ][i_index]["ip"]
+                    output_dict["ipv6"] = best_path_ip
+                    logger.info(
+                        "[DUT [{}]] Best path for the route {} is  {} ".format(
+                            dut, network["ipv6"], best_path_ip
+                        )
+                    )
+
+        else:
+            logger.error("ERROR....! No Best Path  Found in BGP RIB.... FAILED")
+    if is_ipv4_best_path_found:
+        return output_dict
+    else:
+        logger.error("ERROR...! Unable to find the Best Path in the RIB")
+        return False
+
+
 #####################################################
 #
 #                      Testcases
@@ -1409,6 +1485,326 @@ def test_verify_bgp_default_originate_with_aggregate_summary_p1(request):
     write_test_footer(tc_name)
 
 
+def test_verify_default_originate_with_2way_ecmp_p2(request):
+    """
+    Summary: "Verify default-originate route with 3 way ECMP and traffic "
+    """
+
+    tgen = get_topogen()
+    global BGP_CONVERGENCE
+    global DEFAULT_ROUTES
+    DEFAULT_ROUTES = {"ipv4": "0.0.0.0/0", "ipv6": "0::0/0"}
+
+    if BGP_CONVERGENCE != True:
+        pytest.skip("skipped because of BGP Convergence failure")
+    # test case name
+    tc_name = request.node.name
+    write_test_header(tc_name)
+    if tgen.routers_have_failure():
+        check_router_status(tgen)
+    reset_config_on_routers(tgen)
+
+    step("Populating next-hops details")
+    r1_r2_ipv4_neighbor_ips = []
+    r1_r2_ipv6_neighbor_ips = []
+    r1_link = None
+    for index in range(1, 3):
+        r1_link = "r1-link" + str(index)
+        r1_r2_ipv4_neighbor_ips.append(
+            topo["routers"]["r2"]["links"][r1_link]["ipv4"].split("/")[0]
+        )
+        r1_r2_ipv6_neighbor_ips.append(
+            topo["routers"]["r2"]["links"][r1_link]["ipv6"].split("/")[0]
+        )
+
+    step(
+        "Configure default-originate on R1 for all the neighbor of IPv4 and IPv6 peers "
+    )
+    local_as = get_dut_as_number(tgen, dut="r1")
+    for index in range(2):
+        raw_config = {
+            "r1": {
+                "raw_config": [
+                    "router bgp {}".format(local_as),
+                    "address-family ipv4 unicast",
+                    "neighbor {} default-originate".format(
+                        r1_r2_ipv4_neighbor_ips[index]
+                    ),
+                    "exit-address-family",
+                    "address-family ipv6 unicast",
+                    "neighbor {} default-originate ".format(
+                        r1_r2_ipv6_neighbor_ips[index]
+                    ),
+                    "exit-address-family",
+                ]
+            }
+        }
+        result = apply_raw_config(tgen, raw_config)
+        assert result is True, "Testcase {} : Failed Error: {}".format(tc_name, result)
+
+    step(
+        "After configuring default-originate command , verify default  routes are advertised on R2 "
+    )
+
+    r2_link = None
+    for index in range(1, 3):
+        r2_link = "r2-link" + str(index)
+        ipv4_nxt_hop = topo["routers"]["r1"]["links"][r2_link]["ipv4"].split("/")[0]
+        interface = topo["routers"]["r1"]["links"][r2_link]["interface"]
+        ipv6_link_local_nxt_hop = get_frr_ipv6_linklocal(tgen, "r1", intf=interface)
+        DEFAULT_ROUTE_NXT_HOP = {"ipv4": ipv4_nxt_hop, "ipv6": ipv6_link_local_nxt_hop}
+
+        result = verify_rib_default_route(
+            tgen,
+            topo,
+            dut="r2",
+            routes=DEFAULT_ROUTES,
+            expected_nexthop=DEFAULT_ROUTE_NXT_HOP,
+        )
+        assert result is True, "Testcase {} : Failed \n Error: {}".format(
+            tc_name, result
+        )
+
+    step("Ping R1 configure IPv4 and IPv6 loopback address from R2")
+    pingaddr = topo["routers"]["r1"]["links"]["lo"]["ipv4"].split("/")[0]
+    router = tgen.gears["r2"]
+    output = router.run("ping -c 4 -w 4 {}".format(pingaddr))
+    assert " 0% packet loss" in output, "Ping R1->R2  FAILED"
+    logger.info("Ping from R1 to R2 ... success")
+
+    step("Shuting up the active route")
+    network = {"ipv4": "0.0.0.0/0", "ipv6": "::/0"}
+    ipv_dict = get_best_path_route_in_FIB(tgen, topo, dut="r2", network=network)
+    dut_links = topo["routers"]["r1"]["links"]
+    active_interface = None
+    for key, values in dut_links.items():
+        ipv4_address = dut_links[key]["ipv4"].split("/")[0]
+        ipv6_address = dut_links[key]["ipv6"].split("/")[0]
+        if ipv_dict["ipv4"] == ipv4_address and ipv_dict["ipv6"] == ipv6_address:
+            active_interface = dut_links[key]["interface"]
+
+    logger.info(
+        "Shutting down the interface {} on router {} ".format(active_interface, "r1")
+    )
+    shutdown_bringup_interface(tgen, "r1", active_interface, False)
+
+    step("Verify the complete convergence  to fail after  shutting  the interface")
+    result = verify_bgp_convergence(tgen, topo, expected=False)
+    assert (
+        result is not True
+    ), " Testcase {} : After shuting down the interface  Convergence is expected to be Failed".format(
+        tc_name
+    )
+
+    step(
+        "Verify  routes from active best path is not received from  r1 after  shuting the interface"
+    )
+    r2_link = None
+    for index in range(1, 3):
+        r2_link = "r2-link" + str(index)
+        ipv4_nxt_hop = topo["routers"]["r1"]["links"][r2_link]["ipv4"].split("/")[0]
+        interface = topo["routers"]["r1"]["links"][r2_link]["interface"]
+        ipv6_link_local_nxt_hop = get_frr_ipv6_linklocal(tgen, "r1", intf=interface)
+        DEFAULT_ROUTE_NXT_HOP = {"ipv4": ipv4_nxt_hop, "ipv6": ipv6_link_local_nxt_hop}
+        if index == 1:
+            result = verify_rib_default_route(
+                tgen,
+                topo,
+                dut="r2",
+                routes=DEFAULT_ROUTES,
+                expected_nexthop=DEFAULT_ROUTE_NXT_HOP,
+                expected=False,
+            )
+            assert result is not True, "Testcase {} : Failed \n Error: {}".format(
+                tc_name, result
+            )
+        else:
+            result = verify_rib_default_route(
+                tgen,
+                topo,
+                dut="r2",
+                routes=DEFAULT_ROUTES,
+                expected_nexthop=DEFAULT_ROUTE_NXT_HOP,
+            )
+            assert result is True, "Testcase {} : Failed \n Error: {}".format(
+                tc_name, result
+            )
+
+    step("Ping R1 configure IPv4 and IPv6 loopback address from R2")
+    pingaddr = topo["routers"]["r1"]["links"]["lo"]["ipv4"].split("/")[0]
+    router = tgen.gears["r2"]
+    output = router.run("ping -c 4 -w 4 {}".format(pingaddr))
+    assert " 0% packet loss" in output, "Ping R1->R2  FAILED"
+    logger.info("Ping from R1 to R2 ... success")
+
+    step("No Shuting up the active route")
+
+    shutdown_bringup_interface(tgen, "r1", active_interface, True)
+
+    step("Verify the complete convergence after bringup the interface")
+    result = verify_bgp_convergence(tgen, topo)
+    assert (
+        result is True
+    ), " Testcase {} : After bringing up  the interface  complete convergence is expected ".format(
+        tc_name
+    )
+
+    step("Verify all the routes are received from  r1 after no shuting the interface")
+    r2_link = None
+    for index in range(1, 3):
+        r2_link = "r2-link" + str(index)
+        ipv4_nxt_hop = topo["routers"]["r1"]["links"][r2_link]["ipv4"].split("/")[0]
+        interface = topo["routers"]["r1"]["links"][r2_link]["interface"]
+        ipv6_link_local_nxt_hop = get_frr_ipv6_linklocal(tgen, "r1", intf=interface)
+        DEFAULT_ROUTE_NXT_HOP = {"ipv4": ipv4_nxt_hop, "ipv6": ipv6_link_local_nxt_hop}
+        if index == 1:
+            result = verify_rib_default_route(
+                tgen,
+                topo,
+                dut="r2",
+                routes=DEFAULT_ROUTES,
+                expected_nexthop=DEFAULT_ROUTE_NXT_HOP,
+            )
+            assert result is True, "Testcase {} : Failed \n Error: {}".format(
+                tc_name, result
+            )
+        else:
+            result = verify_rib_default_route(
+                tgen,
+                topo,
+                dut="r2",
+                routes=DEFAULT_ROUTES,
+                expected_nexthop=DEFAULT_ROUTE_NXT_HOP,
+            )
+            assert result is True, "Testcase {} : Failed \n Error: {}".format(
+                tc_name, result
+            )
+
+    step(
+        "Configure IPv4 and IPv6  route-map with deny option on R2 to filter default route  0.0.0.0/0 and 0::0/0"
+    )
+    DEFAULT_ROUTES = {"ipv4": "0.0.0.0/0", "ipv6": "0::0/0"}
+    input_dict_3 = {
+        "r2": {
+            "prefix_lists": {
+                "ipv4": {
+                    "Pv4": [
+                        {
+                            "seqid": "1",
+                            "network": DEFAULT_ROUTES["ipv4"],
+                            "action": "permit",
+                        }
+                    ]
+                },
+                "ipv6": {
+                    "Pv6": [
+                        {
+                            "seqid": "1",
+                            "network": DEFAULT_ROUTES["ipv6"],
+                            "action": "permit",
+                        }
+                    ]
+                },
+            }
+        }
+    }
+    result = create_prefix_lists(tgen, input_dict_3)
+    assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
+
+    input_dict_3 = {
+        "r2": {
+            "route_maps": {
+                "RMv4": [
+                    {
+                        "action": "deny",
+                        "seq_id": "1",
+                        "match": {"ipv4": {"prefix_lists": "Pv4"}},
+                    },
+                ],
+                "RMv6": [
+                    {
+                        "action": "deny",
+                        "seq_id": "1",
+                        "match": {"ipv6": {"prefix_lists": "Pv6"}},
+                    },
+                ],
+            }
+        }
+    }
+    result = create_route_maps(tgen, input_dict_3)
+    assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
+
+    step("Apply route-map IN direction of R2 ( R2-R1) for IPv4 and IPv6 BGP neighbors")
+    r2_link = None
+    for index in range(1, 3):
+        r2_link = "r2-link" + str(index)
+        input_dict_4 = {
+            "r2": {
+                "bgp": {
+                    "address_family": {
+                        "ipv4": {
+                            "unicast": {
+                                "neighbor": {
+                                    "r1": {
+                                        "dest_link": {
+                                            r2_link: {
+                                                "route_maps": [
+                                                    {"name": "RMv4", "direction": "in"}
+                                                ]
+                                            },
+                                        }
+                                    }
+                                }
+                            }
+                        },
+                        "ipv6": {
+                            "unicast": {
+                                "neighbor": {
+                                    "r1": {
+                                        "dest_link": {
+                                            r2_link: {
+                                                "route_maps": [
+                                                    {"name": "RMv6", "direction": "in"}
+                                                ]
+                                            },
+                                        }
+                                    }
+                                }
+                            }
+                        },
+                    }
+                }
+            }
+        }
+        result = create_router_bgp(tgen, topo, input_dict_4)
+        assert result is True, "Testcase {} : Failed \n Error: {}".format(
+            tc_name, result
+        )
+
+    step("After applying the route-map the routes are not expected in RIB ")
+    r2_link = None
+    for index in range(1, 3):
+        r2_link = "r2-link" + str(index)
+        ipv4_nxt_hop = topo["routers"]["r1"]["links"][r2_link]["ipv4"].split("/")[0]
+        interface = topo["routers"]["r1"]["links"][r2_link]["interface"]
+        ipv6_link_local_nxt_hop = get_frr_ipv6_linklocal(tgen, "r1", intf=interface)
+        DEFAULT_ROUTE_NXT_HOP = {"ipv4": ipv4_nxt_hop, "ipv6": ipv6_link_local_nxt_hop}
+
+        result = verify_rib_default_route(
+            tgen,
+            topo,
+            dut="r2",
+            routes=DEFAULT_ROUTES,
+            expected_nexthop=DEFAULT_ROUTE_NXT_HOP,
+            expected=False,
+        )
+        assert result is not True, "Testcase {} : Failed \n Error: {}".format(
+            tc_name, result
+        )
+
+    write_test_footer(tc_name)
+
+
 if __name__ == "__main__":
     args = ["-s"] + sys.argv[1:]
     sys.exit(pytest.main(args))