[mirror_frr.git] / tests / topotests / bgp_gr_functionality_topo3 / bgp_gr_functionality_topo3.py

#!/usr/bin/env python
#
# Copyright (c) 2019 by VMware, Inc. ("VMware")
# 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.
#
# 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.
#

import os
import sys
import time
import pytest
from time import sleep

import traceback
import ipaddress

# Save the Current Working Directory to find configuration files.
CWD = os.path.dirname(os.path.realpath(__file__))
sys.path.append(os.path.join("../"))
sys.path.append(os.path.join("../lib/"))

# pylint: disable=C0413
# Import topogen and topotest helpers
from lib.topogen import Topogen, get_topogen
from lib.topolog import logger

# Required to instantiate the topology builder class.

# Import topoJson from lib, to create topology and initial configuration
from lib.topojson import build_config_from_json
from lib.bgp import (
    clear_bgp,
    verify_bgp_rib,
    verify_graceful_restart,
    create_router_bgp,
    verify_r_bit,
    verify_eor,
    verify_f_bit,
    verify_bgp_convergence,
    verify_gr_address_family,
    modify_bgp_config_when_bgpd_down,
    verify_graceful_restart_timers,
    verify_bgp_convergence_from_running_config,
)

# Import common_config to use commomnly used APIs
from lib.common_config import (
    create_common_configuration,
    InvalidCLIError,
    retry,
    generate_ips,
    FRRCFG_FILE,
    find_interface_with_greater_ip,
    check_address_types,
    validate_ip_address,
    run_frr_cmd,
    get_frr_ipv6_linklocal,
)

from lib.common_config import (
    write_test_header,
    reset_config_on_routers,
    start_topology,
    kill_router_daemons,
    start_router_daemons,
    verify_rib,
    check_address_types,
    write_test_footer,
    check_router_status,
    step,
    get_frr_ipv6_linklocal,
    create_static_routes,
    required_linux_kernel_version,
)

pytestmark = [pytest.mark.bgpd]


# Global variables
BGP_CONVERGENCE = False
GR_RESTART_TIMER = 5
GR_SELECT_DEFER_TIMER = 5
GR_STALEPATH_TIMER = 5
# Global variables
# STATIC_ROUTES=[]
NETWORK1_1 = {"ipv4": "192.0.2.1/32", "ipv6": "2001:DB8::1:1/128"}
NETWORK1_2 = {"ipv4": "192.0.2.2/32", "ipv6": "2001:DB8::2:1/128"}
NETWORK2_1 = {"ipv4": "192.0.2.3/32", "ipv6": "2001:DB8::3:1/128"}
NETWORK2_2 = {"ipv4": "192.0.2.4/32", "ipv6": "2001:DB8::4:1/128"}
NETWORK3_1 = {"ipv4": "192.0.2.5/32", "ipv6": "2001:DB8::5:1/128"}
NETWORK3_2 = {"ipv4": "192.0.2.6/32", "ipv6": "2001:DB8::6:1/128"}
NETWORK4_1 = {"ipv4": "192.0.2.7/32", "ipv6": "2001:DB8::7:1/128"}
NETWORK4_2 = {"ipv4": "192.0.2.8/32", "ipv6": "2001:DB8::8:1/128"}
NETWORK5_1 = {"ipv4": "192.0.2.9/32", "ipv6": "2001:DB8::9:1/128"}
NETWORK5_2 = {"ipv4": "192.0.2.10/32", "ipv6": "2001:DB8::10:1/128"}

NEXT_HOP_IP = {"ipv4": "Null0", "ipv6": "Null0"}

PREFERRED_NEXT_HOP = "link_local"


def configure_gr_followed_by_clear(tgen, topo, input_dict, tc_name, dut, peer):
    """
    result = configure_gr_followed_by_clear(tgen, topo, dut)
    assert result is True, \
        "Testcase {} :Failed \n Error {}". \
            format(tc_name, result)
    """

    result = create_router_bgp(tgen, topo, input_dict)
    assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
    for addr_type in ADDR_TYPES:
        clear_bgp(tgen, addr_type, dut)

    for addr_type in ADDR_TYPES:
        clear_bgp(tgen, addr_type, peer)

    result = verify_bgp_convergence_from_running_config(tgen)
    assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)

    return True


