[mirror_frr.git] / tests / topotests / ospf_gr_helper / test_ospf_gr_helper3.py

#!/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))
Commit	Line	Data
019a4d6c	1	#!/usr/bin/python
acddc0ed	2	# SPDX-License-Identifier: ISC
019a4d6c DS	3
	4	#
	5	# Copyright (c) 2021 by VMware, Inc. ("VMware")
	6	# Used Copyright (c) 2018 by Network Device Education Foundation, Inc.
	7	# ("NetDEF") in this file.
	8	#
019a4d6c DS	9
	10
	11	"""OSPF Basic Functionality Automation."""
	12	import os
	13	import sys
	14	import time
	15	import pytest
	16
	17	# Save the Current Working Directory to find configuration files.
	18	CWD = os.path.dirname(os.path.realpath(__file__))
	19	sys.path.append(os.path.join(CWD, "../"))
	20
	21	# pylint: disable=C0413
	22	# Import topogen and topotest helpers
	23	from lib.topogen import Topogen, get_topogen
	24
	25	# Import topoJson from lib, to create topology and initial configuration
	26	from lib.common_config import (
	27	start_topology,
	28	write_test_header,
	29	write_test_footer,
	30	reset_config_on_routers,
	31	step,
	32	create_interfaces_cfg,
019a4d6c DS	33	scapy_send_raw_packet,
	34	)
	35
	36	from lib.topolog import logger
	37	from lib.topojson import build_config_from_json
	38
	39	from lib.ospf import (
	40	verify_ospf_neighbor,
	41	clear_ospf,
	42	verify_ospf_gr_helper,
	43	create_router_ospf,
	44	)
	45
cbdecd68 DS	46	pytestmark = [pytest.mark.ospfd]
cbdecd68 DS	47
019a4d6c DS	48	# Global variables
	49	topo = None
	50	Iters = 5
	51	sw_name = None
	52	intf = None
	53	intf1 = None
	54	pkt = None
	55
	56	"""
	57	Topology:
	58
	59	Please view in a fixed-width font such as Courier.
	60	Topo : Broadcast Networks
	61	DUT - HR RR
	62	+---+ +---+ +---+ +---+
	63	\|R0 + +R1 + +R2 + +R3 \|
	64	+-+-+ +-+-+ +-+-+ +-+-+
	65	\| \| \| \|
	66	\| \| \| \|
	67	--+-----------+--------------+---------------+-----
	68	Ethernet Segment
	69
	70	Testcases:
	71
	72	TC1. Verify by default helper support is disabled for FRR ospf
	73	TC2. OSPF GR on Broadcast : Verify DUT enters Helper mode when neighbor
	74	sends grace lsa, helps RR to restart gracefully (RR = DR)
	75	TC3. OSPF GR on Broadcast : Verify DUT enters Helper mode when neighbor
	76	sends grace lsa, helps RR to restart gracefully (RR = BDR)
	77	TC4. OSPF GR on Broadcast : Verify DUT enters Helper mode when neighbor
	78	sends grace lsa, helps RR to restart gracefully (RR = DRother)
	79	TC5. OSPF GR on P2P : Verify DUT enters Helper mode when neighbor sends
	80	grace lsa, helps RR to restart gracefully.
	81	TC6. Verify all the show commands newly introducted as part of ospf
	82	helper support - Json Key verification wrt to show commands.
	83	TC7. Verify helper when grace lsa is received with different configured
	84	value in process level (higher, lower, grace lsa timer above 1800)
	85	TC8. Verify helper functionality when dut is helping RR and new grace lsa
	86	is received from RR.
	87	"""
	88
	89
	90	def setup_module(mod):
	91	"""
	92	Sets up the pytest environment
	93
	94	* `mod`: module name
	95	"""
	96	global topo, intf, intf1, sw_name, pkt
	97	testsuite_run_time = time.asctime(time.localtime(time.time()))
	98	logger.info("Testsuite start time: {}".format(testsuite_run_time))
	99	logger.info("=" * 40)
	100
	101	logger.info("Running setup_module to create topology")
	102
	103	# This function initiates the topology build with Topogen...
	104	json_file = "{}/ospf_gr_helper.json".format(CWD)
	105	tgen = Topogen(json_file, mod.__name__)
	106	global topo
	107	topo = tgen.json_topo
	108	# ... and here it calls Mininet initialization functions.
	109
