#!/usr/bin/env python
+# SPDX-License-Identifier: ISC
#
# Copyright (c) 2022 by VMware, Inc. ("VMware")
+# Shreenidhi A R <rshreenidhi@vmware.com>
# Used Copyright (c) 2018 by Network Device Education Foundation, Inc. ("NetDEF")
# in this file.
#
-# Permission to use, copy, modify, and/or distribute this software
-# for any purpose with or without fee is hereby granted, provided
-# that the above copyright notice and this permission notice appear
-# in all copies.
-# Shreenidhi A R <rshreenidhi@vmware.com>
-# THE SOFTWARE IS PROVIDED "AS IS" AND VMWARE DISCLAIMS ALL WARRANTIES
-# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
-# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL VMWARE BE LIABLE FOR
-# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
-# DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
-# WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
-# ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
-# OF THIS SOFTWARE.
-#
"""
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
import datetime
from copy import deepcopy
from lib.topolog import logger
+from time import sleep
# pylint: disable=C0413
# Import topogen and topotest helpers
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,
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
step("Taking uptime snapshot before configuring default - originate")
uptime_before_ipv4 = get_rib_route_uptime(tgen, "ipv4", "r2", ipv4_uptime_dict)
uptime_before_ipv6 = get_rib_route_uptime(tgen, "ipv6", "r2", ipv6_uptime_dict)
+ sleep(1)
step(
"Configure default-originate on R1 link-1 again for IPv4 and IPv6 address family"
step("Taking uptime snapshot before removing redisctribute static ")
uptime_before_ipv4 = get_rib_route_uptime(tgen, "ipv4", "r2", ipv4_uptime_dict)
uptime_before_ipv6 = get_rib_route_uptime(tgen, "ipv6", "r2", ipv6_uptime_dict)
+ sleep(1)
step("Remove redistribute static from IPv4 and IPv6 address family ")
input_dict_1 = {
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))
projects / mirror_frr.git / blobdiff