def verify_stale_routes_list(tgen, addr_type, dut, input_dict):
    """
    This API is use verify Stale routes on refering the network with next hop value
    Parameters
    ----------
    * `tgen`: topogen object
    * `dut`: input dut router name
    * `addr_type` : ip type ipv4/ipv6
    * `input_dict` : input dict, has details of static routes
    Usage
    -----
    dut = 'r1'
    input_dict = {
                "r3": {
                    "static_routes": [

                        {
                            "network": [NETWORK1_1[addr_type]],
                            "no_of_ip": 2,
                            "vrf": "RED"
                        }
                    ]
                }
            }

    result = verify_stale_routes_list(tgen, addr_type, dut, input_dict)
    Returns
    -------
    errormsg(str) or True
    """
    logger.debug("Entering lib API: verify_stale_routes_list()")
    router_list = tgen.routers()
    additional_nexthops_in_required_nhs = []
    list1 = []
    list2 = []
    found_hops = []
    for routerInput in input_dict.keys():
        for router, rnode in router_list.items():
            if router != dut:
                continue
            # Verifying RIB routes
            command = "show bgp"
            # Static routes
            sleep(2)
            logger.info("Checking router {} BGP RIB:".format(dut))
            if "static_routes" in input_dict[routerInput]:
                static_routes = input_dict[routerInput]["static_routes"]
                for static_route in static_routes:
                    found_routes = []
                    missing_routes = []
                    st_found = False
                    nh_found = False
                    vrf = static_route.setdefault("vrf", None)
                    community = static_route.setdefault("community", None)
                    largeCommunity = static_route.setdefault("largeCommunity", None)
                    if vrf:
                        cmd = "{} vrf {} {}".format(command, vrf, addr_type)
                        if community:
                            cmd = "{} community {}".format(cmd, community)
                        if largeCommunity:
                            cmd = "{} large-community {}".format(cmd, largeCommunity)
                    else:
                        cmd = "{} {}".format(command, addr_type)
                    cmd = "{} json".format(cmd)
                    rib_routes_json = run_frr_cmd(rnode, cmd, isjson=True)
                    # Verifying output dictionary rib_routes_json is not empty
                    if bool(rib_routes_json) == False:
                        errormsg = "[DUT: {}]: No route found in rib of router".format(
                            router
                        )
                        return errormsg
                    elif "warning" in rib_routes_json:
                        errormsg = "[DUT: {}]: {}".format(
                            router, rib_routes_json["warning"]
                        )
                        return errormsg
                    network = static_route["network"]
                    if "no_of_ip" in static_route:
                        no_of_ip = static_route["no_of_ip"]
                    else:
                        no_of_ip = 1
                    # Generating IPs for verification
                    ip_list = generate_ips(network, no_of_ip)

                    for st_rt in ip_list:
                        st_rt = str(ipaddress.ip_network(st_rt))
                        _addr_type = validate_ip_address(st_rt)
                        if _addr_type != addr_type:
                            continue
                        if st_rt in rib_routes_json["routes"]:
                            st_found = True

                            found_routes.append(st_rt)
                            for mnh in range(0, len(rib_routes_json["routes"][st_rt])):
                                found_hops.append(
                                    [
                                        rib_r["ip"]
                                        for rib_r in rib_routes_json["routes"][st_rt][
                                            mnh
                                        ]["nexthops"]
                                    ]
                                )
                            return found_hops
                        else:
                            return "error  msg - no hops found"


def setup_module(mod):
    """
    Sets up the pytest environment

    * `mod`: module name
    """

    # Required linux kernel version for this suite to run.
    result = required_linux_kernel_version("4.16")
    if result is not True:
        pytest.skip("Kernel requirements are not met, kernel version should be >=4.16")

    global ADDR_TYPES

    testsuite_run_time = time.asctime(time.localtime(time.time()))
    logger.info("Testsuite start time: {}".format(testsuite_run_time))
    logger.info("=" * 40)

    logger.info("Running setup_module to create topology")

    # This function initiates the topology build with Topogen...
    json_file = "{}/bgp_gr_functionality_topo3.json".format(CWD)
    tgen = Topogen(json_file, mod.__name__)
    global topo
    topo = tgen.json_topo
    # ... and here it calls Mininet initialization functions.

    # Starting topology, create tmp files which are loaded to routers
    #  to start daemons and then start routers
    start_topology(tgen)

    # Creating configuration from JSON
    build_config_from_json(tgen, topo)

    # Api call verify whether BGP is converged
    ADDR_TYPES = check_address_types()

    for addr_type in ADDR_TYPES:
        BGP_CONVERGENCE = verify_bgp_convergence(tgen, topo)
        assert BGP_CONVERGENCE is True, "setup_module : Failed \n Error:" " {}".format(
            BGP_CONVERGENCE
        )

    logger.info("Running setup_module() done")


def teardown_module(mod):
    """
    Teardown the pytest environment

    * `mod`: module name
    """

    logger.info("Running teardown_module to delete topology")

    tgen = get_topogen()

    # Stop toplogy and Remove tmp files
    tgen.stop_topology()

    logger.info(
        "Testsuite end time: {}".format(time.asctime(time.localtime(time.time())))
    )
    logger.info("=" * 40)


