[mirror_frr.git] / tests / topotests / ospfv3_basic_functionality / test_ospfv3_ecmp_lan.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
import json
from copy import deepcopy
from ipaddress import IPv4Address
from lib.topotest import frr_unicode
import ipaddress

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

# 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,
    verify_rib,
    create_static_routes,
    step,
    create_route_maps,
    shutdown_bringup_interface,
    create_interfaces_cfg,
    get_frr_ipv6_linklocal,
)
from lib.topolog import logger
from lib.topojson import build_config_from_json

from lib.ospf import (
    verify_ospf6_neighbor,
    config_ospf_interface,
    clear_ospf,
    verify_ospf6_rib,
    create_router_ospf,
    verify_ospf6_interface,
    verify_ospf6_database,
    config_ospf6_interface,
)

from ipaddress import IPv6Address

pytestmark = [pytest.mark.ospfd, pytest.mark.staticd]


# Global variables
topo = None

NETWORK = {
    "ipv4": [
        "11.0.20.1/32",
        "11.0.20.2/32",
        "11.0.20.3/32",
        "11.0.20.4/32",
        "11.0.20.5/32",
    ],
    "ipv6": ["2::1/128", "2::2/128", "2::3/128", "2::4/128", "2::5/128"],
}
MASK = {"ipv6": "32", "ipv6": "128"}

"""
TOPOOLOGY =
      Please view in a fixed-width font such as Courier.
      Topo : Broadcast Networks
        +---+       +---+          +---+           +---+
        |R0 +       +R1 +          +R2 +           +R3 |
        +-+-+       +-+-+          +-+-+           +-+-+
          |           |              |               |
          |           |              |               |
        --+-----------+--------------+---------------+-----
                         Ethernet Segment

TESTCASES =
1.  Verify OSPF ECMP with max path configured as 8
    (Edge having 1 uplink port as broadcast network,
    connect to 8 TORs - LAN case)

 """


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

    * `mod`: module name
    """
    global topo, switch_name
    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 = "{}/ospfv3_ecmp_lan.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)
    # Api call verify whether OSPF is converged
    ospf_covergence = verify_ospf6_neighbor(tgen, topo, lan=True)
    assert ospf_covergence is True, "setup_module :Failed \n Error:  {}".format(
        ospf_covergence
    )

    switch_name = [k for k in topo["switches"].keys()][0]

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


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

    * `mod`: module name
    """

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

    tgen = get_topogen()

    # Stop toplogy and Remove tmp files

    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
    logger.info(
        "Testsuite end time: {}".format(time.asctime(time.localtime(time.time())))
    )
    logger.info("=" * 40)


def red_static(dut, config=True):
    """Local def for Redstribute static routes inside ospf."""
    global topo
    tgen = get_topogen()
    if config:
        ospf_red = {dut: {"ospf6": {"redistribute": [{"redist_type": "static"}]}}}
    else:
        ospf_red = {
            dut: {
                "ospf6": {"redistribute": [{"redist_type": "static", "delete": True}]}
            }
        }
    result = create_router_ospf(tgen, topo, ospf_red)
    assert result is True, "Testcase : Failed \n Error: {}".format(result)


def red_connected(dut, config=True):
    """Local def for Redstribute connected routes inside ospf."""
    global topo
    tgen = get_topogen()
    if config:
        ospf_red = {dut: {"ospf6": {"redistribute": [{"redist_type": "connected"}]}}}
    else:
        ospf_red = {
            dut: {
                "ospf6": {
                    "redistribute": [{"redist_type": "connected", "del_action": True}]
                }
            }
        }
    result = create_router_ospf(tgen, topo, ospf_red)
    assert result is True, "Testcase: Failed \n Error: {}".format(result)


def get_llip(onrouter, intf):
    """
    API to get the link local ipv6 address of a particular interface

    Parameters
    ----------
    * `fromnode`: Source node
    * `tonode` : interface for which link local ip needs to be returned.

    Usage
    -----
    result = get_llip('r1', 'r2-link0')

    Returns
    -------
    1) link local ipv6 address from the interface.
    2) errormsg - when link local ip not found.
    """
    tgen = get_topogen()
    intf = topo["routers"][onrouter]["links"][intf]["interface"]
    llip = get_frr_ipv6_linklocal(tgen, onrouter, intf)
    if llip:
        logger.info("llip ipv6 address to be set as NH is %s", llip)
        return llip
    return None


