diff options
author | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-04-09 13:16:35 +0000 |
---|---|---|
committer | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-04-09 13:16:35 +0000 |
commit | e2bbf175a2184bd76f6c54ccf8456babeb1a46fc (patch) | |
tree | f0b76550d6e6f500ada964a3a4ee933a45e5a6f1 /tests/topotests/ospf_gr_helper | |
parent | Initial commit. (diff) | |
download | frr-e2bbf175a2184bd76f6c54ccf8456babeb1a46fc.tar.xz frr-e2bbf175a2184bd76f6c54ccf8456babeb1a46fc.zip |
Adding upstream version 9.1.upstream/9.1
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'tests/topotests/ospf_gr_helper')
-rw-r--r-- | tests/topotests/ospf_gr_helper/ospf_gr_helper.json | 119 | ||||
-rw-r--r-- | tests/topotests/ospf_gr_helper/test_ospf_gr_helper1.py | 380 | ||||
-rw-r--r-- | tests/topotests/ospf_gr_helper/test_ospf_gr_helper2.py | 356 | ||||
-rw-r--r-- | tests/topotests/ospf_gr_helper/test_ospf_gr_helper3.py | 310 |
4 files changed, 1165 insertions, 0 deletions
diff --git a/tests/topotests/ospf_gr_helper/ospf_gr_helper.json b/tests/topotests/ospf_gr_helper/ospf_gr_helper.json new file mode 100644 index 0000000..b9ca022 --- /dev/null +++ b/tests/topotests/ospf_gr_helper/ospf_gr_helper.json @@ -0,0 +1,119 @@ +{ + "ipv4base": "10.0.0.0", + "ipv4mask": 24, + "link_ip_start": { + "ipv4": "10.0.0.0", + "v4mask": 24 + }, + "lo_prefix": { + "ipv4": "1.0.", + "v4mask": 32 + }, + "switches": { + "s1": { + "links": { + "r0": { + "ipv4": "17.1.1.2/24", + "ospf": { + "area": "0.0.0.0", + "hello_interval": 1, + "dead_interval": 4, + "priority": 98 + } + }, + "r1": { + "ipv4": "17.1.1.1/24", + "ospf": { + "area": "0.0.0.0", + "hello_interval": 1, + "dead_interval": 4, + "priority": 99 + } + }, + "r2": { + "ipv4": "17.1.1.3/24", + "ospf": { + "area": "0.0.0.0", + "hello_interval": 1, + "dead_interval": 4, + "priority": 0 + } + }, + "r3": { + "ipv4": "17.1.1.4/24", + "ospf": { + "area": "0.0.0.0", + "hello_interval": 1, + "dead_interval": 4, + "priority": 0 + } + } + } + } + }, + "routers": { + "r0": { + "links": { + "lo": { + "ipv4": "auto", + "type": "loopback" + } + }, + "ospf": { + "router_id": "100.1.1.0", + "neighbors": { + "r1": {}, + "r2": {}, + "r3": {} + } + } + }, + "r1": { + "links": { + "lo": { + "ipv4": "auto", + "type": "loopback" + } + }, + "ospf": { + "router_id": "1.1.1.1", + "neighbors": { + "r0": {}, + "r2": {}, + "r3": {} + } + }, + "opq_lsa_hex": "01005e00000570708bd051ef080045c0005cc18b0000015904f711010101e00000050204004801010101000000001e8d0000000000000000000000000001000102090300000001010101800000013bd1002c000100040000070800020001010000000003000411010101" + }, + "r2": { + "links": { + "lo": { + "ipv4": "auto", + "type": "loopback" + } + }, + "ospf": { + "router_id": "100.1.1.2", + "neighbors": { + "r1": {}, + "r0": {} + } + } + }, + "r3": { + "links": { + "lo": { + "ipv4": "auto", + "type": "loopback" + } + }, + "ospf": { + "router_id": "100.1.1.3", + "neighbors": { + "r0": {}, + "r1": {} + } + } + } + } +} diff --git a/tests/topotests/ospf_gr_helper/test_ospf_gr_helper1.py b/tests/topotests/ospf_gr_helper/test_ospf_gr_helper1.py new file mode 100644 index 0000000..7937428 --- /dev/null +++ b/tests/topotests/ospf_gr_helper/test_ospf_gr_helper1.py @@ -0,0 +1,380 @@ +#!/usr/bin/python +# SPDX-License-Identifier: ISC + +# +# Copyright (c) 2021 by VMware, Inc. ("VMware") +# Used Copyright (c) 2018 by Network Device Education Foundation, Inc. +# ("NetDEF") in this file. +# + + +"""OSPF Basic Functionality Automation.""" +import os +import sys +import time +import pytest + +# Save the Current Working Directory to find configuration files. +CWD = os.path.dirname(os.path.realpath(__file__)) +sys.path.append(os.path.join(CWD, "../")) + +# pylint: disable=C0413 +# Import topogen and topotest helpers +from lib.topogen import Topogen, get_topogen + +# Import topoJson from lib, to create topology and initial configuration +from lib.common_config import ( + start_topology, + write_test_header, + write_test_footer, + reset_config_on_routers, + step, + create_interfaces_cfg, + scapy_send_raw_packet, +) + +from lib.topolog import logger +from lib.topojson import build_config_from_json + +from lib.ospf import ( + verify_ospf_neighbor, + clear_ospf, + verify_ospf_gr_helper, + create_router_ospf, +) + +pytestmark = [pytest.mark.ospfd] + +# Global variables +topo = None +Iters = 5 +sw_name = None +intf = None +intf1 = None +pkt = None + +""" +Topology: + + Please view in a fixed-width font such as Courier. + Topo : Broadcast Networks + DUT - HR RR + +---+ +---+ +---+ +---+ + |R0 + +R1 + +R2 + +R3 | + +-+-+ +-+-+ +-+-+ +-+-+ + | | | | + | | | | + --+-----------+--------------+---------------+----- + Ethernet Segment + +Testcases: + +TC1. Verify by default helper support is disabled for FRR ospf +TC2. OSPF GR on Broadcast : Verify DUT enters Helper mode when neighbor + sends grace lsa, helps RR to restart gracefully (RR = DR) +TC3. OSPF GR on Broadcast : Verify DUT enters Helper mode when neighbor + sends grace lsa, helps RR to restart gracefully (RR = BDR) +TC4. OSPF GR on Broadcast : Verify DUT enters Helper mode when neighbor + sends grace lsa, helps RR to restart gracefully (RR = DRother) +TC5. OSPF GR on P2P : Verify DUT enters Helper mode when neighbor sends + grace lsa, helps RR to restart gracefully. +TC6. Verify all the show commands newly introducted as part of ospf + helper support - Json Key verification wrt to show commands. +TC7. Verify helper when grace lsa is received with different configured + value in process level (higher, lower, grace lsa timer above 1800) +TC8. Verify helper functionality when dut is helping RR and new grace lsa + is received from RR. +""" + + +def setup_module(mod): + """ + Sets up the pytest environment + + * `mod`: module name + """ + global topo, intf, intf1, sw_name, pkt + 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 = "{}/ospf_gr_helper.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) + + # Don't run this test if we have any failure. + if tgen.routers_have_failure(): + pytest.skip(tgen.errors) + + ospf_covergence = verify_ospf_neighbor(tgen, topo, lan=True) + assert ospf_covergence is True, "setup_module :Failed \n Error: {}".format( + ospf_covergence + ) + + sw_name = "s1" + intf = topo["routers"]["r0"]["links"][sw_name]["interface"] + intf1 = topo["routers"]["r1"]["links"][sw_name]["interface"] + pkt = topo["routers"]["r1"]["opq_lsa_hex"] + + logger.info("Running setup_module() done") + + +def teardown_module(): + """Teardown the pytest environment""" + + logger.info("Running teardown_module to delete topology") + + tgen = get_topogen() + + try: + # Stop toplogy and Remove tmp files + tgen.stop_topology() + + except OSError: + # OSError exception is raised when mininet tries to stop switch + # though switch is stopped once but mininet tries to stop same + # switch again, where it ended up with exception + pass + + +def delete_ospf(): + """delete ospf process after each test""" + tgen = get_topogen() + step("Delete ospf process") + for rtr in topo["routers"]: + ospf_del = {rtr: {"ospf": {"delete": True}}} + result = create_router_ospf(tgen, topo, ospf_del) + assert result is True, "Testcase: Failed \n Error: {}".format(result) + + +# ################################## +# Test cases start here. +# ################################## + + +def test_ospf_gr_helper_tc1_p0(request): + """Verify by default helper support is disabled for FRR ospf""" + + tc_name = request.node.name + write_test_header(tc_name) + tgen = get_topogen() + + # Don't run this test if we have any failure. + if tgen.routers_have_failure(): + pytest.skip(tgen.errors) + + global topo, intf, intf1, pkt + + step("Bring up the base config as per the topology") + reset_config_on_routers(tgen) + ospf_covergence = verify_ospf_neighbor(tgen, topo, lan=True) + assert ( + ospf_covergence is True + ), "OSPF is not after reset config \n Error: {}".format(ospf_covergence) + + step("Verify that GR helper route is disabled by default to the in the DUT.") + input_dict = { + "helperSupport": "Disabled", + "strictLsaCheck": "Enabled", + "restartSupport": "Planned and Unplanned Restarts", + "supportedGracePeriod": 1800, + } + dut = "r0" + result = verify_ospf_gr_helper(tgen, topo, dut, input_dict) + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + step("Verify that DUT does not enter helper mode upon receiving the grace lsa.") + + # send grace lsa + scapy_send_raw_packet(tgen, topo, "r1", intf1, pkt) + + input_dict = {"activeRestarterCnt": 1} + dut = "r0" + result = verify_ospf_gr_helper(tgen, topo, dut, input_dict, expected=False) + assert ( + result is not True + ), "Testcase {} : Failed. DUT entered helper role \n Error: {}".format( + tc_name, result + ) + + step("Configure graceful restart in the DUT") + ospf_gr_r0 = { + "r0": {"ospf": {"graceful-restart": {"helper enable": [], "opaque": True}}} + } + result = create_router_ospf(tgen, topo, ospf_gr_r0) + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + step("Verify that GR helper route is enabled in the DUT.") + input_dict = { + "helperSupport": "Enabled", + "strictLsaCheck": "Enabled", + "restartSupport": "Planned and Unplanned Restarts", + "supportedGracePeriod": 1800, + } + dut = "r0" + result = verify_ospf_gr_helper(tgen, topo, dut, input_dict) + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + ospf_gr_r1 = { + "r1": {"ospf": {"graceful-restart": {"helper enable": [], "opaque": True}}} + } + result = create_router_ospf(tgen, topo, ospf_gr_r1) + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + step("Perform GR in RR.") + step("Verify that DUT does enter helper mode upon receiving the grace lsa.") + input_dict = {"activeRestarterCnt": 1} + gracelsa_sent = False + repeat = 0 + dut = "r0" + while not gracelsa_sent and repeat < Iters: + gracelsa_sent = scapy_send_raw_packet(tgen, topo, "r1", intf1, pkt) + result = verify_ospf_gr_helper(tgen, topo, dut, input_dict) + if isinstance(result, str): + repeat += 1 + gracelsa_sent = False + + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + step("Unconfigure the GR helper command.") + ospf_gr_r0 = { + "r0": { + "ospf": { + "graceful-restart": { + "helper enable": [], + "opaque": True, + "delete": True, + } + } + } + } + result = create_router_ospf(tgen, topo, ospf_gr_r0) + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + input_dict = {"helperSupport": "Disabled"} + dut = "r0" + result = verify_ospf_gr_helper(tgen, topo, dut, input_dict) + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + step("Configure gr helper using the router id") + ospf_gr_r0 = { + "r0": { + "ospf": {"graceful-restart": {"helper enable": ["1.1.1.1"], "opaque": True}} + } + } + result = create_router_ospf(tgen, topo, ospf_gr_r0) + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + step("Verify that DUT does enter helper mode upon receiving the grace lsa.") + input_dict = {"activeRestarterCnt": 1} + gracelsa_sent = False + repeat = 0 + dut = "r0" + while not gracelsa_sent and repeat < Iters: + gracelsa_sent = scapy_send_raw_packet(tgen, topo, "r1", intf1, pkt) + result = verify_ospf_gr_helper(tgen, topo, dut, input_dict) + if isinstance(result, str): + repeat += 1 + gracelsa_sent = False + + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + step("Un Configure gr helper using the router id") + ospf_gr_r0 = { + "r0": { + "ospf": { + "graceful-restart": { + "helper enable": ["1.1.1.1"], + "opaque": True, + "delete": True, + } + } + } + } + result = create_router_ospf(tgen, topo, ospf_gr_r0) + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + step("Verify that GR helper router is disabled in the DUT for router id x.x.x.x") + input_dict = {"enabledRouterIds": [{"routerId": "1.1.1.1"}]} + dut = "r0" + result = verify_ospf_gr_helper(tgen, topo, dut, input_dict, expected=False) + assert ( + result is not True + ), "Testcase {} : Failed, Helper role enabled for RR\n Error: {}".format( + tc_name, result + ) + delete_ospf() + write_test_footer(tc_name) + + +def test_ospf_gr_helper_tc2_p0(request): + """ + OSPF GR on Broadcast : Verify DUT enters Helper mode when neighbor + sends grace lsa, helps RR to restart gracefully (RR = DR) + """ + tc_name = request.node.name + write_test_header(tc_name) + tgen = get_topogen() + + # Don't run this test if we have any failure. + if tgen.routers_have_failure(): + pytest.skip(tgen.errors) + + global topo, intf, intf1, pkt + + step("Bring up the base config as per the topology") + step( + "Configure DR priority as 99 in RR , DUT dr priority = 98 " + "& reset ospf process in all the routers" + ) + reset_config_on_routers(tgen) + ospf_covergence = verify_ospf_neighbor(tgen, topo, lan=True) + assert ( + ospf_covergence is True + ), "OSPF is not after reset config \n Error: {}".format(ospf_covergence) + ospf_gr_r0 = { + "r0": {"ospf": {"graceful-restart": {"helper enable": [], "opaque": True}}} + } + result = create_router_ospf(tgen, topo, ospf_gr_r0) + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + ospf_gr_r1 = { + "r1": {"ospf": {"graceful-restart": {"helper enable": [], "opaque": True}}} + } + result = create_router_ospf(tgen, topo, ospf_gr_r1) + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + step("Verify that DUT enters into helper mode.") + + input_dict = {"activeRestarterCnt": 1} + gracelsa_sent = False + repeat = 0 + dut = "r0" + while not gracelsa_sent and repeat < Iters: + gracelsa_sent = scapy_send_raw_packet(tgen, topo, "r1", intf1, pkt) + result = verify_ospf_gr_helper(tgen, topo, dut, input_dict) + if isinstance(result, str): + repeat += 1 + gracelsa_sent = False + + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + delete_ospf() + write_test_footer(tc_name) + + +if __name__ == "__main__": + args = ["-s"] + sys.argv[1:] + sys.exit(pytest.main(args)) diff --git a/tests/topotests/ospf_gr_helper/test_ospf_gr_helper2.py b/tests/topotests/ospf_gr_helper/test_ospf_gr_helper2.py new file mode 100644 index 0000000..46c0da3 --- /dev/null +++ b/tests/topotests/ospf_gr_helper/test_ospf_gr_helper2.py @@ -0,0 +1,356 @@ +#!/usr/bin/python +# SPDX-License-Identifier: ISC + +# +# Copyright (c) 2021 by VMware, Inc. ("VMware") +# Used Copyright (c) 2018 by Network Device Education Foundation, Inc. +# ("NetDEF") in this file. +# + + +"""OSPF Basic Functionality Automation.""" +import os +import sys +import time +import pytest + +# Save the Current Working Directory to find configuration files. +CWD = os.path.dirname(os.path.realpath(__file__)) +sys.path.append(os.path.join(CWD, "../")) + +# pylint: disable=C0413 +# Import topogen and topotest helpers +from lib.topogen import Topogen, get_topogen + +# Import topoJson from lib, to create topology and initial configuration +from lib.common_config import ( + start_topology, + write_test_header, + write_test_footer, + reset_config_on_routers, + step, + create_interfaces_cfg, + scapy_send_raw_packet, +) + +from lib.topolog import logger +from lib.topojson import build_config_from_json + +from lib.ospf import ( + verify_ospf_neighbor, + clear_ospf, + verify_ospf_gr_helper, + create_router_ospf, +) + +pytestmark = [pytest.mark.ospfd] + +# Global variables +topo = None +Iters = 5 +sw_name = None +intf = None +intf1 = None +pkt = None + +""" +Topology: + + Please view in a fixed-width font such as Courier. + Topo : Broadcast Networks + DUT - HR RR + +---+ +---+ +---+ +---+ + |R0 + +R1 + +R2 + +R3 | + +-+-+ +-+-+ +-+-+ +-+-+ + | | | | + | | | | + --+-----------+--------------+---------------+----- + Ethernet Segment + +Testcases: + +TC1. Verify by default helper support is disabled for FRR ospf +TC2. OSPF GR on Broadcast : Verify DUT enters Helper mode when neighbor + sends grace lsa, helps RR to restart gracefully (RR = DR) +TC3. OSPF GR on Broadcast : Verify DUT enters Helper mode when neighbor + sends grace lsa, helps RR to restart gracefully (RR = BDR) +TC4. OSPF GR on Broadcast : Verify DUT enters Helper mode when neighbor + sends grace lsa, helps RR to restart gracefully (RR = DRother) +TC5. OSPF GR on P2P : Verify DUT enters Helper mode when neighbor sends + grace lsa, helps RR to restart gracefully. +TC6. Verify all the show commands newly introducted as part of ospf + helper support - Json Key verification wrt to show commands. +TC7. Verify helper when grace lsa is received with different configured + value in process level (higher, lower, grace lsa timer above 1800) +TC8. Verify helper functionality when dut is helping RR and new grace lsa + is received from RR. +""" + + +def setup_module(mod): + """ + Sets up the pytest environment + + * `mod`: module name + """ + global topo, intf, intf1, sw_name, pkt + 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 = "{}/ospf_gr_helper.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) + + # Don't run this test if we have any failure. + if tgen.routers_have_failure(): + pytest.skip(tgen.errors) + + ospf_covergence = verify_ospf_neighbor(tgen, topo, lan=True) + assert ospf_covergence is True, "setup_module :Failed \n Error: {}".format( + ospf_covergence + ) + + sw_name = "s1" + intf = topo["routers"]["r0"]["links"][sw_name]["interface"] + intf1 = topo["routers"]["r1"]["links"][sw_name]["interface"] + pkt = topo["routers"]["r1"]["opq_lsa_hex"] + + logger.info("Running setup_module() done") + + +def teardown_module(): + """Teardown the pytest environment""" + + logger.info("Running teardown_module to delete topology") + + tgen = get_topogen() + + try: + # Stop toplogy and Remove tmp files + tgen.stop_topology() + + except OSError: + # OSError exception is raised when mininet tries to stop switch + # though switch is stopped once but mininet tries to stop same + # switch again, where it ended up with exception + pass + + +def delete_ospf(): + """delete ospf process after each test""" + tgen = get_topogen() + step("Delete ospf process") + for rtr in topo["routers"]: + ospf_del = {rtr: {"ospf": {"delete": True}}} + result = create_router_ospf(tgen, topo, ospf_del) + assert result is True, "Testcase: Failed \n Error: {}".format(result) + + +# ################################## +# Test cases start here. +# ################################## + + +def test_ospf_gr_helper_tc3_p1(request): + """ + OSPF GR on Broadcast : Verify DUT enters Helper mode when neighbor + sends grace lsa, helps RR to restart gracefully (RR = BDR) + """ + tc_name = request.node.name + write_test_header(tc_name) + tgen = get_topogen() + + # Don't run this test if we have any failure. + if tgen.routers_have_failure(): + pytest.skip(tgen.errors) + + global topo, intf, intf1, pkt + + step("Bring up the base config as per the topology") + step( + "Configure DR priority as 99 in RR , DUT dr priority = 98 " + "& reset ospf process in all the routers" + ) + reset_config_on_routers(tgen) + ospf_covergence = verify_ospf_neighbor(tgen, topo, lan=True) + assert ( + ospf_covergence is True + ), "OSPF is not after reset config \n Error: {}".format(ospf_covergence) + + step("Configure DR priority 100 on R0 and clear ospf neighbors " + "on all the routers.") + + input_dict = { + "r0": { + "links": { + sw_name: { + "interface": topo["routers"]["r0"]["links"][sw_name]["interface"], + "ospf": {"priority": 100}, + } + } + } + } + + result = create_interfaces_cfg(tgen, input_dict) + assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) + + step("Clear ospf neighbours in all routers") + for rtr in topo["routers"]: + clear_ospf(tgen, rtr) + + step("Verify that DR election is triggered and R0 is elected as DR") + input_dict = { + "r0": { + "ospf": { + "neighbors": { + "r1": {"nbrState": "Full", "role": "Backup"}, + "r2": {"nbrState": "Full", "role": "DROther"}, + "r3": {"nbrState": "Full", "role": "DROther"}, + } + } + } + } + dut = "r0" + result = verify_ospf_neighbor(tgen, topo, dut, input_dict, lan=True) + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + ospf_gr_r0 = { + "r0": {"ospf": {"graceful-restart": {"helper enable": [], "opaque": True}}} + } + result = create_router_ospf(tgen, topo, ospf_gr_r0) + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + ospf_gr_r1 = { + "r1": {"ospf": {"graceful-restart": {"helper enable": [], "opaque": True}}} + } + result = create_router_ospf(tgen, topo, ospf_gr_r1) + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + step("Verify that DUT enters into helper mode.") + + input_dict = {"activeRestarterCnt": 1} + gracelsa_sent = False + repeat = 0 + dut = "r0" + while not gracelsa_sent and repeat < Iters: + gracelsa_sent = scapy_send_raw_packet(tgen, topo, "r1", intf1, pkt) + result = verify_ospf_gr_helper(tgen, topo, dut, input_dict) + if isinstance(result, str): + repeat += 1 + gracelsa_sent = False + + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + delete_ospf() + write_test_footer(tc_name) + + +def test_ospf_gr_helper_tc4_p1(request): + """ + OSPF GR on Broadcast : Verify DUT enters Helper mode when neighbor + sends grace lsa, helps RR to restart gracefully (RR = DRother) + """ + tc_name = request.node.name + write_test_header(tc_name) + tgen = get_topogen() + + # Don't run this test if we have any failure. + if tgen.routers_have_failure(): + pytest.skip(tgen.errors) + + global topo, intf, intf1, pkt + + step("Bring up the base config as per the topology") + step( + "Configure DR priority as 99 in RR , DUT dr priority = 98 " + "& reset ospf process in all the routers" + ) + reset_config_on_routers(tgen) + ospf_covergence = verify_ospf_neighbor(tgen, topo, lan=True) + assert ( + ospf_covergence is True + ), "OSPF is not after reset config \n Error: {}".format(ospf_covergence) + step("Configure DR priority 0 on R0 and clear ospf neighbors on all the routers.") + + input_dict = { + "r0": { + "links": { + sw_name: { + "interface": topo["routers"]["r0"]["links"][sw_name]["interface"], + "ospf": {"priority": 0}, + } + } + } + } + + result = create_interfaces_cfg(tgen, input_dict) + assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) + + step("Clear ospf neighbours in all routers") + for rtr in topo["routers"]: + clear_ospf(tgen, rtr) + + step("Verify that DR election is triggered and R0 is elected as 2-Way") + input_dict = { + "r0": { + "ospf": { + "neighbors": { + "r1": {"nbrState": "Full", "role": "DR"}, + "r2": {"nbrState": "2-Way", "role": "DROther"}, + "r3": {"nbrState": "2-Way", "role": "DROther"}, + } + } + } + } + dut = "r0" + result = verify_ospf_neighbor(tgen, topo, dut, input_dict, lan=True) + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + ospf_gr_r0 = { + "r0": {"ospf": {"graceful-restart": {"helper enable": [], "opaque": True}}} + } + result = create_router_ospf(tgen, topo, ospf_gr_r0) + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + ospf_gr_r1 = { + "r1": {"ospf": {"graceful-restart": {"helper enable": [], "opaque": True}}} + } + result = create_router_ospf(tgen, topo, ospf_gr_r1) + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + step("Verify that DUT enters into helper mode.") + + input_dict = {"activeRestarterCnt": 1} + gracelsa_sent = False + repeat = 0 + dut = "r0" + while not gracelsa_sent and repeat < Iters: + gracelsa_sent = scapy_send_raw_packet(tgen, topo, "r1", intf1, pkt) + result = verify_ospf_gr_helper(tgen, topo, dut, input_dict) + if isinstance(result, str): + repeat += 1 + gracelsa_sent = False + + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + delete_ospf() + + write_test_footer(tc_name) + + +if __name__ == "__main__": + args = ["-s"] + sys.argv[1:] + sys.exit(pytest.main(args)) diff --git a/tests/topotests/ospf_gr_helper/test_ospf_gr_helper3.py b/tests/topotests/ospf_gr_helper/test_ospf_gr_helper3.py new file mode 100644 index 0000000..3be2819 --- /dev/null +++ b/tests/topotests/ospf_gr_helper/test_ospf_gr_helper3.py @@ -0,0 +1,310 @@ +#!/usr/bin/python +# SPDX-License-Identifier: ISC + +# +# Copyright (c) 2021 by VMware, Inc. ("VMware") +# Used Copyright (c) 2018 by Network Device Education Foundation, Inc. +# ("NetDEF") in this file. +# + + +"""OSPF Basic Functionality Automation.""" +import os +import sys +import time +import pytest + +# Save the Current Working Directory to find configuration files. +CWD = os.path.dirname(os.path.realpath(__file__)) +sys.path.append(os.path.join(CWD, "../")) + +# pylint: disable=C0413 +# Import topogen and topotest helpers +from lib.topogen import Topogen, get_topogen + +# Import topoJson from lib, to create topology and initial configuration +from lib.common_config import ( + start_topology, + write_test_header, + write_test_footer, + reset_config_on_routers, + step, + create_interfaces_cfg, + scapy_send_raw_packet, +) + +from lib.topolog import logger +from lib.topojson import build_config_from_json + +from lib.ospf import ( + verify_ospf_neighbor, + clear_ospf, + verify_ospf_gr_helper, + create_router_ospf, +) + +pytestmark = [pytest.mark.ospfd] + +# Global variables +topo = None +Iters = 5 +sw_name = None +intf = None +intf1 = None +pkt = None + +""" +Topology: + + Please view in a fixed-width font such as Courier. + Topo : Broadcast Networks + DUT - HR RR + +---+ +---+ +---+ +---+ + |R0 + +R1 + +R2 + +R3 | + +-+-+ +-+-+ +-+-+ +-+-+ + | | | | + | | | | + --+-----------+--------------+---------------+----- + Ethernet Segment + +Testcases: + +TC1. Verify by default helper support is disabled for FRR ospf +TC2. OSPF GR on Broadcast : Verify DUT enters Helper mode when neighbor + sends grace lsa, helps RR to restart gracefully (RR = DR) +TC3. OSPF GR on Broadcast : Verify DUT enters Helper mode when neighbor + sends grace lsa, helps RR to restart gracefully (RR = BDR) +TC4. OSPF GR on Broadcast : Verify DUT enters Helper mode when neighbor + sends grace lsa, helps RR to restart gracefully (RR = DRother) +TC5. OSPF GR on P2P : Verify DUT enters Helper mode when neighbor sends + grace lsa, helps RR to restart gracefully. +TC6. Verify all the show commands newly introducted as part of ospf + helper support - Json Key verification wrt to show commands. +TC7. Verify helper when grace lsa is received with different configured + value in process level (higher, lower, grace lsa timer above 1800) +TC8. Verify helper functionality when dut is helping RR and new grace lsa + is received from RR. +""" + + +def setup_module(mod): + """ + Sets up the pytest environment + + * `mod`: module name + """ + global topo, intf, intf1, sw_name, pkt + 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 = "{}/ospf_gr_helper.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) + + # Don't run this test if we have any failure. + if tgen.routers_have_failure(): + pytest.skip(tgen.errors) + + ospf_covergence = verify_ospf_neighbor(tgen, topo, lan=True) + assert ospf_covergence is True, "setup_module :Failed \n Error: {}".format( + ospf_covergence + ) + + sw_name = "s1" + intf = topo["routers"]["r0"]["links"][sw_name]["interface"] + intf1 = topo["routers"]["r1"]["links"][sw_name]["interface"] + pkt = topo["routers"]["r1"]["opq_lsa_hex"] + + logger.info("Running setup_module() done") + + +def teardown_module(): + """Teardown the pytest environment""" + + logger.info("Running teardown_module to delete topology") + + tgen = get_topogen() + + try: + # Stop toplogy and Remove tmp files + tgen.stop_topology() + + except OSError: + # OSError exception is raised when mininet tries to stop switch + # though switch is stopped once but mininet tries to stop same + # switch again, where it ended up with exception + pass + + +def delete_ospf(): + """delete ospf process after each test""" + tgen = get_topogen() + step("Delete ospf process") + for rtr in topo["routers"]: + ospf_del = {rtr: {"ospf": {"delete": True}}} + result = create_router_ospf(tgen, topo, ospf_del) + assert result is True, "Testcase: Failed \n Error: {}".format(result) + + +# ################################## +# Test cases start here. +# ################################## + + +def test_ospf_gr_helper_tc7_p1(request): + """ + Test ospf gr helper + Verify helper when grace lsa is received with different configured + value in process level (higher, lower, grace lsa timer above 1800) + """ + tc_name = request.node.name + write_test_header(tc_name) + tgen = get_topogen() + + # Don't run this test if we have any failure. + if tgen.routers_have_failure(): + pytest.skip(tgen.errors) + + global topo, intf, intf1, pkt + + step("Bring up the base config as per the topology") + step( + "Configure DR priority as 99 in RR , DUT dr priority = 98 " + "& reset ospf process in all the routers" + ) + step( + "Enable GR on RR and DUT with grace period on RR = 333" + "and grace period on DUT = 300" + ) + reset_config_on_routers(tgen) + ospf_covergence = verify_ospf_neighbor(tgen, topo, lan=True) + assert ( + ospf_covergence is True + ), "OSPF is not after reset config \n Error: {}".format(ospf_covergence) + ospf_gr_r0 = { + "r0": {"ospf": {"graceful-restart": {"helper enable": [], "opaque": True}}} + } + result = create_router_ospf(tgen, topo, ospf_gr_r0) + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + ospf_gr_r1 = { + "r1": {"ospf": {"graceful-restart": {"helper enable": [], "opaque": True}}} + } + result = create_router_ospf(tgen, topo, ospf_gr_r1) + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + input_dict = {"supportedGracePeriod": 1800} + dut = "r0" + result = verify_ospf_gr_helper(tgen, topo, dut, input_dict) + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + step("Configure grace period = 1801 on RR and restart ospf .") + grace_period_1801 = "01005e00000570708bd051ef080045c0005cbeb10000015907d111010101e00000050204004801010101000000009714000000000000000000000000000100010209030000000101010180000001c8e9002c000100040000016800020001010000000003000411010101" + gracelsa_sent = scapy_send_raw_packet(tgen, topo, "r1", intf1, grace_period_1801) + + step("Verify R0 does not enter helper mode.") + input_dict = {"activeRestarterCnt": 1} + dut = "r0" + result = verify_ospf_gr_helper(tgen, topo, dut, input_dict, expected=False) + assert ( + result is not True + ), "Testcase {} : Failed. DUT entered helper role \n Error: {}".format( + tc_name, result + ) + + delete_ospf() + + write_test_footer(tc_name) + + +def test_ospf_gr_helper_tc8_p1(request): + """ + Test ospf gr helper + + Verify helper functionality when dut is helping RR and new grace lsa + is received from RR. + """ + tc_name = request.node.name + write_test_header(tc_name) + tgen = get_topogen() + + # Don't run this test if we have any failure. + if tgen.routers_have_failure(): + pytest.skip(tgen.errors) + + global topo, intf, intf1, pkt + + step("Bring up the base config as per the topology") + step("Enable GR") + reset_config_on_routers(tgen) + ospf_covergence = verify_ospf_neighbor(tgen, topo, lan=True) + assert ( + ospf_covergence is True + ), "OSPF is not after reset config \n Error: {}".format(ospf_covergence) + ospf_gr_r0 = { + "r0": {"ospf": {"graceful-restart": {"helper enable": [], "opaque": True}}} + } + result = create_router_ospf(tgen, topo, ospf_gr_r0) + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + ospf_gr_r1 = { + "r1": {"ospf": {"graceful-restart": {"helper enable": [], "opaque": True}}} + } + result = create_router_ospf(tgen, topo, ospf_gr_r1) + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + input_dict = {"supportedGracePeriod": 1800} + dut = "r0" + result = verify_ospf_gr_helper(tgen, topo, dut, input_dict) + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + step("Verify that DUT enters into helper mode.") + + input_dict = {"activeRestarterCnt": 1} + gracelsa_sent = False + repeat = 0 + dut = "r0" + while not gracelsa_sent and repeat < Iters: + gracelsa_sent = scapy_send_raw_packet(tgen, topo, "r1", intf1, pkt) + result = verify_ospf_gr_helper(tgen, topo, dut, input_dict) + if isinstance(result, str): + repeat += 1 + gracelsa_sent = False + + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + step("Send the Grace LSA again to DUT when RR is in GR.") + input_dict = {"activeRestarterCnt": 1} + gracelsa_sent = False + repeat = 0 + dut = "r0" + while not gracelsa_sent and repeat < Iters: + gracelsa_sent = scapy_send_raw_packet(tgen, topo, "r1", intf1, pkt) + result = verify_ospf_gr_helper(tgen, topo, dut, input_dict) + if isinstance(result, str): + repeat += 1 + gracelsa_sent = False + + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + delete_ospf() + + write_test_footer(tc_name) + + +if __name__ == "__main__": + args = ["-s"] + sys.argv[1:] + sys.exit(pytest.main(args)) |