################################################################################
#
#                       TEST CASES
#
################################################################################
def test_bgp_gr_stale_routes(request):
    tgen = get_topogen()
    tc_name = request.node.name
    write_test_header(tc_name)

    step("Verify the router failures")
    if tgen.routers_have_failure():
        check_router_status(tgen)

    step("Creating 5 static Routes in Router R3 with NULL0 as Next hop")
    for addr_type in ADDR_TYPES:
        input_dict_1 = {
            "r3": {
                "static_routes": [
                    {
                        "network": [NETWORK1_1[addr_type]] + [NETWORK1_2[addr_type]],
                        "next_hop": NEXT_HOP_IP[addr_type],
                    },
                    {
                        "network": [NETWORK2_1[addr_type]] + [NETWORK2_2[addr_type]],
                        "next_hop": NEXT_HOP_IP[addr_type],
                    },
                    {
                        "network": [NETWORK3_1[addr_type]] + [NETWORK3_2[addr_type]],
                        "next_hop": NEXT_HOP_IP[addr_type],
                    },
                    {
                        "network": [NETWORK4_1[addr_type]] + [NETWORK4_2[addr_type]],
                        "next_hop": NEXT_HOP_IP[addr_type],
                    },
                    {
                        "network": [NETWORK5_1[addr_type]] + [NETWORK5_2[addr_type]],
                        "next_hop": NEXT_HOP_IP[addr_type],
                    },
                ]
            }
        }
        result = create_static_routes(tgen, input_dict_1)
        assert result is True, "Testcase {} : Failed \n Error: {}".format(
            tc_name, result
        )
    step("verifying Created  Route  at R3 in VRF default")
    for addr_type in ADDR_TYPES:
        dut = "r3"
        input_dict_1 = {"r3": topo["routers"]["r3"]}
        result = verify_bgp_rib(tgen, addr_type, dut, input_dict_1)
        assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
    # done
    step("verifying Created  Route  at R2 in VRF default")
    for addr_type in ADDR_TYPES:
        dut = "r2"
        input_dict_1 = {"r2": topo["routers"]["r2"]}
        result = verify_bgp_rib(tgen, addr_type, dut, input_dict_1)
        assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
    step("importing vrf RED on R2 under Address Family")
    for addr_type in ADDR_TYPES:
        input_import_vrf = {
            "r2": {
                "bgp": [
                    {
                        "local_as": 200,
                        "vrf": "RED",
                        "address_family": {
                            addr_type: {"unicast": {"import": {"vrf": "default"}}}
                        },
                    }
                ]
            }
        }
        result = create_router_bgp(tgen, topo, input_import_vrf)
        assert result is True, "Testcase {} : Failed \n Error: {}".format(
            tc_name, result
        )
    # done
    step("verifying static  Routes  at R2 in VRF RED")
    for addr_type in ADDR_TYPES:
        dut = "r2"
        input_dict_1 = {"r2": topo["routers"]["r2"]}
        result = verify_bgp_rib(tgen, addr_type, dut, input_dict_1)
        assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)

    step("verifying static  Routes  at R1 in VRF RED")
    for addr_type in ADDR_TYPES:
        dut = "r1"
        input_dict_1 = {"r1": topo["routers"]["r1"]}
        result = verify_bgp_rib(tgen, addr_type, dut, input_dict_1)
        assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)

    step("Configuring Graceful restart at R2 and R3 ")
    input_dict = {
        "r2": {
            "bgp": {
                "local_as": "200",
                "graceful-restart": {
                    "graceful-restart": True,
                },
            }
        },
        "r3": {
            "bgp": {"local_as": "300", "graceful-restart": {"graceful-restart": True}}
        },
    }

    configure_gr_followed_by_clear(tgen, topo, input_dict, tc_name, dut="r2", peer="r3")

    step("verify Graceful restart at R2")
    for addr_type in ADDR_TYPES:
        result = verify_graceful_restart(
            tgen, topo, addr_type, input_dict, dut="r2", peer="r3"
        )
        assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)

    step("verify Graceful restart at R3")
    for addr_type in ADDR_TYPES:
        result = verify_graceful_restart(
            tgen, topo, addr_type, input_dict, dut="r3", peer="r2"
        )
        assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)

    step("Configuring Graceful-restart-disable at R3")
    input_dict = {
        "r2": {
            "bgp": {
                "local_as": "200",
                "graceful-restart": {
                    "graceful-restart": False,
                },
            }
        },
        "r3": {
            "bgp": {"local_as": "300", "graceful-restart": {"graceful-restart": False}}
        },
    }
    configure_gr_followed_by_clear(tgen, topo, input_dict, tc_name, dut="r3", peer="r2")

    step("Verify Graceful-restart-disable at R3")
    for addr_type in ADDR_TYPES:
        result = verify_graceful_restart(
            tgen, topo, addr_type, input_dict, dut="r3", peer="r2"
        )
        assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)

    for iteration in range(5):
        step("graceful-restart-disable:True  at R3")
        input_dict = {
            "r3": {
                "bgp": {
                    "graceful-restart": {
                        "graceful-restart-disable": True,
                    }
                }
            }
        }
        configure_gr_followed_by_clear(
            tgen, topo, input_dict, tc_name, dut="r3", peer="r2"
        )

        step("Verifying  Routes at R2 on enabling GRD")
        dut = "r2"
        for addr_type in ADDR_TYPES:
            input_dict_1 = {"r2": topo["routers"]["r2"]}
            result = verify_bgp_rib(tgen, addr_type, dut, input_dict_1)
            assert result is True, "Testcase {} :Failed \n Error {}".format(
                tc_name, result
            )

        step("Verify stale Routes in Router R2 enabling GRD")
        for addr_type in ADDR_TYPES:
            dut = "r2"
            protocol = "bgp"
            verify_nh_for_static_rtes = {
                "r3": {
                    "static_routes": [
                        {
                            "network": [NETWORK1_1[addr_type]],
                            "no_of_ip": 2,
                            "vrf": "RED",
                        }
                    ]
                }
            }
            bgp_rib_next_hops = verify_stale_routes_list(
                tgen, addr_type, dut, verify_nh_for_static_rtes
            )
            assert (
                len(bgp_rib_next_hops) == 1
            ) is True, "Testcase {} : Failed \n Error: {}".format(
                tc_name, bgp_rib_next_hops, expected=True
            )

        step("graceful-restart-disable:False at R3")
        input_dict = {
            "r3": {
                "bgp": {
                    "graceful-restart": {
                        "graceful-restart-disable": False,
                    }
                }
            }
        }
        configure_gr_followed_by_clear(
            tgen, topo, input_dict, tc_name, dut="r3", peer="r2"
        )

        step("Verifying  Routes at R2 on disabling GRD")
        dut = "r2"
        for addr_type in ADDR_TYPES:
            input_dict_1 = {"r2": topo["routers"]["r2"]}
            result = verify_bgp_rib(tgen, addr_type, dut, input_dict_1)
            assert result is True, "Testcase {} :Failed \n Error {}".format(
                tc_name, result
            )

        step("Verify stale Routes in Router R2 on disabling GRD")
        for addr_type in ADDR_TYPES:
            dut = "r2"
            protocol = "bgp"
            verify_nh_for_static_rtes = {
                "r3": {
                    "static_routes": [
                        {
                            "network": [NETWORK1_1[addr_type]],
                            "no_of_ip": 2,
                            "vrf": "RED",
                        }
                    ]
                }
            }
            bgp_rib_next_hops = verify_stale_routes_list(
                tgen, addr_type, dut, verify_nh_for_static_rtes
            )

            stale_route_status = len(bgp_rib_next_hops) == 1
            assert (
                stale_route_status is True
            ), "Testcase {} : Failed \n Error: {}".format(
                tc_name, stale_route_status, expected=True
            )
    write_test_footer(tc_name)