def get_glipv6(onrouter, intf):
    """
    API to get the global ipv6 address of a particular interface

    Parameters
    ----------
    * `onrouter`: Source node
    * `intf` : interface for which link local ip needs to be returned.

    Usage
    -----
    result = get_glipv6('r1', 'r2-link0')

    Returns
    -------
    1) global ipv6 address from the interface.
    2) errormsg - when link local ip not found.
    """
    glipv6 = (topo["routers"][onrouter]["links"][intf]["ipv6"]).split("/")[0]
    if glipv6:
        logger.info("Global ipv6 address to be set as NH is %s", glipv6)
        return glipv6
    return None


# ##################################
# Test cases start here.
# ##################################


def test_ospfv3_lan_ecmp_tc18_p0(request):
    """
    OSPF ECMP.

    Verify OSPF ECMP with max path configured as 8
    (Edge having 1 uplink port as broadcast network,
    connect to 8 TORs - LAN case)

    """
    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
    step("Bring up the base config as per the topology")
    step(". Configure ospf in all the routers on LAN interface.")

    reset_config_on_routers(tgen)

    step("Verify that OSPF is up with 8 neighborship sessions.")

    ospf_covergence = verify_ospf6_neighbor(tgen, topo, lan=True)
    assert ospf_covergence is True, "Testcase Failed \n Error:  {}".format(
        ospf_covergence
    )

    step(
        "Configure a static route in all the routes and "
        "redistribute static/connected in OSPF."
    )

    for rtr in topo["routers"]:
        input_dict = {
            rtr: {
                "static_routes": [
                    {"network": NETWORK["ipv6"][0], "no_of_ip": 5, "next_hop": "Null0"}
                ]
            }
        }
        result = create_static_routes(tgen, input_dict)
        assert result is True, "Testcase {} : Failed \n Error: {}".format(
            tc_name, result
        )

        dut = rtr
        red_static(dut)

    step(
        "Verify that route in R0 in stalled with 8 hops. "
        "Verify ospf route table and ip route table."
    )

    nh = []
    for rtr in topo["routers"]:
        llip = get_llip(rtr, switch_name)
        assert llip is not None, "Testcase {} : Failed \n Error: {}".format(
            tc_name, result
        )
        nh.append(llip)

    llip = get_llip("r1", switch_name)
    assert llip is not None, "Testcase {} : Failed \n Error: {}".format(tc_name, result)

    nh.remove(llip)
    dut = "r1"
    result = verify_ospf6_rib(tgen, dut, input_dict)
    assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)

    protocol = "ospf6"
    result = verify_rib(tgen, "ipv6", dut, input_dict, protocol=protocol)
    assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)

    step(" clear ip ospf interface on DUT(r0)")
    clear_ospf(tgen, "r0", ospf="ospf6")

    step(
        "Verify that after clearing the ospf interface all the "
        "neighbours are up and routes are installed with 8 next hop "
        "in ospf and ip route tables on R0"
    )

    dut = "r0"
    ospf_covergence = verify_ospf6_neighbor(tgen, topo, dut=dut, lan=True)
    assert ospf_covergence is True, "Testcase Failed \n Error:  {}".format(
        ospf_covergence
    )

    step(" clear ip ospf interface on R2")
    clear_ospf(tgen, "r2")

    dut = "r2"
    ospf_covergence = verify_ospf6_neighbor(tgen, topo, dut=dut, lan=True)
    assert ospf_covergence is True, "Testcase Failed \n Error:  {}".format(
        ospf_covergence
    )

    step("Delete static/connected cmd in ospf in all the routes one by one.")
    for rtr in topo["routers"]:
        input_dict = {
            rtr: {
                "static_routes": [
                    {
                        "network": NETWORK["ipv6"][0],
                        "no_of_ip": 5,
                        "next_hop": "Null0",
                        "delete": True,
                    }
                ]
            }
        }
        result = create_static_routes(tgen, input_dict)
        assert result is True, "Testcase {} : Failed \n Error: {}".format(
            tc_name, result
        )

    write_test_footer(tc_name)


if __name__ == "__main__":
    args = ["-s"] + sys.argv[1:]
    sys.exit(pytest.main(args))
