frr/tests/topotests/ospf_basic_functionality/test_ospf_p2mp.py
Acee 0d8ef0477c ospfd: OSPF P2MP Delayed Reflooding configuration
Currently, delayed reflooding on P2MP interfaces for LSAs received
from neighbors on the interface is unconditionally (see commit
c706f0e32b). In some cases, this
change wasn't desirable and this feature makes delayed reflooding
configurable for P2MP interfaces via the CLI command:
"ip ospf network point-to-multipoint delay-reflood" in interface
submode.

Signed-off-by: Acee <aceelindem@gmail.com>
2023-05-22 15:51:41 -04:00

700 lines
21 KiB
Python

#!/usr/bin/python
# SPDX-License-Identifier: ISC
#
# Copyright (c) 2020 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
from copy import deepcopy
from ipaddress import IPv4Address
# 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,
step,
create_interfaces_cfg,
retry,
run_frr_cmd,
)
from lib.topolog import logger
from lib.topojson import build_config_from_json
from lib.topotest import frr_unicode, json_cmp
from lib.ospf import (
verify_ospf_interface,
)
pytestmark = [pytest.mark.ospfd, pytest.mark.staticd]
# Global variables
topo = None
"""
TOPOOLOGY =
Please view in a fixed-width font such as Courier.
+---+ A1 +---+
+R1 +------------+R2 |
+-+-+- +--++
| -- -- |
| -- A0 -- |
A0| ---- |
| ---- | A2
| -- -- |
| -- -- |
+-+-+- +-+-+
+R0 +-------------+R3 |
+---+ A3 +---+
TESTCASES =
1. OSPF P2MP -Verify state change events on p2mp network.
"""
def setup_module(mod):
"""
Sets up the pytest environment
* `mod`: module 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 = "{}/ospf_p2mp.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)
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
tgen.stop_topology()
logger.info(
"Testsuite end time: {}".format(time.asctime(time.localtime(time.time())))
)
logger.info("=" * 40)
# ##################################
# Test cases start here.
# ##################################
def test_ospf_p2mp_tc1_p0(request):
"""OSPF IFSM -Verify state change events on p2mp network."""
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")
reset_config_on_routers(tgen)
step(
"Verify that OSPF is subscribed to multi cast services "
"(All SPF, all DR Routers)."
)
step("Verify that interface is enabled in ospf.")
step("Verify that config is successful.")
dut = "r0"
input_dict = {
"r0": {
"links": {
"r3": {"ospf": {"mcastMemberOspfAllRouters": True, "ospfEnabled": True}}
}
}
}
result = verify_ospf_interface(tgen, topo, dut=dut, input_dict=input_dict)
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
step("Delete the ip address")
topo1 = {
"r0": {
"links": {
"r3": {
"ipv4": topo["routers"]["r0"]["links"]["r3"]["ipv4"],
"interface": topo["routers"]["r0"]["links"]["r3"]["interface"],
"delete": True,
}
}
}
}
result = create_interfaces_cfg(tgen, topo1)
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
step("Change the ip on the R0 interface")
topo_modify_change_ip = deepcopy(topo)
intf_ip = topo_modify_change_ip["routers"]["r0"]["links"]["r3"]["ipv4"]
topo_modify_change_ip["routers"]["r0"]["links"]["r3"]["ipv4"] = str(
IPv4Address(frr_unicode(intf_ip.split("/")[0])) + 3
) + "/{}".format(intf_ip.split("/")[1])
build_config_from_json(tgen, topo_modify_change_ip, save_bkup=False)
step("Verify that interface is enabled in ospf.")
dut = "r0"
input_dict = {
"r0": {
"links": {
"r3": {
"ospf": {
"ipAddress": topo_modify_change_ip["routers"]["r0"]["links"][
"r3"
]["ipv4"].split("/")[0],
"ipAddressPrefixlen": int(
topo_modify_change_ip["routers"]["r0"]["links"]["r3"][
"ipv4"
].split("/")[1]
),
}
}
}
}
}
result = verify_ospf_interface(tgen, topo, dut=dut, input_dict=input_dict)
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
step("Modify the mask on the R0 interface")
ip_addr = topo_modify_change_ip["routers"]["r0"]["links"]["r3"]["ipv4"]
mask = topo_modify_change_ip["routers"]["r0"]["links"]["r3"]["ipv4"]
step("Delete the ip address")
topo1 = {
"r0": {
"links": {
"r3": {
"ipv4": ip_addr,
"interface": topo["routers"]["r0"]["links"]["r3"]["interface"],
"delete": True,
}
}
}
}
result = create_interfaces_cfg(tgen, topo1)
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
step("Change the ip on the R0 interface")
topo_modify_change_ip = deepcopy(topo)
intf_ip = topo_modify_change_ip["routers"]["r0"]["links"]["r3"]["ipv4"]
topo_modify_change_ip["routers"]["r0"]["links"]["r3"]["ipv4"] = str(
IPv4Address(frr_unicode(intf_ip.split("/")[0])) + 3
) + "/{}".format(int(intf_ip.split("/")[1]) + 1)
build_config_from_json(tgen, topo_modify_change_ip, save_bkup=False)
step("Verify that interface is enabled in ospf.")
dut = "r0"
input_dict = {
"r0": {
"links": {
"r3": {
"ospf": {
"ipAddress": topo_modify_change_ip["routers"]["r0"]["links"][
"r3"
]["ipv4"].split("/")[0],
"ipAddressPrefixlen": int(
topo_modify_change_ip["routers"]["r0"]["links"]["r3"][
"ipv4"
].split("/")[1]
),
}
}
}
}
}
result = verify_ospf_interface(tgen, topo, dut=dut, input_dict=input_dict)
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
topo1 = {
"r0": {
"links": {
"r3": {
"ipv4": topo_modify_change_ip["routers"]["r0"]["links"]["r3"][
"ipv4"
],
"interface": topo_modify_change_ip["routers"]["r0"]["links"]["r3"][
"interface"
],
"delete": True,
}
}
}
}
result = create_interfaces_cfg(tgen, topo1)
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
build_config_from_json(tgen, topo, save_bkup=False)
step("Change the area id on the interface")
input_dict = {
"r0": {
"links": {
"r3": {
"interface": topo["routers"]["r0"]["links"]["r3"]["interface"],
"ospf": {"area": "0.0.0.0"},
"delete": True,
}
}
}
}
result = create_interfaces_cfg(tgen, input_dict)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
input_dict = {
"r0": {
"links": {
"r3": {
"interface": topo["routers"]["r0"]["links"]["r3"]["interface"],
"ospf": {"area": "0.0.0.1"},
}
}
}
}
result = create_interfaces_cfg(tgen, input_dict)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("Verify that interface is enabled in ospf.")
dut = "r0"
input_dict = {
"r0": {"links": {"r3": {"ospf": {"area": "0.0.0.1", "ospfEnabled": True}}}}
}
result = verify_ospf_interface(tgen, topo, dut=dut, input_dict=input_dict)
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
input_dict = {
"r0": {
"links": {
"r3": {
"interface": topo["routers"]["r0"]["links"]["r3"]["interface"],
"ospf": {"area": "0.0.0.1"},
"delete": True,
}
}
}
}
result = create_interfaces_cfg(tgen, input_dict)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
input_dict = {
"r0": {
"links": {
"r3": {
"interface": topo["routers"]["r0"]["links"]["r3"]["interface"],
"ospf": {"area": "0.0.0.0"},
}
}
}
}
result = create_interfaces_cfg(tgen, input_dict)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("Verify if interface is enabled with network type P2MP")
input_dict = {
"r0": {
"links": {
"r3": {
"interface": topo["routers"]["r0"]["links"]["r3"]["interface"],
"ospf": {"area": "0.0.0.0", "networkType": "POINTOMULTIPOINT"},
}
}
}
}
result = create_interfaces_cfg(tgen, input_dict)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
write_test_footer(tc_name)
def test_ospf_p2mp_tc_delay_reflood(request):
"""OSPF IFSM -Verify "delay-reflood" parameter in p2mp network."""
tc_name = request.node.name
write_test_header(tc_name)
tgen = get_topogen()
r0 = tgen.gears["r0"]
# Don't run this test if we have any failure.
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
global topo
step("Verify for interface with network type P2MP that delay-reflood is configured")
r0.vtysh_multicmd(
"conf t\ninterface r0-r1-eth0\nip ospf network point-to-multipoint delay-reflood"
)
dut = "r0"
input_dict = {
"r0": {
"links": {
"r1": {
"ospf": {
"mcastMemberOspfAllRouters": True,
"ospfEnabled": True,
"networkType": "POINTOMULTIPOINT",
"p2mpDelayReflood": True,
}
},
"r2": {
"ospf": {
"mcastMemberOspfAllRouters": True,
"ospfEnabled": True,
"networkType": "POINTOMULTIPOINT",
"p2mpDelayReflood": False,
}
},
"r3": {
"ospf": {
"mcastMemberOspfAllRouters": True,
"ospfEnabled": True,
"networkType": "POINTOMULTIPOINT",
"p2mpDelayReflood": False,
}
},
}
}
}
result = verify_ospf_interface(tgen, topo, dut=dut, input_dict=input_dict)
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
delay_reflood_cfg = (
tgen.net["r0"]
.cmd(
'vtysh -c "show running" | grep "^ ip ospf network point-to-multipoint delay-reflood"'
)
.rstrip()
)
assertmsg = "delay-reflood' configuration applied, but not present in configuration"
assert (
delay_reflood_cfg == " ip ospf network point-to-multipoint delay-reflood"
), assertmsg
step("Verify for interface with network type P2MP that delay-reflood is removed")
r0.vtysh_multicmd(
"conf t\ninterface r0-r1-eth0\nip ospf network point-to-multipoint"
)
input_dict = {
"r0": {
"links": {
"r1": {
"ospf": {
"mcastMemberOspfAllRouters": True,
"ospfEnabled": True,
"networkType": "POINTOMULTIPOINT",
"p2mpDelayReflood": False,
}
},
"r2": {
"ospf": {
"mcastMemberOspfAllRouters": True,
"ospfEnabled": True,
"networkType": "POINTOMULTIPOINT",
"p2mpDelayReflood": False,
}
},
"r3": {
"ospf": {
"mcastMemberOspfAllRouters": True,
"ospfEnabled": True,
"networkType": "POINTOMULTIPOINT",
"p2mpDelayReflood": False,
}
},
}
}
}
result = verify_ospf_interface(tgen, topo, dut=dut, input_dict=input_dict)
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
delay_reflood_cfg = (
tgen.net["r0"]
.cmd(
'vtysh -c "show running" | grep "^ ip ospf network point-to-multipoint delay-reflood"'
)
.rstrip()
)
assertmsg = (
"delay-reflood' configuration removed, but still present in configuration"
)
assert (
delay_reflood_cfg != " ip ospf network point-to-multipoint delay-reflood"
), assertmsg
step(
"Verify for interface with network type P2MP that delay-reflood is removed with removal of network type"
)
r0.vtysh_multicmd(
"conf t\ninterface r0-r1-eth0\nip ospf network point-to-multipoint delay-reflood"
)
r0.vtysh_multicmd(
"conf t\ninterface r0-r1-eth0\nno ip ospf network point-to-multipoint"
)
r0.vtysh_multicmd(
"conf t\ninterface r0-r1-eth0\nip ospf network point-to-multipoint"
)
input_dict = {
"r0": {
"links": {
"r1": {
"ospf": {
"mcastMemberOspfAllRouters": True,
"ospfEnabled": True,
"networkType": "POINTOMULTIPOINT",
"p2mpDelayReflood": False,
}
},
"r2": {
"ospf": {
"mcastMemberOspfAllRouters": True,
"ospfEnabled": True,
"networkType": "POINTOMULTIPOINT",
"p2mpDelayReflood": False,
}
},
"r3": {
"ospf": {
"mcastMemberOspfAllRouters": True,
"ospfEnabled": True,
"networkType": "POINTOMULTIPOINT",
"p2mpDelayReflood": False,
}
},
}
}
}
result = verify_ospf_interface(tgen, topo, dut=dut, input_dict=input_dict)
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
delay_reflood_cfg = (
tgen.net["r0"]
.cmd(
'vtysh -c "show running" | grep "^ ip ospf network point-to-multipoint delay-reflood"'
)
.rstrip()
)
assertmsg = (
"delay-reflood' configuration removed, but still present in configuration"
)
assert (
delay_reflood_cfg != " ip ospf network point-to-multipoint delay-reflood"
), assertmsg
write_test_footer(tc_name)
@retry(retry_timeout=30)
def verify_ospf_json(tgen, dut, input_dict, cmd="show ip ospf database json"):
del tgen
show_ospf_json = run_frr_cmd(dut, cmd, isjson=True)
if not bool(show_ospf_json):
return "ospf is not running"
result = json_cmp(show_ospf_json, input_dict)
return str(result) if result else None
@pytest.mark.parametrize("tgen", [2], indirect=True)
def test_ospf_nbrs(tgen):
db_full = {
"areas": {
"0.0.0.0": {
"routerLinkStates": [
{
"lsId": "100.1.1.0",
"advertisedRouter": "100.1.1.0",
"numOfRouterLinks": 6,
},
{
"lsId": "100.1.1.1",
"advertisedRouter": "100.1.1.1",
"numOfRouterLinks": 6,
},
{
"lsId": "100.1.1.2",
"advertisedRouter": "100.1.1.2",
"numOfRouterLinks": 6,
},
{
"lsId": "100.1.1.3",
"advertisedRouter": "100.1.1.3",
"numOfRouterLinks": 7,
},
]
}
}
}
input = [
[
"r0",
"show ip ospf n json",
{
"neighbors": {
"100.1.1.1": [
{
"nbrState": "Full/DROther",
}
],
"100.1.1.2": [
{
"nbrState": "Full/DROther",
}
],
"100.1.1.3": [
{
"nbrState": "Full/DROther",
}
],
}
},
],
[
"r1",
"show ip ospf n json",
{
"neighbors": {
"100.1.1.0": [
{
"nbrState": "Full/DROther",
}
],
"100.1.1.2": [
{
"nbrState": "Full/DROther",
}
],
"100.1.1.3": [
{
"nbrState": "Full/DROther",
}
],
}
},
],
[
"r2",
"show ip ospf n json",
{
"neighbors": {
"100.1.1.0": [
{
"nbrState": "Full/DROther",
}
],
"100.1.1.1": [
{
"nbrState": "Full/DROther",
}
],
"100.1.1.3": [
{
"nbrState": "Full/DROther",
}
],
}
},
],
[
"r3",
"show ip ospf n json",
{
"neighbors": {
"100.1.1.0": [
{
"nbrState": "Full/DROther",
}
],
"100.1.1.1": [
{
"nbrState": "Full/DROther",
}
],
"100.1.1.2": [
{
"nbrState": "Full/DROther",
}
],
}
},
],
["r0", "show ip ospf database json", db_full],
["r1", "show ip ospf database json", db_full],
["r2", "show ip ospf database json", db_full],
["r3", "show ip ospf database json", db_full],
["r0", "show ip ospf database json", db_full],
["r0", "show ip ospf database router json", {}],
["r0", "show ip ospf interface traffic json", {}],
["r1", "show ip ospf interface traffic json", {}],
["r2", "show ip ospf interface traffic json", {}],
["r3", "show ip ospf interface traffic json", {}],
]
for cmd_set in input:
step("test_ospf: %s - %s" % (cmd_set[0], cmd_set[1]))
assert (
verify_ospf_json(tgen, tgen.gears[cmd_set[0]], cmd_set[2], cmd_set[1])
is None
)
if __name__ == "__main__":
args = ["-s"] + sys.argv[1:]
sys.exit(pytest.main(args))