Commit	Line	Data
771ac547 A	1	#!/usr/bin/env python
	2	#
	3	# Copyright (c) 2019 by VMware, Inc. ("VMware")
	4	# Used Copyright (c) 2018 by Network Device Education Foundation, Inc. ("NetDEF")
	5	# in this file.
	6	#
	7	# Permission to use, copy, modify, and/or distribute this software
	8	# for any purpose with or without fee is hereby granted, provided
	9	# that the above copyright notice and this permission notice appear
	10	# in all copies.
	11	#
	12	# THE SOFTWARE IS PROVIDED "AS IS" AND VMWARE DISCLAIMS ALL WARRANTIES
	13	# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
	14	# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL VMWARE BE LIABLE FOR
	15	# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
	16	# DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
	17	# WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
	18	# ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
	19	# OF THIS SOFTWARE.
	20	#
	21
	22	import os
	23	import sys
	24	import time
	25	import pytest
	26	from time import sleep
	27
	28	import traceback
	29	import ipaddress
	30
	31	# Save the Current Working Directory to find configuration files.
	32	CWD = os.path.dirname(os.path.realpath(__file__))
	33	sys.path.append(os.path.join("../"))
	34	sys.path.append(os.path.join("../lib/"))
	35
	36	# pylint: disable=C0413
	37	# Import topogen and topotest helpers
	38	from lib.topogen import Topogen, get_topogen
	39	from lib.topolog import logger
	40
	41	# Required to instantiate the topology builder class.
	42
	43	# Import topoJson from lib, to create topology and initial configuration
	44	from lib.topojson import build_config_from_json
	45	from lib.bgp import (
	46	clear_bgp,
	47	verify_bgp_rib,
	48	verify_graceful_restart,
	49	create_router_bgp,
	50	verify_r_bit,
	51	verify_eor,
	52	verify_f_bit,
	53	verify_bgp_convergence,
	54	verify_gr_address_family,
	55	modify_bgp_config_when_bgpd_down,
	56	verify_graceful_restart_timers,
	57	verify_bgp_convergence_from_running_config,
	58	)
84ab3836	59
771ac547	60	# Import common_config to use commomnly used APIs
84ab3836 KK	61	from lib.common_config import (
	62	create_common_configuration,
	63	InvalidCLIError,
	64	retry,
	65	generate_ips,
	66	FRRCFG_FILE,
	67	find_interface_with_greater_ip,
	68	check_address_types,
	69	validate_ip_address,
	70	run_frr_cmd,
	71	get_frr_ipv6_linklocal,
	72	)