Commit	Line	Data
d1b5fa5b	1	#!/usr/bin/python
acddc0ed	2	# SPDX-License-Identifier: ISC
d1b5fa5b	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	#
d1b5fa5b	9
	10
	11	"""OSPF Basic Functionality Automation."""
	12	import os
	13	import sys
	14	import time
	15	import pytest
	16	import json
	17	from copy import deepcopy
	18	from ipaddress import IPv4Address
	19	from lib.topotest import frr_unicode
	20	import ipaddress
	21
	22	# Save the Current Working Directory to find configuration files.
	23	CWD = os.path.dirname(os.path.realpath(__file__))
	24	sys.path.append(os.path.join(CWD, "../"))
	25	sys.path.append(os.path.join(CWD, "../lib/"))
	26
	27	# pylint: disable=C0413
	28	# Import topogen and topotest helpers
	29	from lib.topogen import Topogen, get_topogen
	30
	31	# Import topoJson from lib, to create topology and initial configuration
	32	from lib.common_config import (
	33	start_topology,
	34	write_test_header,
	35	write_test_footer,
	36	reset_config_on_routers,
	37	verify_rib,
	38	create_static_routes,
	39	step,
	40	create_route_maps,
	41	shutdown_bringup_interface,
	42	create_interfaces_cfg,
d1b5fa5b	43	get_frr_ipv6_linklocal,
	44	)
	45	from lib.topolog import logger
	46	from lib.topojson import build_config_from_json
	47
	48	from lib.ospf import (
	49	verify_ospf6_neighbor,
	50	config_ospf_interface,
	51	clear_ospf,
	52	verify_ospf6_rib,
	53	create_router_ospf,
	54	verify_ospf6_interface,
	55	verify_ospf6_database,
	56	config_ospf6_interface,
	57	)
	58
	59	from ipaddress import IPv6Address
	60
	61	pytestmark = [pytest.mark.ospfd, pytest.mark.staticd]
	62
	63
	64	# Global variables
	65	topo = None
	66
	67	NETWORK = {
	68	"ipv4": [
	69	"11.0.20.1/32",
	70	"11.0.20.2/32",
	71	"11.0.20.3/32",
	72	"11.0.20.4/32",
	73	"11.0.20.5/32",
	74	],
	75	"ipv6": ["2::1/128", "2::2/128", "2::3/128", "2::4/128", "2::5/128"],
	76	}
	77	MASK = {"ipv6": "32", "ipv6": "128"}
	78
	79	"""
	80	TOPOOLOGY =
	81	Please view in a fixed-width font such as Courier.
	82	Topo : Broadcast Networks
	83	+---+ +---+ +---+ +---+
	84	\|R0 + +R1 + +R2 + +R3 \|
	85	+-+-+ +-+-+ +-+-+ +-+-+
	86	\| \| \| \|
	87	\| \| \| \|
	88	--+-----------+--------------+---------------+-----
	89	Ethernet Segment
	90
	91	TESTCASES =
	92	1. Verify OSPF ECMP with max path configured as 8
	93	(Edge having 1 uplink port as broadcast network,
	94	connect to 8 TORs - LAN case)
	95
	96	"""
	97
	98
	99	def setup_module(mod):
	100	"""
	101	Sets up the pytest environment
	102
	103	* `mod`: module name
	104	"""
	105	global topo, switch_name
	106	testsuite_run_time = time.asctime(time.localtime(time.time()))
107	logger.info("Testsuite start time: {}".format(testsuite_run_time))
108	logger.info("=" * 40)
109
110	logger.info("Running setup_module to create topology")
111
112	# This function initiates the topology build with Topogen...
113	json_file = "{}/ospfv3_ecmp_lan.json".format(CWD)
114	tgen = Topogen(json_file, mod.__name__)
115	global topo
116	topo = tgen.json_topo
117	# ... and here it calls Mininet initialization functions.
118
d1b5fa5b	119	# Starting topology, create tmp files which are loaded to routers
d60a3f0e	120	# to start daemons and then start routers
991a971f	121	start_topology(tgen)
d1b5fa5b	122
	123	# Creating configuration from JSON
	124	build_config_from_json(tgen, topo)
	125
	126	# Don't run this test if we have any failure.
	127	if tgen.routers_have_failure():
	128	pytest.skip(tgen.errors)
	129	# Api call verify whether OSPF is converged
	130	ospf_covergence = verify_ospf6_neighbor(tgen, topo, lan=True)
