[mirror_frr.git] / tests / topotests / ospf_basic_functionality / test_ospf_p2mp.py

#!/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)


@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": [
                        {
                            "state": "Full/DROther",
                        }
                    ],
                    "100.1.1.2": [
                        {
                            "state": "Full/DROther",
                        }
                    ],
                    "100.1.1.3": [
                        {
                            "state": "Full/DROther",
                        }
                    ],
                }
            },
        ],
        [
            "r1",
            "show ip ospf n json",
            {
                "neighbors": {
                    "100.1.1.0": [
                        {
                            "state": "Full/DROther",
                        }
                    ],
                    "100.1.1.2": [
                        {
                            "state": "Full/DROther",
                        }
                    ],
                    "100.1.1.3": [
                        {
                            "state": "Full/DROther",
                        }
                    ],
                }
            },
        ],
        [
            "r2",
            "show ip ospf n json",
            {
                "neighbors": {
                    "100.1.1.0": [
                        {
                            "state": "Full/DROther",
                        }
                    ],
                    "100.1.1.1": [
                        {
                            "state": "Full/DROther",
                        }
                    ],
                    "100.1.1.3": [
                        {
                            "state": "Full/DROther",
                        }
                    ],
                }
            },
        ],
        [
            "r3",
            "show ip ospf n json",
            {
                "neighbors": {
                    "100.1.1.0": [
                        {
                            "state": "Full/DROther",
                        }
                    ],
                    "100.1.1.1": [
                        {
                            "state": "Full/DROther",
                        }
                    ],
                    "100.1.1.2": [
                        {
                            "state": "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))
Commit	Line	Data
dc5298d7	1	#!/usr/bin/python
acddc0ed	2	# SPDX-License-Identifier: ISC
dc5298d7	3
	4	#
	5	# Copyright (c) 2020 by VMware, Inc. ("VMware")
	6	# Used Copyright (c) 2018 by Network Device Education Foundation, Inc.
	7	# ("NetDEF") in this file.
	8	#
dc5298d7	9
	10
	11	"""OSPF Basic Functionality Automation."""
	12	import os
	13	import sys
	14	import time
	15	import pytest
dc5298d7	16	from copy import deepcopy
	17	from ipaddress import IPv4Address
	18
	19	# Save the Current Working Directory to find configuration files.
	20	CWD = os.path.dirname(os.path.realpath(__file__))
	21	sys.path.append(os.path.join(CWD, "../"))
	22	sys.path.append(os.path.join(CWD, "../lib/"))
	23
	24	# pylint: disable=C0413
	25	# Import topogen and topotest helpers
dc5298d7	26	from lib.topogen import Topogen, get_topogen
dc5298d7	27
	28	# Import topoJson from lib, to create topology and initial configuration
	29	from lib.common_config import (
	30	start_topology,
	31	write_test_header,
	32	write_test_footer,
	33	reset_config_on_routers,
dc5298d7	34	step,
dc5298d7	35	create_interfaces_cfg,
675ba67d LB	36	retry,
675ba67d LB	37	run_frr_cmd,
dc5298d7	38	)
dc5298d7	39	from lib.topolog import logger
4953ca97	40	from lib.topojson import build_config_from_json
675ba67d	41	from lib.topotest import frr_unicode, json_cmp
dc5298d7	42
dc5298d7	43	from lib.ospf import (
dc5298d7	44	verify_ospf_interface,
dc5298d7	45	)
dc5298d7	46
6ff492b1	47
e82b531d	48	pytestmark = [pytest.mark.ospfd, pytest.mark.staticd]
6ff492b1	49
dc5298d7	50	# Global variables
	51	topo = None
	52
dc5298d7	53
	54	"""
	55	TOPOOLOGY =
	56	Please view in a fixed-width font such as Courier.
	57	+---+ A1 +---+
	58	+R1 +------------+R2 \|
	59	+-+-+- +--++
	60	\| -- -- \|
	61	\| -- A0 -- \|
	62	A0\| ---- \|
	63	\| ---- \| A2
	64	\| -- -- \|
	65	\| -- -- \|
	66	+-+-+- +-+-+
	67	+R0 +-------------+R3 \|
	68	+---+ A3 +---+
	69
	70	TESTCASES =
	71	1. OSPF P2MP -Verify state change events on p2mp network.
	72	"""
	73
	74
dc5298d7	75	def setup_module(mod):
	76	"""
	77	Sets up the pytest environment
	78
	79	* `mod`: module name
	80	"""
dc5298d7	81	testsuite_run_time = time.asctime(time.localtime(time.time()))
	82	logger.info("Testsuite start time: {}".format(testsuite_run_time))
	83	logger.info("=" * 40)
	84
	85	logger.info("Running setup_module to create topology")
	86
	87	# This function initiates the topology build with Topogen...
e82b531d CH	88	json_file = "{}/ospf_p2mp.json".format(CWD)
	89	tgen = Topogen(json_file, mod.__name__)
	90	global topo
	91	topo = tgen.json_topo
dc5298d7	92	# ... and here it calls Mininet initialization functions.
dc5298d7	93
dc5298d7	94	# Starting topology, create tmp files which are loaded to routers
d60a3f0e	95	# to start daemons and then start routers
991a971f	96	start_topology(tgen)
dc5298d7	97
	98	# Creating configuration from JSON
	99	build_config_from_json(tgen, topo)
	100
	101	# Don't run this test if we have any failure.
	102	if tgen.routers_have_failure():
	103	pytest.skip(tgen.errors)
	104
dc5298d7	105	logger.info("Running setup_module() done")
	106
	107
	108	def teardown_module(mod):
	109	"""
	110	Teardown the pytest environment.
	111
	112	* `mod`: module name
	113	"""
	114
	115	logger.info("Running teardown_module to delete topology")
	116
	117	tgen = get_topogen()
	118
	119	# Stop toplogy and Remove tmp files
	120	tgen.stop_topology()
	121
	122	logger.info(
	123	"Testsuite end time: {}".format(time.asctime(time.localtime(time.time())))
	124	)
	125	logger.info("=" * 40)
	126
	127
	128	# ##################################
	129	# Test cases start here.
	130	# ##################################
	131
	132
	133	def test_ospf_p2mp_tc1_p0(request):
	134	"""OSPF IFSM -Verify state change events on p2mp network."""
	135	tc_name = request.node.name
	136	write_test_header(tc_name)
	137	tgen = get_topogen()
	138
	139	# Don't run this test if we have any failure.
	140	if tgen.routers_have_failure():
	141	pytest.skip(tgen.errors)
	142
	143	global topo
	144	step("Bring up the base config as per the topology")
	145	reset_config_on_routers(tgen)
	146	step(
	147	"Verify that OSPF is subscribed to multi cast services "
	148	"(All SPF, all DR Routers)."
	149	)
	150	step("Verify that interface is enabled in ospf.")
	151	step("Verify that config is successful.")
	152	dut = "r0"
	153	input_dict = {
	154	"r0": {
	155	"links": {
	156	"r3": {"ospf": {"mcastMemberOspfAllRouters": True, "ospfEnabled": True}}
	157	}
	158	}
	159	}
	160	result = verify_ospf_interface(tgen, topo, dut=dut, input_dict=input_dict)
	161	assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
	162
	163	step("Delete the ip address")
	164	topo1 = {
	165	"r0": {
	166	"links": {
	167	"r3": {
	168	"ipv4": topo["routers"]["r0"]["links"]["r3"]["ipv4"],
169	"interface": topo["routers"]["r0"]["links"]["r3"]["interface"],
170	"delete": True,
171	}
172	}
173	}
174	}
175
176	result = create_interfaces_cfg(tgen, topo1)
177	assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
178
179	step("Change the ip on the R0 interface")
180
181	topo_modify_change_ip = deepcopy(topo)
182	intf_ip = topo_modify_change_ip["routers"]["r0"]["links"]["r3"]["ipv4"]
183	topo_modify_change_ip["routers"]["r0"]["links"]["r3"]["ipv4"] = str(
d7d21c3a	184	IPv4Address(frr_unicode(intf_ip.split("/")[0])) + 3
dc5298d7	185	) + "/{}".format(intf_ip.split("/")[1])
	186
	187	build_config_from_json(tgen, topo_modify_change_ip, save_bkup=False)
	188	step("Verify that interface is enabled in ospf.")
	189	dut = "r0"
	190	input_dict = {
	191	"r0": {
	192	"links": {
	193	"r3": {
	194	"ospf": {
	195	"ipAddress": topo_modify_change_ip["routers"]["r0"]["links"][
	196	"r3"
	197	]["ipv4"].split("/")[0],
	198	"ipAddressPrefixlen": int(
	199	topo_modify_change_ip["routers"]["r0"]["links"]["r3"][
	200	"ipv4"
	201	].split("/")[1]
	202	),
	203	}
	204	}
	205	}
	206	}
	207	}
	208	result = verify_ospf_interface(tgen, topo, dut=dut, input_dict=input_dict)
	209	assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
	210
	211	step("Modify the mask on the R0 interface")
	212	ip_addr = topo_modify_change_ip["routers"]["r0"]["links"]["r3"]["ipv4"]
	213	mask = topo_modify_change_ip["routers"]["r0"]["links"]["r3"]["ipv4"]
	214	step("Delete the ip address")
	215	topo1 = {
	216	"r0": {
	217	"links": {
	218	"r3": {
	219	"ipv4": ip_addr,
	220	"interface": topo["routers"]["r0"]["links"]["r3"]["interface"],
	221	"delete": True,
	222	}
	223	}
	224	}
	225	}
	226
	227	result = create_interfaces_cfg(tgen, topo1)
	228	assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
	229
	230	step("Change the ip on the R0 interface")
	231
	232	topo_modify_change_ip = deepcopy(topo)
	233	intf_ip = topo_modify_change_ip["routers"]["r0"]["links"]["r3"]["ipv4"]
	234	topo_modify_change_ip["routers"]["r0"]["links"]["r3"]["ipv4"] = str(
d7d21c3a	235	IPv4Address(frr_unicode(intf_ip.split("/")[0])) + 3
dc5298d7	236	) + "/{}".format(int(intf_ip.split("/")[1]) + 1)
	237
	238	build_config_from_json(tgen, topo_modify_change_ip, save_bkup=False)
	239	step("Verify that interface is enabled in ospf.")
	240	dut = "r0"
	241	input_dict = {
	242	"r0": {
	243	"links": {
	244	"r3": {
	245	"ospf": {
	246	"ipAddress": topo_modify_change_ip["routers"]["r0"]["links"][
	247	"r3"
	248	]["ipv4"].split("/")[0],
	249	"ipAddressPrefixlen": int(
	250	topo_modify_change_ip["routers"]["r0"]["links"]["r3"][
	251	"ipv4"
	252	].split("/")[1]
	253	),
	254	}
	255	}
	256	}
	257	}
	258	}
	259	result = verify_ospf_interface(tgen, topo, dut=dut, input_dict=input_dict)
	260	assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
	261
	262	topo1 = {
	263	"r0": {
	264	"links": {
	265	"r3": {
	266	"ipv4": topo_modify_change_ip["routers"]["r0"]["links"]["r3"][
	267	"ipv4"
	268	],
	269	"interface": topo_modify_change_ip["routers"]["r0"]["links"]["r3"][
	270	"interface"
	271	],
	272	"delete": True,
	273	}
	274	}
	275	}
	276	}
	277
	278	result = create_interfaces_cfg(tgen, topo1)
	279	assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
	280
	281	build_config_from_json(tgen, topo, save_bkup=False)
	282
	283	step("Change the area id on the interface")
	284	input_dict = {
	285	"r0": {
	286	"links": {
	287	"r3": {
	288	"interface": topo["routers"]["r0"]["links"]["r3"]["interface"],
	289	"ospf": {"area": "0.0.0.0"},
	290	"delete": True,
	291	}
	292	}
	293	}
	294	}
	295
	296	result = create_interfaces_cfg(tgen, input_dict)
	297	assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
	298
	299	input_dict = {
300	"r0": {
301	"links": {
302	"r3": {
303	"interface": topo["routers"]["r0"]["links"]["r3"]["interface"],
304	"ospf": {"area": "0.0.0.1"},
305	}
306	}
307	}
308	}
309
310	result = create_interfaces_cfg(tgen, input_dict)
311	assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
312	step("Verify that interface is enabled in ospf.")
313	dut = "r0"
314	input_dict = {
315	"r0": {"links": {"r3": {"ospf": {"area": "0.0.0.1", "ospfEnabled": True}}}}
316	}
317	result = verify_ospf_interface(tgen, topo, dut=dut, input_dict=input_dict)
318	assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
319
320	input_dict = {
321	"r0": {
322	"links": {
323	"r3": {
324	"interface": topo["routers"]["r0"]["links"]["r3"]["interface"],
325	"ospf": {"area": "0.0.0.1"},
326	"delete": True,
327	}
328	}
329	}
330	}
331
332	result = create_interfaces_cfg(tgen, input_dict)
333	assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
334
335	input_dict = {
336	"r0": {
337	"links": {
338	"r3": {
339	"interface": topo["routers"]["r0"]["links"]["r3"]["interface"],
340	"ospf": {"area": "0.0.0.0"},
341	}
342	}
343	}
344	}
345
346	result = create_interfaces_cfg(tgen, input_dict)
347	assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
348
349	step("Verify if interface is enabled with network type P2MP")
350	input_dict = {
351	"r0": {
352	"links": {
353	"r3": {
354	"interface": topo["routers"]["r0"]["links"]["r3"]["interface"],
5980ad0a	355	"ospf": {"area": "0.0.0.0", "networkType": "POINTOMULTIPOINT"},
dc5298d7	356	}
	357	}
	358	}
	359	}
	360	result = create_interfaces_cfg(tgen, input_dict)
	361	assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
	362
	363	write_test_footer(tc_name)
	364
	365
675ba67d LB	366	@retry(retry_timeout=30)
	367	def verify_ospf_json(tgen, dut, input_dict, cmd="show ip ospf database json"):
	368	del tgen
	369	show_ospf_json = run_frr_cmd(dut, cmd, isjson=True)
	370	if not bool(show_ospf_json):
	371	return "ospf is not running"
	372	result = json_cmp(show_ospf_json, input_dict)
	373	return str(result) if result else None
	374
	375
	376	@pytest.mark.parametrize("tgen", [2], indirect=True)
	377	def test_ospf_nbrs(tgen):
	378	db_full = {
	379	"areas": {
	380	"0.0.0.0": {
	381	"routerLinkStates": [
	382	{
	383	"lsId": "100.1.1.0",
	384	"advertisedRouter": "100.1.1.0",
	385	"numOfRouterLinks": 6,
	386	},
	387	{
	388	"lsId": "100.1.1.1",
	389	"advertisedRouter": "100.1.1.1",
	390	"numOfRouterLinks": 6,
	391	},
	392	{
	393	"lsId": "100.1.1.2",
	394	"advertisedRouter": "100.1.1.2",
	395	"numOfRouterLinks": 6,
	396	},
	397	{
	398	"lsId": "100.1.1.3",
	399	"advertisedRouter": "100.1.1.3",
	400	"numOfRouterLinks": 7,
	401	},
	402	]
	403	}
	404	}
	405	}
	406	input = [
	407	[
	408	"r0",
	409	"show ip ospf n json",
	410	{
	411	"neighbors": {
	412	"100.1.1.1": [
	413	{
	414	"state": "Full/DROther",
	415	}
	416	],
	417	"100.1.1.2": [
	418	{
	419	"state": "Full/DROther",
	420	}
	421	],
	422	"100.1.1.3": [
	423	{
	424	"state": "Full/DROther",
	425	}
	426	],
	427	}
	428	},
	429	],
430	[
431	"r1",
432	"show ip ospf n json",
433	{
434	"neighbors": {
435	"100.1.1.0": [
436	{
437	"state": "Full/DROther",
438	}
439	],
440	"100.1.1.2": [
441	{
442	"state": "Full/DROther",
443	}
444	],
445	"100.1.1.3": [
446	{
447	"state": "Full/DROther",
448	}
449	],
450	}
451	},
452	],
453	[
454	"r2",
455	"show ip ospf n json",
456	{
457	"neighbors": {
458	"100.1.1.0": [
459	{
460	"state": "Full/DROther",
461	}
462	],
463	"100.1.1.1": [
464	{
465	"state": "Full/DROther",
466	}
467	],
468	"100.1.1.3": [
469	{
470	"state": "Full/DROther",
471	}
472	],
473	}
474	},
475	],
476	[
477	"r3",
478	"show ip ospf n json",
479	{
480	"neighbors": {
481	"100.1.1.0": [
482	{
483	"state": "Full/DROther",
484	}
485	],
486	"100.1.1.1": [
487	{
488	"state": "Full/DROther",
489	}
490	],
491	"100.1.1.2": [
492	{
493	"state": "Full/DROther",
494	}
495	],
496	}
497	},
498	],
499	["r0", "show ip ospf database json", db_full],
500	["r1", "show ip ospf database json", db_full],
501	["r2", "show ip ospf database json", db_full],
502	["r3", "show ip ospf database json", db_full],
503	["r0", "show ip ospf database json", db_full],
504	["r0", "show ip ospf database router json", {}],
505	["r0", "show ip ospf interface traffic json", {}],
506	["r1", "show ip ospf interface traffic json", {}],
507	["r2", "show ip ospf interface traffic json", {}],
508	["r3", "show ip ospf interface traffic json", {}],
509	]
510	for cmd_set in input:
511	step("test_ospf: %s - %s" % (cmd_set[0], cmd_set[1]))
512	assert (
513	verify_ospf_json(tgen, tgen.gears[cmd_set[0]], cmd_set[2], cmd_set[1])
514	is None
515	)
516
517
dc5298d7	518	if __name__ == "__main__":
	519	args = ["-s"] + sys.argv[1:]
	520	sys.exit(pytest.main(args))