771ac547 A	73
	74	from lib.common_config import (
	75	write_test_header,
	76	reset_config_on_routers,
	77	start_topology,
	78	kill_router_daemons,
	79	start_router_daemons,
	80	verify_rib,
	81	check_address_types,
	82	write_test_footer,
	83	check_router_status,
	84	step,
	85	get_frr_ipv6_linklocal,
	86	create_static_routes,
	87	required_linux_kernel_version,
	88	)
	89
	90	pytestmark = [pytest.mark.bgpd]
	91
	92
	93	# Global variables
	94	BGP_CONVERGENCE = False
	95	GR_RESTART_TIMER = 5
	96	GR_SELECT_DEFER_TIMER = 5
	97	GR_STALEPATH_TIMER = 5
	98	# Global variables
	99	# STATIC_ROUTES=[]
	100	NETWORK1_1 = {"ipv4": "192.0.2.1/32", "ipv6": "2001:DB8::1:1/128"}
	101	NETWORK1_2 = {"ipv4": "192.0.2.2/32", "ipv6": "2001:DB8::2:1/128"}
	102	NETWORK2_1 = {"ipv4": "192.0.2.3/32", "ipv6": "2001:DB8::3:1/128"}
	103	NETWORK2_2 = {"ipv4": "192.0.2.4/32", "ipv6": "2001:DB8::4:1/128"}
	104	NETWORK3_1 = {"ipv4": "192.0.2.5/32", "ipv6": "2001:DB8::5:1/128"}
	105	NETWORK3_2 = {"ipv4": "192.0.2.6/32", "ipv6": "2001:DB8::6:1/128"}
	106	NETWORK4_1 = {"ipv4": "192.0.2.7/32", "ipv6": "2001:DB8::7:1/128"}
	107	NETWORK4_2 = {"ipv4": "192.0.2.8/32", "ipv6": "2001:DB8::8:1/128"}
	108	NETWORK5_1 = {"ipv4": "192.0.2.9/32", "ipv6": "2001:DB8::9:1/128"}
	109	NETWORK5_2 = {"ipv4": "192.0.2.10/32", "ipv6": "2001:DB8::10:1/128"}
	110
	111	NEXT_HOP_IP = {"ipv4": "Null0", "ipv6": "Null0"}
	112
	113	PREFERRED_NEXT_HOP = "link_local"
	114
	115
84ab3836	116	def configure_gr_followed_by_clear(tgen, topo, input_dict, tc_name, dut, peer):
771ac547 A	117	"""
	118	result = configure_gr_followed_by_clear(tgen, topo, dut)
	119	assert result is True, \
	120	"Testcase {} :Failed \n Error {}". \
	121	format(tc_name, result)
	122	"""
	123
	124	result = create_router_bgp(tgen, topo, input_dict)
84ab3836	125	assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
771ac547 A	126	for addr_type in ADDR_TYPES:
	127	clear_bgp(tgen, addr_type, dut)
	128
	129	for addr_type in ADDR_TYPES:
	130	clear_bgp(tgen, addr_type, peer)
	131
	132	result = verify_bgp_convergence_from_running_config(tgen)
	133	assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
	134
	135	return True
	136
84ab3836	137
771ac547 A	138	def verify_stale_routes_list(tgen, addr_type, dut, input_dict):
	139	"""
	140	This API is use verify Stale routes on refering the network with next hop value
	141	Parameters
	142	----------
	143	* `tgen`: topogen object
	144	* `dut`: input dut router name
	145	* `addr_type` : ip type ipv4/ipv6
	146	* `input_dict` : input dict, has details of static routes
	147	Usage
	148	-----
	149	dut = 'r1'
	150	input_dict = {
	151	"r3": {
	152	"static_routes": [
	153
	154	{
	155	"network": [NETWORK1_1[addr_type]],
	156	"no_of_ip": 2,
	157	"vrf": "RED"
	158	}
	159	]
	160	}
	161	}
	162
	163	result = verify_stale_routes_list(tgen, addr_type, dut, input_dict)
	164	Returns
	165	-------
	166	errormsg(str) or True
	167	"""
	168	logger.debug("Entering lib API: verify_stale_routes_list()")
	169	router_list = tgen.routers()
	170	additional_nexthops_in_required_nhs = []
	171	list1 = []
	172	list2 = []
	173	found_hops = []
	174	for routerInput in input_dict.keys():
	175	for router, rnode in router_list.items():
	176	if router != dut:
	177	continue
	178	# Verifying RIB routes
	179	command = "show bgp"
	180	# Static routes
	181	sleep(2)
84ab3836 KK	182	logger.info("Checking router {} BGP RIB:".format(dut))
84ab3836 KK	183	if "static_routes" in input_dict[routerInput]:
771ac547 A	184	static_routes = input_dict[routerInput]["static_routes"]
	185	for static_route in static_routes:
	186	found_routes = []
	187	missing_routes = []
	188	st_found = False
	189	nh_found = False
	190	vrf = static_route.setdefault("vrf", None)
	191	community = static_route.setdefault("community", None)