74dd0c84	131	assert ospf_covergence is True, "setup_module :Failed \n Error: {}".format(
d1b5fa5b	132	ospf_covergence
	133	)
	134
	135	switch_name = [k for k in topo["switches"].keys()][0]
	136
	137	logger.info("Running setup_module() done")
	138
	139
	140	def teardown_module(mod):
	141	"""
	142	Teardown the pytest environment.
	143
	144	* `mod`: module name
	145	"""
	146
	147	logger.info("Running teardown_module to delete topology")
	148
	149	tgen = get_topogen()
	150
	151	# Stop toplogy and Remove tmp files
	152
	153	try:
	154	# Stop toplogy and Remove tmp files
	155	tgen.stop_topology()
	156
	157	except OSError:
	158	# OSError exception is raised when mininet tries to stop switch
	159	# though switch is stopped once but mininet tries to stop same
	160	# switch again, where it ended up with exception
	161	pass
	162	logger.info(
	163	"Testsuite end time: {}".format(time.asctime(time.localtime(time.time())))
	164	)
	165	logger.info("=" * 40)
	166
	167
	168	def red_static(dut, config=True):
	169	"""Local def for Redstribute static routes inside ospf."""
	170	global topo
	171	tgen = get_topogen()
	172	if config:
	173	ospf_red = {dut: {"ospf6": {"redistribute": [{"redist_type": "static"}]}}}
	174	else:
	175	ospf_red = {
	176	dut: {
	177	"ospf6": {"redistribute": [{"redist_type": "static", "delete": True}]}
	178	}
	179	}
	180	result = create_router_ospf(tgen, topo, ospf_red)
	181	assert result is True, "Testcase : Failed \n Error: {}".format(result)
	182
	183
	184	def red_connected(dut, config=True):
	185	"""Local def for Redstribute connected routes inside ospf."""
	186	global topo
	187	tgen = get_topogen()
	188	if config:
	189	ospf_red = {dut: {"ospf6": {"redistribute": [{"redist_type": "connected"}]}}}
	190	else:
	191	ospf_red = {
	192	dut: {
	193	"ospf6": {
	194	"redistribute": [{"redist_type": "connected", "del_action": True}]
	195	}
196	}
197	}
198	result = create_router_ospf(tgen, topo, ospf_red)
199	assert result is True, "Testcase: Failed \n Error: {}".format(result)
200
201
202	def get_llip(onrouter, intf):
203	"""
dea6dce3	204	API to get the link local ipv6 address of a particular interface
d1b5fa5b	205
	206	Parameters
	207	----------
	208	* `fromnode`: Source node
	209	* `tonode` : interface for which link local ip needs to be returned.
	210
	211	Usage
	212	-----
	213	result = get_llip('r1', 'r2-link0')
	214
	215	Returns
	216	-------
	217	1) link local ipv6 address from the interface.
	218	2) errormsg - when link local ip not found.
	219	"""
	220	tgen = get_topogen()
	221	intf = topo["routers"][onrouter]["links"][intf]["interface"]
	222	llip = get_frr_ipv6_linklocal(tgen, onrouter, intf)
	223	if llip:
	224	logger.info("llip ipv6 address to be set as NH is %s", llip)
	225	return llip
	226	return None
	227
	228
	229	def get_glipv6(onrouter, intf):
	230	"""
dea6dce3	231	API to get the global ipv6 address of a particular interface
d1b5fa5b	232
	233	Parameters
	234	----------
	235	* `onrouter`: Source node
	236	* `intf` : interface for which link local ip needs to be returned.
	237
	238	Usage
	239	-----
	240	result = get_glipv6('r1', 'r2-link0')
	241
	242	Returns
	243	-------
	244	1) global ipv6 address from the interface.
	245	2) errormsg - when link local ip not found.
	246	"""
	247	glipv6 = (topo["routers"][onrouter]["links"][intf]["ipv6"]).split("/")[0]
	248	if glipv6:
	249	logger.info("Global ipv6 address to be set as NH is %s", glipv6)
	250	return glipv6
	251	return None
	252
	253
	254	# ##################################
	255	# Test cases start here.
	256	# ##################################
	257
	258
	259	def test_ospfv3_lan_ecmp_tc18_p0(request):
	260	"""
	261	OSPF ECMP.
	262
	263	Verify OSPF ECMP with max path configured as 8
	264	(Edge having 1 uplink port as broadcast network,
	265	connect to 8 TORs - LAN case)
	266
	267	"""
	268	tc_name = request.node.name
	269	write_test_header(tc_name)
	270	tgen = get_topogen()
	271
	272	# Don't run this test if we have any failure.
	273	if tgen.routers_have_failure():
	274	pytest.skip(tgen.errors)
	275
	276	global topo
	277	step("Bring up the base config as per the topology")
	278	step(". Configure ospf in all the routers on LAN interface.")
	279
	280	reset_config_on_routers(tgen)
	281
	282	step("Verify that OSPF is up with 8 neighborship sessions.")
	283
	284	ospf_covergence = verify_ospf6_neighbor(tgen, topo, lan=True)
