From: Donald Sharp Date: Thu, 26 May 2022 17:32:27 +0000 (-0400) Subject: tests: Convert test_ospf_gr_helper into running more parallel X-Git-Tag: base_8.3~18^2 X-Git-Url: https://git.puffer.fish/?a=commitdiff_plain;h=019a4d6cd5788b35b8fba39e43d0e0abd924e43e;p=mirror%2Ffrr.git tests: Convert test_ospf_gr_helper into running more parallel With this change run time in parallel for these tests go from 10:37 -> 4:11 on my machine. Signed-off-by: Donald Sharp --- diff --git a/tests/topotests/ospf_gr_helper/test_ospf_gr_helper.py b/tests/topotests/ospf_gr_helper/test_ospf_gr_helper.py deleted file mode 100644 index ff182be66f..0000000000 --- a/tests/topotests/ospf_gr_helper/test_ospf_gr_helper.py +++ /dev/null @@ -1,721 +0,0 @@ -#!/usr/bin/python - -# -# Copyright (c) 2021 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. -# - - -"""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, - topo_daemons, - 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, -) - -# 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. - - # get list of daemons needs to be started for this suite. - daemons = topo_daemons(tgen, topo) - - # Starting topology, create tmp files which are loaded to routers - # to start daemons and then start routers - start_topology(tgen, daemons) - - # 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", - "restartSupoort": "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", - "restartSupoort": "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) - - -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 pririty 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": {"state": "Full", "role": "Backup"}, - "r2": {"state": "Full", "role": "DROther"}, - "r3": {"state": "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 pririty 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": 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": {"state": "Full", "role": "DR"}, - "r2": {"state": "2-Way", "role": "DROther"}, - "r3": {"state": "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) - - -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)) 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 0000000000..8c855620be --- /dev/null +++ b/tests/topotests/ospf_gr_helper/test_ospf_gr_helper1.py @@ -0,0 +1,395 @@ +#!/usr/bin/python + +# +# Copyright (c) 2021 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. +# + + +"""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, + topo_daemons, + 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, +) + +# 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. + + # get list of daemons needs to be started for this suite. + daemons = topo_daemons(tgen, topo) + + # Starting topology, create tmp files which are loaded to routers + # to start daemons and then start routers + start_topology(tgen, daemons) + + # 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", + "restartSupoort": "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", + "restartSupoort": "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 0000000000..e7d0621df8 --- /dev/null +++ b/tests/topotests/ospf_gr_helper/test_ospf_gr_helper2.py @@ -0,0 +1,373 @@ +#!/usr/bin/python + +# +# Copyright (c) 2021 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. +# + + +"""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, + topo_daemons, + 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, +) + +# 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. + + # get list of daemons needs to be started for this suite. + daemons = topo_daemons(tgen, topo) + + # Starting topology, create tmp files which are loaded to routers + # to start daemons and then start routers + start_topology(tgen, daemons) + + # 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 pririty 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": {"state": "Full", "role": "Backup"}, + "r2": {"state": "Full", "role": "DROther"}, + "r3": {"state": "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 pririty 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": 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": {"state": "Full", "role": "DR"}, + "r2": {"state": "2-Way", "role": "DROther"}, + "r3": {"state": "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 0000000000..4cb3747c56 --- /dev/null +++ b/tests/topotests/ospf_gr_helper/test_ospf_gr_helper3.py @@ -0,0 +1,325 @@ +#!/usr/bin/python + +# +# Copyright (c) 2021 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. +# + + +"""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, + topo_daemons, + 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, +) + +# 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. + + # get list of daemons needs to be started for this suite. + daemons = topo_daemons(tgen, topo) + + # Starting topology, create tmp files which are loaded to routers + # to start daemons and then start routers + start_topology(tgen, daemons) + + # 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))