84ab3836	192	largeCommunity = static_route.setdefault("largeCommunity", None)
771ac547	193	if vrf:
84ab3836	194	cmd = "{} vrf {} {}".format(command, vrf, addr_type)
771ac547	195	if community:
84ab3836	196	cmd = "{} community {}".format(cmd, community)
771ac547	197	if largeCommunity:
84ab3836	198	cmd = "{} large-community {}".format(cmd, largeCommunity)
771ac547	199	else:
84ab3836	200	cmd = "{} {}".format(command, addr_type)
771ac547 A	201	cmd = "{} json".format(cmd)
	202	rib_routes_json = run_frr_cmd(rnode, cmd, isjson=True)
	203	# Verifying output dictionary rib_routes_json is not empty
	204	if bool(rib_routes_json) == False:
84ab3836 KK	205	errormsg = "[DUT: {}]: No route found in rib of router".format(
	206	router
	207	)
771ac547 A	208	return errormsg
771ac547 A	209	elif "warning" in rib_routes_json:
84ab3836 KK	210	errormsg = "[DUT: {}]: {}".format(
	211	router, rib_routes_json["warning"]
	212	)
771ac547 A	213	return errormsg
	214	network = static_route["network"]
	215	if "no_of_ip" in static_route:
	216	no_of_ip = static_route["no_of_ip"]
	217	else:
	218	no_of_ip = 1
	219	# Generating IPs for verification
	220	ip_list = generate_ips(network, no_of_ip)
	221
	222	for st_rt in ip_list:
	223	st_rt = str(ipaddress.ip_network(st_rt))
	224	_addr_type = validate_ip_address(st_rt)
	225	if _addr_type != addr_type:
	226	continue
	227	if st_rt in rib_routes_json["routes"]:
	228	st_found = True
	229
	230	found_routes.append(st_rt)
84ab3836 KK	231	for mnh in range(0, len(rib_routes_json["routes"][st_rt])):
	232	found_hops.append(
	233	[
	234	rib_r["ip"]
	235	for rib_r in rib_routes_json["routes"][st_rt][
	236	mnh
	237	]["nexthops"]
	238	]
	239	)
771ac547 A	240	return found_hops
771ac547 A	241	else:
84ab3836	242	return "error msg - no hops found"
771ac547 A	243
	244
	245	def setup_module(mod):
	246	"""
	247	Sets up the pytest environment
	248
	249	* `mod`: module name
	250	"""
	251
	252	# Required linux kernel version for this suite to run.
	253	result = required_linux_kernel_version("4.16")
	254	if result is not True:
d63c7094	255	pytest.skip("Kernel requirements are not met, kernel version should be >=4.16")
771ac547 A	256
	257	global ADDR_TYPES
	258
	259	testsuite_run_time = time.asctime(time.localtime(time.time()))
	260	logger.info("Testsuite start time: {}".format(testsuite_run_time))
	261	logger.info("=" * 40)
	262
	263	logger.info("Running setup_module to create topology")
	264
	265	# This function initiates the topology build with Topogen...
	266	json_file = "{}/bgp_gr_functionality_topo3.json".format(CWD)
	267	tgen = Topogen(json_file, mod.__name__)
	268	global topo
	269	topo = tgen.json_topo
	270	# ... and here it calls Mininet initialization functions.
	271
	272	# Starting topology, create tmp files which are loaded to routers
d60a3f0e	273	# to start daemons and then start routers
771ac547 A	274	start_topology(tgen)
	275
	276	# Creating configuration from JSON
	277	build_config_from_json(tgen, topo)
	278
	279	# Api call verify whether BGP is converged
	280	ADDR_TYPES = check_address_types()
	281
	282	for addr_type in ADDR_TYPES:
	283	BGP_CONVERGENCE = verify_bgp_convergence(tgen, topo)
	284	assert BGP_CONVERGENCE is True, "setup_module : Failed \n Error:" " {}".format(
	285	BGP_CONVERGENCE
	286	)
	287
	288	logger.info("Running setup_module() done")
	289
	290
	291	def teardown_module(mod):
	292	"""
	293	Teardown the pytest environment
	294
	295	* `mod`: module name
	296	"""
	297
	298	logger.info("Running teardown_module to delete topology")
	299
	300	tgen = get_topogen()
	301
	302	# Stop toplogy and Remove tmp files
	303	tgen.stop_topology()
	304
	305	logger.info(
	306	"Testsuite end time: {}".format(time.asctime(time.localtime(time.time())))
	307	)
	308	logger.info("=" * 40)
84ab3836 KK	309
84ab3836 KK	310
771ac547 A	311	################################################################################
	312	#
	313	# TEST CASES
	314	#
	315	################################################################################
	316	def test_bgp_gr_stale_routes(request):
	317	tgen = get_topogen()
	318	tc_name = request.node.name
	319	write_test_header(tc_name)
	320
	321	step("Verify the router failures")
	322	if tgen.routers_have_failure():
	323	check_router_status(tgen)
	324
	325	step("Creating 5 static Routes in Router R3 with NULL0 as Next hop")
	326	for addr_type in ADDR_TYPES:
84ab3836 KK	327	input_dict_1 = {
	328	"r3": {
	329	"static_routes": [
	330	{
	331	"network": [NETWORK1_1[addr_type]] + [NETWORK1_2[addr_type]],
	332	"next_hop": NEXT_HOP_IP[addr_type],
	333	},
	334	{
	335	"network": [NETWORK2_1[addr_type]] + [NETWORK2_2[addr_type]],
	336	"next_hop": NEXT_HOP_IP[addr_type],
	337	},
	338	{
	339	"network": [NETWORK3_1[addr_type]] + [NETWORK3_2[addr_type]],
	340	"next_hop": NEXT_HOP_IP[addr_type],
	341	},
	342	{
	343	"network": [NETWORK4_1[addr_type]] + [NETWORK4_2[addr_type]],
	344	"next_hop": NEXT_HOP_IP[addr_type],
	345	},
	346	{
	347	"network": [NETWORK5_1[addr_type]] + [NETWORK5_2[addr_type]],
	348	"next_hop": NEXT_HOP_IP[addr_type],
	349	},
	350	]
771ac547	351	}
84ab3836 KK	352	}
	353	result = create_static_routes(tgen, input_dict_1)
	354	assert result is True, "Testcase {} : Failed \n Error: {}".format(
	355	tc_name, result
	356	)
771ac547 A	357	step("verifying Created Route at R3 in VRF default")
771ac547 A	358	for addr_type in ADDR_TYPES:
84ab3836 KK	359	dut = "r3"
84ab3836 KK	360	input_dict_1 = {"r3": topo["routers"]["r3"]}
771ac547	361	result = verify_bgp_rib(tgen, addr_type, dut, input_dict_1)
84ab3836 KK	362	assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
84ab3836 KK	363	# done
771ac547 A	364	step("verifying Created Route at R2 in VRF default")
771ac547 A	365	for addr_type in ADDR_TYPES:
84ab3836 KK	366	dut = "r2"
84ab3836 KK	367	input_dict_1 = {"r2": topo["routers"]["r2"]}
771ac547	368	result = verify_bgp_rib(tgen, addr_type, dut, input_dict_1)
84ab3836	369	assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
771ac547 A	370	step("importing vrf RED on R2 under Address Family")
771ac547 A	371	for addr_type in ADDR_TYPES:
84ab3836	372	input_import_vrf = {
771ac547 A	373	"r2": {
	374	"bgp": [
	375	{
	376	"local_as": 200,
	377	"vrf": "RED",
84ab3836 KK	378	"address_family": {
	379	addr_type: {"unicast": {"import": {"vrf": "default"}}}
	380	},
771ac547 A	381	}
	382	]
	383	}
	384	}
84ab3836 KK	385	result = create_router_bgp(tgen, topo, input_import_vrf)
	386	assert result is True, "Testcase {} : Failed \n Error: {}".format(
	387	tc_name, result
	388	)
	389	# done
771ac547 A	390	step("verifying static Routes at R2 in VRF RED")
771ac547 A	391	for addr_type in ADDR_TYPES:
84ab3836 KK	392	dut = "r2"
84ab3836 KK	393	input_dict_1 = {"r2": topo["routers"]["r2"]}
771ac547	394	result = verify_bgp_rib(tgen, addr_type, dut, input_dict_1)
84ab3836	395	assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
771ac547 A	396
	397	step("verifying static Routes at R1 in VRF RED")
	398	for addr_type in ADDR_TYPES:
84ab3836 KK	399	dut = "r1"
84ab3836 KK	400	input_dict_1 = {"r1": topo["routers"]["r1"]}
771ac547	401	result = verify_bgp_rib(tgen, addr_type, dut, input_dict_1)
84ab3836	402	assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
771ac547 A	403
	404	step("Configuring Graceful restart at R2 and R3 ")
	405	input_dict = {
	406	"r2": {
771ac547 A	407	"bgp": {
	408	"local_as": "200",
	409	"graceful-restart": {
	410	"graceful-restart": True,
84ab3836	411	},
771ac547 A	412	}
	413	},
	414	"r3": {
84ab3836 KK	415	"bgp": {"local_as": "300", "graceful-restart": {"graceful-restart": True}}
84ab3836 KK	416	},
771ac547 A	417	}
771ac547 A	418
84ab3836	419	configure_gr_followed_by_clear(tgen, topo, input_dict, tc_name, dut="r2", peer="r3")
771ac547 A	420
	421	step("verify Graceful restart at R2")
	422	for addr_type in ADDR_TYPES:
84ab3836 KK	423	result = verify_graceful_restart(
	424	tgen, topo, addr_type, input_dict, dut="r2", peer="r3"
	425	)
	426	assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
771ac547 A	427
	428	step("verify Graceful restart at R3")
	429	for addr_type in ADDR_TYPES:
84ab3836 KK	430	result = verify_graceful_restart(
	431	tgen, topo, addr_type, input_dict, dut="r3", peer="r2"
	432	)
	433	assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