74dd0c84	285	assert ospf_covergence is True, "Testcase Failed \n Error: {}".format(
d1b5fa5b	286	ospf_covergence
	287	)
	288
	289	step(
	290	"Configure a static route in all the routes and "
	291	"redistribute static/connected in OSPF."
	292	)
	293
	294	for rtr in topo["routers"]:
	295	input_dict = {
	296	rtr: {
	297	"static_routes": [
	298	{"network": NETWORK["ipv6"][0], "no_of_ip": 5, "next_hop": "Null0"}
	299	]
	300	}
	301	}
	302	result = create_static_routes(tgen, input_dict)
	303	assert result is True, "Testcase {} : Failed \n Error: {}".format(
	304	tc_name, result
	305	)
	306
	307	dut = rtr
	308	red_static(dut)
	309
	310	step(
	311	"Verify that route in R0 in stalled with 8 hops. "
	312	"Verify ospf route table and ip route table."
	313	)
	314
	315	nh = []
	316	for rtr in topo["routers"]:
	317	llip = get_llip(rtr, switch_name)
	318	assert llip is not None, "Testcase {} : Failed \n Error: {}".format(
	319	tc_name, result
	320	)
	321	nh.append(llip)
	322
	323	llip = get_llip("r1", switch_name)
	324	assert llip is not None, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
	325
	326	nh.remove(llip)
	327	dut = "r1"
	328	result = verify_ospf6_rib(tgen, dut, input_dict)
	329	assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
	330
	331	protocol = "ospf6"
	332	result = verify_rib(tgen, "ipv6", dut, input_dict, protocol=protocol)
	333	assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
	334
	335	step(" clear ip ospf interface on DUT(r0)")
	336	clear_ospf(tgen, "r0", ospf="ospf6")
	337
	338	step(
	339	"Verify that after clearing the ospf interface all the "
	340	"neighbours are up and routes are installed with 8 next hop "
	341	"in ospf and ip route tables on R0"
	342	)
	343
	344	dut = "r0"
	345	ospf_covergence = verify_ospf6_neighbor(tgen, topo, dut=dut, lan=True)
74dd0c84	346	assert ospf_covergence is True, "Testcase Failed \n Error: {}".format(
d1b5fa5b	347	ospf_covergence
	348	)
	349
	350	step(" clear ip ospf interface on R2")
	351	clear_ospf(tgen, "r2")
	352
	353	dut = "r2"
	354	ospf_covergence = verify_ospf6_neighbor(tgen, topo, dut=dut, lan=True)
74dd0c84	355	assert ospf_covergence is True, "Testcase Failed \n Error: {}".format(
d1b5fa5b	356	ospf_covergence
	357	)
	358
	359	step("Delete static/connected cmd in ospf in all the routes one by one.")
	360	for rtr in topo["routers"]:
	361	input_dict = {
	362	rtr: {
	363	"static_routes": [
	364	{
	365	"network": NETWORK["ipv6"][0],
	366	"no_of_ip": 5,
	367	"next_hop": "Null0",
	368	"delete": True,
	369	}
	370	]
	371	}
	372	}
	373	result = create_static_routes(tgen, input_dict)
	374	assert result is True, "Testcase {} : Failed \n Error: {}".format(
	375	tc_name, result
	376	)
	377
	378	write_test_footer(tc_name)
	379
	380
	381	if __name__ == "__main__":
	382	args = ["-s"] + sys.argv[1:]
	383	sys.exit(pytest.main(args))