019a4d6c DS	110	# Starting topology, create tmp files which are loaded to routers
019a4d6c DS	111	# to start daemons and then start routers
991a971f	112	start_topology(tgen)
019a4d6c DS	113
	114	# Creating configuration from JSON
	115	build_config_from_json(tgen, topo)
	116
	117	# Don't run this test if we have any failure.
	118	if tgen.routers_have_failure():
	119	pytest.skip(tgen.errors)
	120
	121	ospf_covergence = verify_ospf_neighbor(tgen, topo, lan=True)
	122	assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format(
	123	ospf_covergence
	124	)
	125
	126	sw_name = "s1"
	127	intf = topo["routers"]["r0"]["links"][sw_name]["interface"]
	128	intf1 = topo["routers"]["r1"]["links"][sw_name]["interface"]
	129	pkt = topo["routers"]["r1"]["opq_lsa_hex"]
	130
	131	logger.info("Running setup_module() done")
	132
	133
	134	def teardown_module():
	135	"""Teardown the pytest environment"""
	136
	137	logger.info("Running teardown_module to delete topology")
	138
	139	tgen = get_topogen()
	140
	141	try:
	142	# Stop toplogy and Remove tmp files
82db2cd1	143	tgen.stop_topology()
019a4d6c DS	144
	145	except OSError:
	146	# OSError exception is raised when mininet tries to stop switch
	147	# though switch is stopped once but mininet tries to stop same
	148	# switch again, where it ended up with exception
	149	pass
	150
	151
	152	def delete_ospf():
	153	"""delete ospf process after each test"""
	154	tgen = get_topogen()
	155	step("Delete ospf process")
	156	for rtr in topo["routers"]:
	157	ospf_del = {rtr: {"ospf": {"delete": True}}}
	158	result = create_router_ospf(tgen, topo, ospf_del)
	159	assert result is True, "Testcase: Failed \n Error: {}".format(result)
	160
	161
	162	# ##################################
	163	# Test cases start here.
	164	# ##################################
	165
	166
	167	def test_ospf_gr_helper_tc7_p1(request):
	168	"""
	169	Test ospf gr helper
	170	Verify helper when grace lsa is received with different configured
	171	value in process level (higher, lower, grace lsa timer above 1800)
	172	"""
	173	tc_name = request.node.name
	174	write_test_header(tc_name)
	175	tgen = get_topogen()
	176
	177	# Don't run this test if we have any failure.
	178	if tgen.routers_have_failure():
	179	pytest.skip(tgen.errors)
	180
	181	global topo, intf, intf1, pkt
	182
	183	step("Bring up the base config as per the topology")
	184	step(
	185	"Configure DR priority as 99 in RR , DUT dr priority = 98 "
	186	"& reset ospf process in all the routers"
	187	)
	188	step(
	189	"Enable GR on RR and DUT with grace period on RR = 333"
	190	"and grace period on DUT = 300"
	191	)
	192	reset_config_on_routers(tgen)
	193	ospf_covergence = verify_ospf_neighbor(tgen, topo, lan=True)
	194	assert (
	195	ospf_covergence is True
	196	), "OSPF is not after reset config \n Error:" " {}".format(ospf_covergence)
	197	ospf_gr_r0 = {
	198	"r0": {"ospf": {"graceful-restart": {"helper enable": [], "opaque": True}}}
	199	}
	200	result = create_router_ospf(tgen, topo, ospf_gr_r0)
	201	assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
	202
	203	ospf_gr_r1 = {
	204	"r1": {"ospf": {"graceful-restart": {"helper enable": [], "opaque": True}}}
	205	}
	206	result = create_router_ospf(tgen, topo, ospf_gr_r1)
	207	assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