771ac547 A	434
	435	step("Configuring Graceful-restart-disable at R3")
	436	input_dict = {
	437	"r2": {
771ac547 A	438	"bgp": {
	439	"local_as": "200",
	440	"graceful-restart": {
	441	"graceful-restart": False,
84ab3836	442	},
771ac547 A	443	}
	444	},
	445	"r3": {
84ab3836 KK	446	"bgp": {"local_as": "300", "graceful-restart": {"graceful-restart": False}}
84ab3836 KK	447	},
771ac547	448	}
84ab3836	449	configure_gr_followed_by_clear(tgen, topo, input_dict, tc_name, dut="r3", peer="r2")
771ac547 A	450
	451	step("Verify Graceful-restart-disable at R3")
	452	for addr_type in ADDR_TYPES:
84ab3836 KK	453	result = verify_graceful_restart(
	454	tgen, topo, addr_type, input_dict, dut="r3", peer="r2"
	455	)
	456	assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
771ac547 A	457
	458	for iteration in range(5):
	459	step("graceful-restart-disable:True at R3")
	460	input_dict = {
84ab3836 KK	461	"r3": {
	462	"bgp": {
	463	"graceful-restart": {
	464	"graceful-restart-disable": True,
	465	}
771ac547 A	466	}
771ac547 A	467	}
771ac547	468	}
84ab3836 KK	469	configure_gr_followed_by_clear(
	470	tgen, topo, input_dict, tc_name, dut="r3", peer="r2"
	471	)
771ac547 A	472
771ac547 A	473	step("Verifying Routes at R2 on enabling GRD")
84ab3836	474	dut = "r2"
771ac547	475	for addr_type in ADDR_TYPES:
84ab3836	476	input_dict_1 = {"r2": topo["routers"]["r2"]}
771ac547	477	result = verify_bgp_rib(tgen, addr_type, dut, input_dict_1)
84ab3836 KK	478	assert result is True, "Testcase {} :Failed \n Error {}".format(
	479	tc_name, result
	480	)
771ac547 A	481
	482	step("Verify stale Routes in Router R2 enabling GRD")
	483	for addr_type in ADDR_TYPES:
	484	dut = "r2"
	485	protocol = "bgp"
	486	verify_nh_for_static_rtes = {
	487	"r3": {
	488	"static_routes": [
771ac547 A	489	{
	490	"network": [NETWORK1_1[addr_type]],
	491	"no_of_ip": 2,
84ab3836	492	"vrf": "RED",
771ac547 A	493	}
	494	]
	495	}
	496	}
	497	bgp_rib_next_hops = verify_stale_routes_list(
84ab3836 KK	498	tgen, addr_type, dut, verify_nh_for_static_rtes
	499	)
	500	assert (
	501	len(bgp_rib_next_hops) == 1
	502	) is True, "Testcase {} : Failed \n Error: {}".format(
	503	tc_name, bgp_rib_next_hops, expected=True
	504	)
771ac547 A	505
	506	step("graceful-restart-disable:False at R3")
	507	input_dict = {
84ab3836 KK	508	"r3": {
	509	"bgp": {
	510	"graceful-restart": {
	511	"graceful-restart-disable": False,
	512	}
771ac547 A	513	}
771ac547 A	514	}
771ac547	515	}
84ab3836 KK	516	configure_gr_followed_by_clear(
	517	tgen, topo, input_dict, tc_name, dut="r3", peer="r2"
	518	)
771ac547 A	519
771ac547 A	520	step("Verifying Routes at R2 on disabling GRD")
84ab3836	521	dut = "r2"
771ac547	522	for addr_type in ADDR_TYPES:
84ab3836	523	input_dict_1 = {"r2": topo["routers"]["r2"]}
771ac547	524	result = verify_bgp_rib(tgen, addr_type, dut, input_dict_1)
84ab3836 KK	525	assert result is True, "Testcase {} :Failed \n Error {}".format(
	526	tc_name, result
	527	)
771ac547 A	528
	529	step("Verify stale Routes in Router R2 on disabling GRD")
	530	for addr_type in ADDR_TYPES:
	531	dut = "r2"
	532	protocol = "bgp"
	533	verify_nh_for_static_rtes = {
	534	"r3": {
	535	"static_routes": [
771ac547 A	536	{
	537	"network": [NETWORK1_1[addr_type]],
	538	"no_of_ip": 2,
84ab3836	539	"vrf": "RED",
771ac547 A	540	}
	541	]
	542	}
	543	}
84ab3836 KK	544	bgp_rib_next_hops = verify_stale_routes_list(
	545	tgen, addr_type, dut, verify_nh_for_static_rtes
	546	)
	547
	548	stale_route_status = len(bgp_rib_next_hops) == 1
	549	assert (
	550	stale_route_status is True
	551	), "Testcase {} : Failed \n Error: {}".format(
	552	tc_name, stale_route_status, expected=True
	553	)
771ac547	554	write_test_footer(tc_name)