verify_gr_address_family,
modify_bgp_config_when_bgpd_down,
verify_graceful_restart_timers,
+ verify_bgp_convergence_from_running_config,
)
from lib.common_config import (
BGP_CONVERGENCE = False
GR_RESTART_TIMER = 5
GR_SELECT_DEFER_TIMER = 5
+GR_STALEPATH_TIMER = 5
PREFERRED_NEXT_HOP = "link_local"
NEXT_HOP_4 = ["192.168.1.1", "192.168.4.2"]
NEXT_HOP_6 = ["fd00:0:0:1::1", "fd00:0:0:4::2"]
"""
# Required linux kernel version for this suite to run.
- result = required_linux_kernel_version("4.15")
+ result = required_linux_kernel_version("4.16")
if result is not True:
pytest.skip("Kernel requirements are not met")
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, topo)
+ assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
+
return True
clear_bgp(tgen, addr_type, "r1")
clear_bgp(tgen, addr_type, "r3")
+ result = verify_bgp_convergence_from_running_config(tgen, topo)
+ assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
+
for addr_type in ADDR_TYPES:
result = verify_graceful_restart(
tgen, topo, addr_type, input_dict, dut="r1", peer="r3"
tc_name, result
)
- result = verify_bgp_convergence(tgen, topo)
+ result = verify_bgp_convergence_from_running_config(tgen, topo)
assert result is True, "Testcase {} : Failed \n Error {}".format(
tc_name, result
)
reset_config_on_routers(tgen)
logger.info(
- " Test Case : BGP_GR_UTP_29"
+ " Test Case : test_BGP_GR_chaos_29"
" BGP GR [Helper Mode]R3-----R1[Restart Mode]"
" and [restart-time 150]R1 initialized"
)
# Kill BGPd daemon on R1
kill_router_daemons(tgen, "r1", ["bgpd"])
- # Waiting for 120 sec
logger.info("[Step 3] : Wait for {} seconds..".format(GR_RESTART_TIMER))
- # Waiting for 120 sec
+ # Waiting for GR_RESTART_TIMER
sleep(GR_RESTART_TIMER)
for addr_type in ADDR_TYPES:
else:
next_hop_6 = NEXT_HOP_6[1]
- result = verify_rib(tgen, addr_type, dut, input_dict_2, next_hop_6)
+ result = verify_rib(tgen, addr_type, dut, input_dict_2, next_hop_6,
+ expected=False)
+ assert result is not True,\
+ "Testcase {} :Failed \n Error {}". \
+ format(tc_name, result)
+ logger.info(" Expected behavior: {}".\
+ format(result))
logger.info("[Step 4] : Start BGPd daemon on R1 and R4..")
for addr_type in ADDR_TYPES:
clear_bgp(tgen, addr_type, "r1")
+ result = verify_bgp_convergence_from_running_config(tgen, topo)
+ assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
+
for addr_type in ADDR_TYPES:
# Verify f-bit after starting BGPd daemon
result = verify_f_bit(
# Kill BGPd daemon on R1
kill_router_daemons(tgen, "r1", ["bgpd"])
- # Waiting for 120 sec
+ # Waiting for GR_RESTART_TIMER
logger.info("Waiting for {} sec..".format(GR_RESTART_TIMER))
sleep(GR_RESTART_TIMER)
logger.info(" Expected behavior: {}".format(result))
# Verifying RIB routes
- result = verify_rib(tgen, addr_type, dut, input_dict_1)
+ result = verify_rib(tgen, addr_type, dut, input_dict_1, expected=False)
assert result is not True, "Testcase {} : Failed \n Error {}".format(
tc_name, result
)
write_test_footer(tc_name)
+def test_BGP_GR_15_p2(request):
+ """
+ Test Objective : Test GR scenarios by enabling Graceful Restart
+ for multiple address families..
+ """
+
+ tgen = get_topogen()
+ tc_name = request.node.name
+ write_test_header(tc_name)
+
+ # Check router status
+ check_router_status(tgen)
+
+ # Don't run this test if we have any failure.
+ if tgen.routers_have_failure():
+ pytest.skip(tgen.errors)
+
+ # Creating configuration from JSON
+ reset_config_on_routers(tgen)
+
+ # Configure graceful-restart
+ input_dict = {
+ "r1": {
+ "bgp": {
+ "address_family": {
+ "ipv4": {
+ "unicast": {
+ "neighbor": {
+ "r6": {"dest_link": {"r1": {"graceful-restart": True}}}
+ }
+ }
+ },
+ "ipv6": {
+ "unicast": {
+ "neighbor": {
+ "r6": {"dest_link": {"r1": {"graceful-restart": True}}}
+ }
+ }
+ },
+ }
+ }
+ },
+ "r6": {
+ "bgp": {
+ "address_family": {
+ "ipv4": {
+ "unicast": {
+ "neighbor": {
+ "r1": {
+ "dest_link": {
+ "r6": {"graceful-restart-helper": True}
+ }
+ }
+ }
+ }
+ },
+ "ipv6": {
+ "unicast": {
+ "neighbor": {
+ "r1": {
+ "dest_link": {
+ "r6": {"graceful-restart-helper": True}
+ }
+ }
+ }
+ }
+ },
+ }
+ }
+ },
+ }
+
+ configure_gr_followed_by_clear(tgen, topo, input_dict, tc_name, dut="r1", peer="r6")
+
+ for addr_type in ADDR_TYPES:
+ result = verify_graceful_restart(
+ tgen, topo, addr_type, input_dict, dut="r1", peer="r6"
+ )
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ logger.info(
+ "[Step 2] : Test Setup "
+ "[Helper Mode]R6-----R1[Restart Mode]"
+ "--------R2[Helper Mode] Initilized"
+ )
+
+ # Configure graceful-restart
+ input_dict = {
+ "r1": {
+ "bgp": {
+ "address_family": {
+ "ipv4": {
+ "unicast": {
+ "neighbor": {
+ "r2": {"dest_link": {"r1": {"graceful-restart": True}}}
+ }
+ }
+ },
+ "ipv6": {
+ "unicast": {
+ "neighbor": {
+ "r2": {"dest_link": {"r1": {"graceful-restart": True}}}
+ }
+ }
+ },
+ }
+ }
+ },
+ "r2": {
+ "bgp": {
+ "address_family": {
+ "ipv4": {
+ "unicast": {
+ "neighbor": {
+ "r1": {
+ "dest_link": {
+ "r2": {"graceful-restart-helper": True}
+ }
+ }
+ }
+ }
+ },
+ "ipv6": {
+ "unicast": {
+ "neighbor": {
+ "r1": {
+ "dest_link": {
+ "r2": {"graceful-restart-helper": True}
+ }
+ }
+ }
+ }
+ },
+ }
+ }
+ },
+ }
+
+ configure_gr_followed_by_clear(tgen, topo, input_dict, tc_name, dut="r1", peer="r2")
+
+ for addr_type in ADDR_TYPES:
+ result = verify_graceful_restart(
+ tgen, topo, addr_type, input_dict, dut="r1", peer="r2"
+ )
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying BGP RIB routes
+ dut = "r6"
+ input_dict_1 = {key: topo["routers"][key] for key in ["r1"]}
+ result = verify_bgp_rib(tgen, addr_type, dut, input_dict_1)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying RIB routes before shutting down BGPd daemon
+ result = verify_rib(tgen, addr_type, dut, input_dict_1)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying BGP RIB routes
+ dut = "r6"
+ input_dict_2 = {key: topo["routers"][key] for key in ["r2"]}
+ result = verify_bgp_rib(tgen, addr_type, dut, input_dict_2)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying RIB routes before shutting down BGPd daemon
+ result = verify_rib(tgen, addr_type, dut, input_dict_2)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Kill BGPd daemon on R1
+ kill_router_daemons(tgen, "r1", ["bgpd"])
+
+ for addr_type in ADDR_TYPES:
+ # Verifying BGP RIB routes
+ dut = "r6"
+ input_dict_1 = {key: topo["routers"][key] for key in ["r1"]}
+ result = verify_bgp_rib(tgen, addr_type, dut, input_dict_1)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying RIB routes before shutting down BGPd daemon
+ result = verify_rib(tgen, addr_type, dut, input_dict_1)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying BGP RIB routes
+ dut = "r6"
+ input_dict_2 = {key: topo["routers"][key] for key in ["r2"]}
+ result = verify_bgp_rib(tgen, addr_type, dut, input_dict_2)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying RIB routes before shutting down BGPd daemon
+ result = verify_rib(tgen, addr_type, dut, input_dict_2)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Start BGPd daemon on R1
+ start_router_daemons(tgen, "r1", ["bgpd"])
+
+ for addr_type in ADDR_TYPES:
+ result = verify_bgp_convergence(tgen, topo)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying BGP RIB routes
+ dut = "r6"
+ input_dict_1 = {key: topo["routers"][key] for key in ["r1"]}
+ result = verify_bgp_rib(tgen, addr_type, dut, input_dict_1)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying RIB routes before shutting down BGPd daemon
+ result = verify_rib(tgen, addr_type, dut, input_dict_1)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying BGP RIB routes
+ dut = "r6"
+ input_dict_2 = {key: topo["routers"][key] for key in ["r2"]}
+ result = verify_bgp_rib(tgen, addr_type, dut, input_dict_2)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying RIB routes before shutting down BGPd daemon
+ result = verify_rib(tgen, addr_type, dut, input_dict_2)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ write_test_footer(tc_name)
+
+
+def BGP_GR_TC_7_p1(request):
+ """
+ Verify that BGP restarting node deletes all the routes received from peer
+ if BGP Graceful capability is not present in BGP Open message from the
+ peer
+ """
+
+ tgen = get_topogen()
+ tc_name = request.node.name
+ write_test_header(tc_name)
+
+ # Check router status
+ check_router_status(tgen)
+
+ # Don't run this test if we have any failure.
+ if tgen.routers_have_failure():
+ pytest.skip(tgen.errors)
+
+ # Creating configuration from JSON
+ reset_config_on_routers(tgen)
+
+ logger.info(
+ " Verify route download to RIB: BGP_GR_TC_7 >> "
+ "BGP GR [Helper Mode]R3-----R1[Restart Mode] "
+ )
+
+ # Configure graceful-restart
+ input_dict = {
+ "r3": {
+ "bgp": {
+ "address_family": {
+ "ipv4": {
+ "unicast": {
+ "neighbor": {
+ "r1": {
+ "dest_link": {
+ "r3": {"graceful-restart-helper": True}
+ }
+ }
+ }
+ }
+ },
+ "ipv6": {
+ "unicast": {
+ "neighbor": {
+ "r1": {
+ "dest_link": {
+ "r3": {"graceful-restart-helper": True}
+ }
+ }
+ }
+ }
+ },
+ }
+ }
+ },
+ "r1": {
+ "bgp": {
+ "graceful-restart": {
+ "graceful-restart": True,
+ "preserve-fw-state": True,
+ },
+ "address_family": {
+ "ipv4": {
+ "unicast": {
+ "neighbor": {
+ "r3": {"dest_link": {"r1": {"graceful-restart": True}}}
+ }
+ }
+ },
+ "ipv6": {
+ "unicast": {
+ "neighbor": {
+ "r3": {"dest_link": {"r1": {"graceful-restart": True}}}
+ }
+ }
+ },
+ },
+ }
+ },
+ }
+
+ configure_gr_followed_by_clear(tgen, topo, input_dict, tc_name, dut="r1", peer="r3")
+
+ for addr_type in ADDR_TYPES:
+ result = verify_graceful_restart(
+ tgen, topo, addr_type, input_dict, dut="r1", peer="r3"
+ )
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying BGP RIB routes received from router R1
+ dut = "r1"
+ input_dict_1 = {key: topo["routers"][key] for key in ["r3"]}
+ result = verify_bgp_rib(tgen, addr_type, dut, input_dict_1)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying RIB routes
+ result = verify_rib(tgen, addr_type, dut, input_dict_1)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ logger.info("R1 goes for reload")
+ kill_router_daemons(tgen, "r1", ["bgpd"])
+
+ # Change the configuration on router R1
+ input_dict_2 = {
+ "r3": {
+ "bgp": {
+ "address_family": {
+ "ipv4": {
+ "unicast": {
+ "neighbor": {
+ "r1": {
+ "dest_link": {
+ "r3": {"graceful-restart-disable": True}
+ }
+ }
+ }
+ }
+ },
+ "ipv6": {
+ "unicast": {
+ "neighbor": {
+ "r1": {
+ "dest_link": {
+ "r3": {"graceful-restart-disable": True}
+ }
+ }
+ }
+ }
+ },
+ }
+ }
+ }
+ }
+
+ result = create_router_bgp(tgen, topo, input_dict_2)
+ assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
+
+ # Change the configuration on R1
+ network = {"ipv4": "103.0.20.1/32", "ipv6": "3::1/128"}
+ for addr_type in ADDR_TYPES:
+ input_dict_2 = {
+ "r3": {
+ "bgp": {
+ "address_family": {
+ addr_type: {
+ "unicast": {
+ "advertise_networks": [
+ {
+ "network": network[addr_type],
+ "no_of_network": 5,
+ "delete": True,
+ }
+ ]
+ }
+ }
+ }
+ }
+ }
+ }
+
+ result = create_router_bgp(tgen, topo, input_dict_2)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ logger.info("R1 is about to come up now")
+ start_router_daemons(tgen, "r1", ["bgpd"])
+ logger.info("R1 is UP Now")
+
+ # Wait for RIB stale timeout
+ logger.info("Verify routes are not present" "in restart router")
+
+ for addr_type in ADDR_TYPES:
+ # Verifying RIB routes
+ dut = "r1"
+ input_dict_1 = {key: topo["routers"][key] for key in ["r3"]}
+ result = verify_rib(tgen, addr_type, dut, input_dict_1, expected=False)
+ assert result is not True, "Testcase {} :Failed \n Error {}".format(
+ tc_name, result
+ )
+
+ write_test_footer(tc_name)
+
+
+def test_BGP_GR_TC_23_p1(request):
+ """
+ Verify that helper routers are deleting stale routes after stale route
+ timer's expiry. If all the routes are not received from restating node
+ after restart.
+ """
+
+ tgen = get_topogen()
+ tc_name = request.node.name
+ write_test_header(tc_name)
+
+ # Check router status
+ check_router_status(tgen)
+
+ # Don't run this test if we have any failure.
+ if tgen.routers_have_failure():
+ pytest.skip(tgen.errors)
+
+ # Creating configuration from JSON
+ reset_config_on_routers(tgen)
+
+ logger.info(
+ "Verify Stale Routes are deleted on helper: BGP_GR_TC_23 >> "
+ "BGP GR [Helper Mode]R1-----R2[Restart Mode] "
+ )
+
+ # Configure graceful-restart
+ input_dict = {
+ "r1": {
+ "bgp": {
+ "graceful-restart": {"timer": {"stalepath-time": GR_STALEPATH_TIMER}},
+ "address_family": {
+ "ipv4": {
+ "unicast": {
+ "neighbor": {
+ "r2": {
+ "dest_link": {
+ "r1": {"graceful-restart-helper": True}
+ }
+ }
+ }
+ }
+ },
+ "ipv6": {
+ "unicast": {
+ "neighbor": {
+ "r2": {
+ "dest_link": {
+ "r1": {"graceful-restart-helper": True}
+ }
+ }
+ }
+ }
+ },
+ },
+ }
+ },
+ "r2": {
+ "bgp": {
+ "graceful-restart": {
+ "graceful-restart": True,
+ "preserve-fw-state": True,
+ },
+ "address_family": {
+ "ipv4": {
+ "unicast": {
+ "neighbor": {
+ "r1": {"dest_link": {"r2": {"graceful-restart": True}}}
+ }
+ }
+ },
+ "ipv6": {
+ "unicast": {
+ "neighbor": {
+ "r1": {"dest_link": {"r2": {"graceful-restart": True}}}
+ }
+ }
+ },
+ },
+ }
+ },
+ }
+
+ configure_gr_followed_by_clear(tgen, topo, input_dict, tc_name, dut="r1", peer="r2")
+
+ for addr_type in ADDR_TYPES:
+ result = verify_graceful_restart(
+ tgen, topo, addr_type, input_dict, dut="r1", peer="r2"
+ )
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying BGP RIB routes received from router R1
+ dut = "r1"
+ input_dict_1 = {key: topo["routers"][key] for key in ["r2"]}
+ result = verify_bgp_rib(tgen, addr_type, dut, input_dict_1)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying RIB routes
+ result = verify_rib(tgen, addr_type, dut, input_dict_1)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ logger.info("R2 goes for reload")
+ kill_router_daemons(tgen, "r2", ["bgpd"])
+
+ # Modify configuration to delete routes and include disable-eor
+ input_dict_3 = {"r2": {"bgp": {"graceful-restart": {"disable-eor": True}}}}
+
+ result = modify_bgp_config_when_bgpd_down(tgen, topo, input_dict_3)
+
+ # Modify configuration to delete routes and include disable-eor
+ network = {"ipv4": "102.0.20.1/32", "ipv6": "2::1/128"}
+ for addr_type in ADDR_TYPES:
+ input_dict_3 = {
+ "r2": {
+ "bgp": {
+ "address_family": {
+ addr_type: {
+ "unicast": {
+ "advertise_networks": [
+ {
+ "network": network[addr_type],
+ "no_of_network": 3,
+ "delete": True,
+ }
+ ]
+ }
+ }
+ }
+ }
+ }
+ }
+
+ result = modify_bgp_config_when_bgpd_down(tgen, topo, input_dict_3)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ logger.info("BGPd comes up for r2")
+ start_router_daemons(tgen, "r2", ["bgpd"])
+
+ # Wait for stalepath timer
+ logger.info("Waiting for stalepath timer({} sec..)".format(GR_STALEPATH_TIMER))
+ sleep(GR_STALEPATH_TIMER)
+
+ for addr_type in ADDR_TYPES:
+ clear_bgp(tgen, addr_type, "r2")
+
+ # Verifying RIB routes
+ dut = "r1"
+ network = {"ipv4": "102.0.20.4/32", "ipv6": "2::4/128"}
+ for addr_type in ADDR_TYPES:
+ input_dict_1 = {
+ "r1": {
+ "bgp": {
+ "address_family": {
+ addr_type: {
+ "unicast": {
+ "advertise_networks": [
+ {"network": network[addr_type], "no_of_network": 2}
+ ]
+ }
+ }
+ }
+ }
+ }
+ }
+
+ # Verify EOR on helper router
+ result = verify_eor(
+ tgen, topo, addr_type, input_dict, dut="r1", peer="r2", expected=False
+ )
+ assert result is not True, (
+ "Testcase " + tc_name + " :Failed \n Error: {}".format(result)
+ )
+
+ # Verifying BGP RIB routes received from router R1
+ dut = "r1"
+ result = verify_bgp_rib(tgen, addr_type, dut, input_dict_1)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ write_test_footer(tc_name)
+
+
+def test_BGP_GR_20_p1(request):
+ """
+ Test Objective : Verify that GR routers delete all the routes
+ received from a node if both the routers are configured as GR
+ helper node
+ """
+
+ tgen = get_topogen()
+ tc_name = request.node.name
+ write_test_header(tc_name)
+
+ # Check router status
+ check_router_status(tgen)
+
+ # Don't run this test if we have any failure.
+ if tgen.routers_have_failure():
+ pytest.skip(tgen.errors)
+
+ # Creating configuration from JSON
+ reset_config_on_routers(tgen)
+
+ logger.info(
+ "[Step 1] : Test Setup " "[Restart Mode]R3-----R1[Restart Mode] Initilized"
+ )
+
+ # Configure graceful-restart
+ input_dict = {
+ "r1": {
+ "bgp": {
+ "address_family": {
+ "ipv4": {
+ "unicast": {
+ "neighbor": {
+ "r3": {
+ "dest_link": {
+ "r1": {"graceful-restart-helper": True}
+ }
+ }
+ }
+ }
+ },
+ "ipv6": {
+ "unicast": {
+ "neighbor": {
+ "r3": {
+ "dest_link": {
+ "r1": {"graceful-restart-helper": True}
+ }
+ }
+ }
+ }
+ },
+ }
+ }
+ },
+ "r3": {
+ "bgp": {
+ "address_family": {
+ "ipv4": {
+ "unicast": {
+ "neighbor": {
+ "r1": {
+ "dest_link": {
+ "r3": {"graceful-restart-helper": True}
+ }
+ }
+ }
+ }
+ },
+ "ipv6": {
+ "unicast": {
+ "neighbor": {
+ "r1": {
+ "dest_link": {
+ "r3": {"graceful-restart-helper": True}
+ }
+ }
+ }
+ }
+ },
+ }
+ }
+ },
+ }
+
+ configure_gr_followed_by_clear(tgen, topo, input_dict, tc_name, dut="r1", peer="r3")
+
+ for addr_type in ADDR_TYPES:
+ result = verify_graceful_restart(
+ tgen, topo, addr_type, input_dict, dut="r1", peer="r3"
+ )
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying BGP RIB routes
+ dut = "r3"
+ input_dict_1 = {key: topo["routers"][key] for key in ["r1"]}
+ result = verify_bgp_rib(tgen, addr_type, dut, input_dict_1)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying RIB routes before shutting down BGPd daemon
+ result = verify_rib(tgen, addr_type, dut, input_dict_1)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Kill BGPd daemon on R1
+ kill_router_daemons(tgen, "r1", ["bgpd"])
+
+ for addr_type in ADDR_TYPES:
+ # Verifying BGP RIB routes
+ dut = "r3"
+ input_dict_1 = {key: topo["routers"][key] for key in ["r1"]}
+ result = verify_bgp_rib(tgen, addr_type, dut, input_dict_1, expected=False)
+ assert result is not True, "Testcase {} :Failed \n Error {}".format(
+ tc_name, result
+ )
+ logger.info(" Expected behavior: {}".format(result))
+
+ # Verifying RIB routes before shutting down BGPd daemon
+ result = verify_rib(tgen, addr_type, dut, input_dict_1, expected=False)
+ assert result is not True, "Testcase {} :Failed \n Error {}".format(
+ tc_name, result
+ )
+ logger.info(" Expected behavior: {}".format(result))
+
+ # Start BGPd daemon on R1
+ start_router_daemons(tgen, "r1", ["bgpd"])
+
+ for addr_type in ADDR_TYPES:
+ result = verify_bgp_convergence(tgen, topo)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying BGP RIB routes
+ dut = "r3"
+ input_dict_1 = {key: topo["routers"][key] for key in ["r1"]}
+ result = verify_bgp_rib(tgen, addr_type, dut, input_dict_1)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying RIB routes before shutting down BGPd daemon
+ result = verify_rib(tgen, addr_type, dut, input_dict_1)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ write_test_footer(tc_name)
+
+
+def test_BGP_GR_21_p2(request):
+ """
+ Test Objective : VVerify BGP-GR feature when helper node is
+ a transit router for it's eBGP peers.
+ """
+
+ tgen = get_topogen()
+ tc_name = request.node.name
+ write_test_header(tc_name)
+
+ # Check router status
+ check_router_status(tgen)
+
+ # Don't run this test if we have any failure.
+ if tgen.routers_have_failure():
+ pytest.skip(tgen.errors)
+
+ # Creating configuration from JSON
+ reset_config_on_routers(tgen)
+
+ logger.info(
+ "[Step 1] : Test Setup " "[Helper Mode]R6-----R1[Restart Mode] Initilized"
+ )
+
+ # Configure graceful-restart
+ input_dict = {
+ "r1": {
+ "bgp": {
+ "address_family": {
+ "ipv4": {
+ "unicast": {
+ "neighbor": {
+ "r6": {
+ "dest_link": {
+ "r1": {"graceful-restart-disable": True}
+ }
+ }
+ }
+ }
+ },
+ "ipv6": {
+ "unicast": {
+ "neighbor": {
+ "r6": {
+ "dest_link": {
+ "r1": {"graceful-restart-disable": True}
+ }
+ }
+ }
+ }
+ },
+ }
+ }
+ },
+ "r6": {
+ "bgp": {
+ "address_family": {
+ "ipv4": {
+ "unicast": {
+ "neighbor": {
+ "r1": {
+ "dest_link": {
+ "r6": {"graceful-restart-helper": True}
+ }
+ }
+ }
+ }
+ },
+ "ipv6": {
+ "unicast": {
+ "neighbor": {
+ "r1": {
+ "dest_link": {
+ "r6": {"graceful-restart-helper": True}
+ }
+ }
+ }
+ }
+ },
+ }
+ }
+ },
+ }
+
+ configure_gr_followed_by_clear(tgen, topo, input_dict, tc_name, dut="r1", peer="r6")
+
+ for addr_type in ADDR_TYPES:
+ result = verify_graceful_restart(
+ tgen, topo, addr_type, input_dict, dut="r1", peer="r6"
+ )
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ logger.info(
+ "[Step 2] : Test Setup "
+ "[Restart Mode]R2-----[Helper Mode]R1[Disable Mode]"
+ "--------R6[Helper Mode] Initilized"
+ )
+
+ # Configure graceful-restart
+ input_dict = {
+ "r1": {
+ "bgp": {
+ "address_family": {
+ "ipv4": {
+ "unicast": {
+ "neighbor": {
+ "r2": {
+ "dest_link": {
+ "r1": {"graceful-restart-helper": True}
+ }
+ }
+ }
+ }
+ },
+ "ipv6": {
+ "unicast": {
+ "neighbor": {
+ "r2": {
+ "dest_link": {
+ "r1": {"graceful-restart-helper": True}
+ }
+ }
+ }
+ }
+ },
+ }
+ }
+ },
+ "r2": {"bgp": {"graceful-restart": {"graceful-restart": True,}}},
+ }
+
+ configure_gr_followed_by_clear(tgen, topo, input_dict, tc_name, dut="r1", peer="r2")
+
+ for addr_type in ADDR_TYPES:
+ result = verify_graceful_restart(
+ tgen, topo, addr_type, input_dict, dut="r1", peer="r2"
+ )
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying BGP RIB routes
+ dut = "r6"
+ input_dict_1 = {key: topo["routers"][key] for key in ["r1"]}
+ result = verify_bgp_rib(tgen, addr_type, dut, input_dict_1)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying RIB routes before shutting down BGPd daemon
+ result = verify_rib(tgen, addr_type, dut, input_dict_1)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying BGP RIB routes
+ dut = "r6"
+ input_dict_2 = {key: topo["routers"][key] for key in ["r2"]}
+ result = verify_bgp_rib(tgen, addr_type, dut, input_dict_2)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying RIB routes before shutting down BGPd daemon
+ result = verify_rib(tgen, addr_type, dut, input_dict_2)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Kill BGPd daemon on R1
+ kill_router_daemons(tgen, "r2", ["bgpd"])
+
+ for addr_type in ADDR_TYPES:
+ # Verifying BGP RIB routes
+ dut = "r6"
+ input_dict_1 = {key: topo["routers"][key] for key in ["r1"]}
+ result = verify_bgp_rib(tgen, addr_type, dut, input_dict_1)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying RIB routes before shutting down BGPd daemon
+ result = verify_rib(tgen, addr_type, dut, input_dict_1)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying BGP RIB routes
+ dut = "r6"
+ input_dict_2 = {key: topo["routers"][key] for key in ["r2"]}
+ result = verify_bgp_rib(tgen, addr_type, dut, input_dict_2)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying RIB routes before shutting down BGPd daemon
+ result = verify_rib(tgen, addr_type, dut, input_dict_2)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Start BGPd daemon on R1
+ start_router_daemons(tgen, "r2", ["bgpd"])
+
+ for addr_type in ADDR_TYPES:
+ result = verify_bgp_convergence(tgen, topo)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying BGP RIB routes
+ dut = "r6"
+ input_dict_1 = {key: topo["routers"][key] for key in ["r1"]}
+ result = verify_bgp_rib(tgen, addr_type, dut, input_dict_1)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying RIB routes after bringing up BGPd daemon
+ result = verify_rib(tgen, addr_type, dut, input_dict_1)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying BGP RIB routes
+ dut = "r6"
+ input_dict_2 = {key: topo["routers"][key] for key in ["r2"]}
+ result = verify_bgp_rib(tgen, addr_type, dut, input_dict_2)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying RIB routes before shutting down BGPd daemon
+ result = verify_rib(tgen, addr_type, dut, input_dict_2)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ write_test_footer(tc_name)
+
+
+def test_BGP_GR_22_p2(request):
+ """
+ Test Objective : Verify BGP-GR feature when helper node
+ is a transit router for it's iBGP peers.
+ """
+
+ tgen = get_topogen()
+ tc_name = request.node.name
+ write_test_header(tc_name)
+
+ # Check router status
+ check_router_status(tgen)
+
+ # Don't run this test if we have any failure.
+ if tgen.routers_have_failure():
+ pytest.skip(tgen.errors)
+
+ # Creating configuration from JSON
+ reset_config_on_routers(tgen)
+
+ logger.info(
+ "[Step 1] : Test Setup " "[Helper Mode]R3-----R1[Restart Mode] Initilized"
+ )
+
+ # Configure graceful-restart
+ input_dict = {
+ "r1": {
+ "bgp": {
+ "address_family": {
+ "ipv4": {
+ "unicast": {
+ "neighbor": {
+ "r3": {
+ "dest_link": {
+ "r1": {
+ "graceful-restart-disable": True,
+ "next_hop_self": True,
+ }
+ }
+ }
+ }
+ }
+ },
+ "ipv6": {
+ "unicast": {
+ "neighbor": {
+ "r3": {
+ "dest_link": {
+ "r1": {
+ "graceful-restart-disable": True,
+ "next_hop_self": True,
+ }
+ }
+ }
+ }
+ }
+ },
+ }
+ }
+ },
+ "r3": {
+ "bgp": {
+ "address_family": {
+ "ipv4": {
+ "unicast": {
+ "neighbor": {
+ "r1": {
+ "dest_link": {
+ "r3": {"graceful-restart-helper": True}
+ }
+ }
+ }
+ }
+ },
+ "ipv6": {
+ "unicast": {
+ "neighbor": {
+ "r1": {
+ "dest_link": {
+ "r3": {"graceful-restart-helper": True}
+ }
+ }
+ }
+ }
+ },
+ }
+ }
+ },
+ }
+
+ configure_gr_followed_by_clear(tgen, topo, input_dict, tc_name, dut="r1", peer="r3")
+
+ for addr_type in ADDR_TYPES:
+ result = verify_graceful_restart(
+ tgen, topo, addr_type, input_dict, dut="r1", peer="r3"
+ )
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ logger.info(
+ "[Step 2] : Test Setup "
+ "[Restart Mode]R2-----[Helper Mode]R1[Disable Mode]"
+ "--------R3[Helper Mode] Initilized"
+ )
+
+ # Configure graceful-restart
+ input_dict = {
+ "r1": {
+ "bgp": {
+ "address_family": {
+ "ipv4": {
+ "unicast": {
+ "neighbor": {
+ "r2": {
+ "dest_link": {
+ "r1": {"graceful-restart-helper": True}
+ }
+ }
+ }
+ }
+ },
+ "ipv6": {
+ "unicast": {
+ "neighbor": {
+ "r2": {
+ "dest_link": {
+ "r1": {"graceful-restart-helper": True}
+ }
+ }
+ }
+ }
+ },
+ }
+ }
+ },
+ "r2": {"bgp": {"graceful-restart": {"graceful-restart": True}}},
+ }
+
+ configure_gr_followed_by_clear(tgen, topo, input_dict, tc_name, dut="r1", peer="r2")
+
+ for addr_type in ADDR_TYPES:
+ result = verify_graceful_restart(
+ tgen, topo, addr_type, input_dict, dut="r1", peer="r2"
+ )
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying BGP RIB routes
+ dut = "r3"
+ input_dict_1 = {key: topo["routers"][key] for key in ["r1"]}
+ result = verify_bgp_rib(tgen, addr_type, dut, input_dict_1)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying RIB routes before shutting down BGPd daemon
+ result = verify_rib(tgen, addr_type, dut, input_dict_1)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying BGP RIB routes
+ dut = "r3"
+ input_dict_2 = {key: topo["routers"][key] for key in ["r2"]}
+ result = verify_bgp_rib(tgen, addr_type, dut, input_dict_2)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying RIB routes before shutting down BGPd daemon
+ result = verify_rib(tgen, addr_type, dut, input_dict_2)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Kill BGPd daemon on R1
+ kill_router_daemons(tgen, "r2", ["bgpd"])
+
+ for addr_type in ADDR_TYPES:
+ # Verifying BGP RIB routes
+ dut = "r3"
+ input_dict_1 = {key: topo["routers"][key] for key in ["r1"]}
+ result = verify_bgp_rib(tgen, addr_type, dut, input_dict_1)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying RIB routes before shutting down BGPd daemon
+ result = verify_rib(tgen, addr_type, dut, input_dict_1)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying BGP RIB routes
+ dut = "r3"
+ input_dict_2 = {key: topo["routers"][key] for key in ["r2"]}
+ result = verify_bgp_rib(tgen, addr_type, dut, input_dict_2)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying RIB routes before shutting down BGPd daemon
+ result = verify_rib(tgen, addr_type, dut, input_dict_2)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Start BGPd daemon on R1
+ start_router_daemons(tgen, "r2", ["bgpd"])
+
+ for addr_type in ADDR_TYPES:
+ result = verify_bgp_convergence(tgen, topo)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying BGP RIB routes
+ dut = "r3"
+ input_dict_1 = {key: topo["routers"][key] for key in ["r1"]}
+ result = verify_bgp_rib(tgen, addr_type, dut, input_dict_1)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying RIB routes before shutting down BGPd daemon
+ result = verify_rib(tgen, addr_type, dut, input_dict_1)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying BGP RIB routes
+ dut = "r3"
+ input_dict_2 = {key: topo["routers"][key] for key in ["r2"]}
+ result = verify_bgp_rib(tgen, addr_type, dut, input_dict_2)
+ assert result is True, "Testcase {} :Failed \n Error {}".format(tc_name, result)
+
+ # Verifying RIB routes before shutting down BGPd daemon
+ result = verify_rib(tgen, addr_type, dut, input_dict_2)
+ assert result is 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))