208
209	input_dict = {"supportedGracePeriod": 1800}
210	dut = "r0"
211	result = verify_ospf_gr_helper(tgen, topo, dut, input_dict)
212	assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
213
214	step("Configure grace period = 1801 on RR and restart ospf .")
215	grace_period_1801 = "01005e00000570708bd051ef080045c0005cbeb10000015907d111010101e00000050204004801010101000000009714000000000000000000000000000100010209030000000101010180000001c8e9002c000100040000016800020001010000000003000411010101"
216	gracelsa_sent = scapy_send_raw_packet(tgen, topo, "r1", intf1, grace_period_1801)
217
218	step("Verify R0 does not enter helper mode.")
219	input_dict = {"activeRestarterCnt": 1}
220	dut = "r0"
221	result = verify_ospf_gr_helper(tgen, topo, dut, input_dict, expected=False)
222	assert (
223	result is not True
224	), "Testcase {} : Failed. DUT entered helper role " " \n Error: {}".format(
225	tc_name, result
226	)
227
228	delete_ospf()
229
230	write_test_footer(tc_name)
231
232
233	def test_ospf_gr_helper_tc8_p1(request):
234	"""
235	Test ospf gr helper
236
237	Verify helper functionality when dut is helping RR and new grace lsa
238	is received from RR.
239	"""
240	tc_name = request.node.name
241	write_test_header(tc_name)
242	tgen = get_topogen()
243
244	# Don't run this test if we have any failure.
245	if tgen.routers_have_failure():
246	pytest.skip(tgen.errors)
247
248	global topo, intf, intf1, pkt
249
250	step("Bring up the base config as per the topology")
251	step("Enable GR")
252	reset_config_on_routers(tgen)
253	ospf_covergence = verify_ospf_neighbor(tgen, topo, lan=True)
254	assert (
255	ospf_covergence is True
256	), "OSPF is not after reset config \n Error:" " {}".format(ospf_covergence)
257	ospf_gr_r0 = {
258	"r0": {"ospf": {"graceful-restart": {"helper enable": [], "opaque": True}}}
259	}
260	result = create_router_ospf(tgen, topo, ospf_gr_r0)
261	assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
262
263	ospf_gr_r1 = {
264	"r1": {"ospf": {"graceful-restart": {"helper enable": [], "opaque": True}}}
265	}
266	result = create_router_ospf(tgen, topo, ospf_gr_r1)
267	assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
268
269	input_dict = {"supportedGracePeriod": 1800}
270	dut = "r0"
271	result = verify_ospf_gr_helper(tgen, topo, dut, input_dict)
272	assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
273
274	step("Verify that DUT enters into helper mode.")
275
276	input_dict = {"activeRestarterCnt": 1}
277	gracelsa_sent = False
278	repeat = 0
279	dut = "r0"
280	while not gracelsa_sent and repeat < Iters:
281	gracelsa_sent = scapy_send_raw_packet(tgen, topo, "r1", intf1, pkt)
282	result = verify_ospf_gr_helper(tgen, topo, dut, input_dict)
283	if isinstance(result, str):
284	repeat += 1
285	gracelsa_sent = False
286
287	assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
288
289	step("Send the Grace LSA again to DUT when RR is in GR.")
290	input_dict = {"activeRestarterCnt": 1}
291	gracelsa_sent = False
292	repeat = 0
293	dut = "r0"
294	while not gracelsa_sent and repeat < Iters:
295	gracelsa_sent = scapy_send_raw_packet(tgen, topo, "r1", intf1, pkt)
296	result = verify_ospf_gr_helper(tgen, topo, dut, input_dict)
297	if isinstance(result, str):
298	repeat += 1
299	gracelsa_sent = False
300
301	assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
302
303	delete_ospf()
304
305	write_test_footer(tc_name)
306
307
308	if __name__ == "__main__":
309	args = ["-s"] + sys.argv[1:]
310	sys.exit(pytest.main(args))