[mirror_frr.git] / tests / topotests / bfd_ospf_topo1 / test_bfd_ospf_topo1.py

#!/usr/bin/env python

#
# test_bfd_ospf_topo1.py
# Part of NetDEF Topology Tests
#
# Copyright (c) 2020 by
# Network Device Education Foundation, Inc. ("NetDEF")
#
# 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 NETDEF DISCLAIMS ALL WARRANTIES
# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NETDEF 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.
#

"""
test_bfd_ospf_topo1.py:

                        +---------+
                        |         |
           eth-rt2 (.1) |   RT1   | eth-rt3 (.1)
             +----------+ 1.1.1.1 +----------+
             |          |         |          |
             |          +---------+          |
             |                               |
             |                   10.0.2.0/24 |
             |                               |
             |                       eth-rt1 | (.2)
             | 10.0.1.0/24              +----+----+
             |                          |         |
             |                          |   RT3   |
             |                          | 3.3.3.3 |
             |                          |         |
        (.2) | eth-rt1                  +----+----+
        +----+----+                  eth-rt4 | (.1)
        |         |                          |
        |   RT2   |                          |
        | 2.2.2.2 |              10.0.4.0/24 |
        |         |                          |
        +----+----+                          |
        (.1) | eth-rt5               eth-rt3 | (.2)
             |                          +----+----+
             |                          |         |
             |                          |   RT4   |
             |                          | 4.4.4.4 |
             |                          |         |
             |                          +----+----+
             | 10.0.3.0/24           eth-rt5 | (.1)
             |                               |
             |                               |
             |                   10.0.5.0/24 |
             |                               |
             |          +---------+          |
             |          |         |          |
             +----------+   RT5   +----------+
           eth-rt2 (.2) | 5.5.5.5 | eth-rt4 (.2)
                        |         |
                        +---------+

"""

import os
import sys
import pytest
import json
from time import sleep
from functools import partial

# 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 import topotest
from lib.topogen import Topogen, TopoRouter, get_topogen
from lib.topolog import logger

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


def setup_module(mod):
    "Sets up the pytest environment"
    topodef = {
        "s1": ("rt1:eth-rt2", "rt2:eth-rt1"),
        "s2": ("rt1:eth-rt3", "rt3:eth-rt1"),
        "s3": ("rt2:eth-rt5", "rt5:eth-rt2"),
        "s4": ("rt3:eth-rt4", "rt4:eth-rt3"),
        "s5": ("rt4:eth-rt5", "rt5:eth-rt4"),
    }
    tgen = Topogen(topodef, mod.__name__)
    tgen.start_topology()

    router_list = tgen.routers()

    # For all registered routers, load the zebra configuration file
    for rname, router in router_list.items():
        router.load_config(
            TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
        )
        router.load_config(
            TopoRouter.RD_BFD, os.path.join(CWD, "{}/bfdd.conf".format(rname))
        )
        router.load_config(
            TopoRouter.RD_OSPF, os.path.join(CWD, "{}/ospfd.conf".format(rname))
        )
        router.load_config(
            TopoRouter.RD_OSPF6, os.path.join(CWD, "{}/ospf6d.conf".format(rname))
        )

    tgen.start_router()


def teardown_module(mod):
    "Teardown the pytest environment"
    tgen = get_topogen()

    # This function tears down the whole topology.
    tgen.stop_topology()


def print_cmd_result(rname, command):
    print(get_topogen().gears[rname].vtysh_cmd(command, isjson=False))


def router_compare_json_output(rname, command, reference, count=40, wait=2):
    "Compare router JSON output"

    logger.info('Comparing router "%s" "%s" output', rname, command)

    tgen = get_topogen()
    filename = "{}/{}/{}".format(CWD, rname, reference)
    expected = json.loads(open(filename).read())

    # Run test function until we get an result. Wait at most 80 seconds.
    test_func = partial(topotest.router_json_cmp, tgen.gears[rname], command, expected)
    _, diff = topotest.run_and_expect(test_func, None, count=count, wait=wait)
    assertmsg = '"{}" JSON output mismatches the expected result'.format(rname)
    assert diff is None, assertmsg


## TEST STEPS


def test_rib_ospf_step1():
    logger.info("Test (step 1): verify RIB for OSPF")
    tgen = get_topogen()

    # Skip if previous fatal error condition is raised
    if tgen.routers_have_failure():
        pytest.skip(tgen.errors)

    router_compare_json_output(
        "rt1", "show ip route ospf json", "step1/show_ip_route.ref"
    )
    router_compare_json_output(
        "rt1", "show ipv6 route ospf json", "step1/show_ipv6_route.ref"
    )


def test_bfd_ospf_sessions_step2():
    logger.info("Test (step 2): verify BFD peers for OSPF")
    tgen = get_topogen()

    # Skip if previous fatal error condition is raised
    if tgen.routers_have_failure():
        pytest.skip(tgen.errors)

    # BFD is just used on three routers
    for rt in ["rt1", "rt2", "rt3"]:
        router_compare_json_output(
            rt, "show bfd peers json", "step2/show_bfd_peers.ref"
        )


def test_bfd_ospf_interface_failure_rt2_step3():
    logger.info("Test (step 3): Check failover handling with RT2 down")
    tgen = get_topogen()

    # Skip if previous fatal error condition is raised
    if tgen.routers_have_failure():
        pytest.skip(tgen.errors)

    # Let's kill the interface on rt2 and see what happens with the RIB and BFD on rt1
    tgen.gears["rt2"].link_enable("eth-rt1", enabled=False)

    # By default BFD provides a recovery time of 900ms plus jitter, so let's wait
    # initial 2 seconds to let the CI not suffer.
    topotest.sleep(2, 'Wait for BFD down notification')

    router_compare_json_output(
        "rt1", "show ip route ospf json", "step3/show_ip_route_rt2_down.ref", 10, 2
    )
    router_compare_json_output(
        "rt1", "show ipv6 route ospf json", "step3/show_ipv6_route_rt2_down.ref", 1, 0
    )
    router_compare_json_output(
        "rt1", "show bfd peers json", "step3/show_bfd_peers_rt2_down.ref", 1, 0
    )

    # Check recovery, this can take some time
    tgen.gears["rt2"].link_enable("eth-rt1", enabled=True)

    router_compare_json_output(
        "rt1", "show ip route ospf json", "step3/show_ip_route_healthy.ref"
    )
    router_compare_json_output(
        "rt1", "show ipv6 route ospf json", "step3/show_ipv6_route_healthy.ref"
    )
    router_compare_json_output(
        "rt1", "show bfd peers json", "step3/show_bfd_peers_healthy.ref"
    )


def test_bfd_ospf_interface_failure_rt3_step3():
    logger.info("Test (step 3): Check failover handling with RT3 down")
    tgen = get_topogen()

    # Skip if previous fatal error condition is raised
    if tgen.routers_have_failure():
        pytest.skip(tgen.errors)

    # Let's kill the interface on rt3 and see what happens with the RIB and BFD on rt1
    tgen.gears["rt3"].link_enable("eth-rt1", enabled=False)

    # By default BFD provides a recovery time of 900ms plus jitter, so let's wait
    # initial 2 seconds to let the CI not suffer.
    topotest.sleep(2, 'Wait for BFD down notification')
    router_compare_json_output(
        "rt1", "show ip route ospf json", "step3/show_ip_route_rt3_down.ref", 10, 2
    )
    router_compare_json_output(
        "rt1", "show ipv6 route ospf json", "step3/show_ipv6_route_rt3_down.ref", 1, 0
    )
    router_compare_json_output(
        "rt1", "show bfd peers json", "step3/show_bfd_peers_rt3_down.ref", 1, 0
    )

    # Check recovery, this can take some time
    tgen.gears["rt3"].link_enable("eth-rt1", enabled=True)

    router_compare_json_output(
        "rt1", "show ip route ospf json", "step3/show_ip_route_healthy.ref"
    )
    router_compare_json_output(
        "rt1", "show ipv6 route ospf json", "step3/show_ipv6_route_healthy.ref"
    )
    router_compare_json_output(
        "rt1", "show bfd peers json", "step3/show_bfd_peers_healthy.ref"
    )


def test_memory_leak():
    "Run the memory leak test and report results."
    tgen = get_topogen()
    if not tgen.is_memleak_enabled():
        pytest.skip("Memory leak test/report is disabled")

    tgen.report_memory_leaks()


if __name__ == "__main__":
    args = ["-s"] + sys.argv[1:]
    sys.exit(pytest.main(args))
Commit	Line	Data
da78f4fe G	1	#!/usr/bin/env python
	2
	3	#
	4	# test_bfd_ospf_topo1.py
	5	# Part of NetDEF Topology Tests
	6	#
	7	# Copyright (c) 2020 by
	8	# Network Device Education Foundation, Inc. ("NetDEF")
	9	#
	10	# Permission to use, copy, modify, and/or distribute this software
	11	# for any purpose with or without fee is hereby granted, provided
	12	# that the above copyright notice and this permission notice appear
	13	# in all copies.
	14	#
	15	# THE SOFTWARE IS PROVIDED "AS IS" AND NETDEF DISCLAIMS ALL WARRANTIES
	16	# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
	17	# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NETDEF BE LIABLE FOR
	18	# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
	19	# DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
	20	# WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
	21	# ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
	22	# OF THIS SOFTWARE.
	23	#
	24
	25	"""
	26	test_bfd_ospf_topo1.py:
	27
	28	+---------+
	29	\| \|
	30	eth-rt2 (.1) \| RT1 \| eth-rt3 (.1)
	31	+----------+ 1.1.1.1 +----------+
	32	\| \| \| \|
	33	\| +---------+ \|
	34	\| \|
	35	\| 10.0.2.0/24 \|
	36	\| \|
	37	\| eth-rt1 \| (.2)
	38	\| 10.0.1.0/24 +----+----+
	39	\| \| \|
	40	\| \| RT3 \|
	41	\| \| 3.3.3.3 \|
	42	\| \| \|
	43	(.2) \| eth-rt1 +----+----+
	44	+----+----+ eth-rt4 \| (.1)
	45	\| \| \|
	46	\| RT2 \| \|
	47	\| 2.2.2.2 \| 10.0.4.0/24 \|
	48	\| \| \|
	49	+----+----+ \|
	50	(.1) \| eth-rt5 eth-rt3 \| (.2)
	51	\| +----+----+
	52	\| \| \|
	53	\| \| RT4 \|
	54	\| \| 4.4.4.4 \|
	55	\| \| \|
	56	\| +----+----+
	57	\| 10.0.3.0/24 eth-rt5 \| (.1)
	58	\| \|
	59	\| \|
	60	\| 10.0.5.0/24 \|
	61	\| \|
	62	\| +---------+ \|
	63	\| \| \| \|
	64	+----------+ RT5 +----------+
65	eth-rt2 (.2) \| 5.5.5.5 \| eth-rt4 (.2)
66	\| \|
67	+---------+
68
69	"""
70
71	import os
72	import sys
73	import pytest
74	import json
da78f4fe	75	from time import sleep
da78f4fe G	76	from functools import partial
	77
	78	# Save the Current Working Directory to find configuration files.
	79	CWD = os.path.dirname(os.path.realpath(__file__))
	80	sys.path.append(os.path.join(CWD, "../"))
	81
	82	# pylint: disable=C0413
	83	# Import topogen and topotest helpers
	84	from lib import topotest
	85	from lib.topogen import Topogen, TopoRouter, get_topogen
	86	from lib.topolog import logger
	87
3dedee4f	88	pytestmark = [pytest.mark.bfdd, pytest.mark.ospfd]
da78f4fe	89
5980ad0a	90
da78f4fe G	91	def setup_module(mod):
da78f4fe G	92	"Sets up the pytest environment"
8db751b8 CH	93	topodef = {
	94	"s1": ("rt1:eth-rt2", "rt2:eth-rt1"),
	95	"s2": ("rt1:eth-rt3", "rt3:eth-rt1"),
	96	"s3": ("rt2:eth-rt5", "rt5:eth-rt2"),
	97	"s4": ("rt3:eth-rt4", "rt4:eth-rt3"),
	98	"s5": ("rt4:eth-rt5", "rt5:eth-rt4"),
	99	}
	100	tgen = Topogen(topodef, mod.__name__)
da78f4fe G	101	tgen.start_topology()
	102
	103	router_list = tgen.routers()
	104
	105	# For all registered routers, load the zebra configuration file
8db751b8	106	for rname, router in router_list.items():
da78f4fe G	107	router.load_config(
	108	TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
	109	)
	110	router.load_config(
	111	TopoRouter.RD_BFD, os.path.join(CWD, "{}/bfdd.conf".format(rname))
	112	)
	113	router.load_config(
	114	TopoRouter.RD_OSPF, os.path.join(CWD, "{}/ospfd.conf".format(rname))
	115	)
	116	router.load_config(
	117	TopoRouter.RD_OSPF6, os.path.join(CWD, "{}/ospf6d.conf".format(rname))
	118	)
	119
	120	tgen.start_router()
	121
	122
	123	def teardown_module(mod):
	124	"Teardown the pytest environment"
	125	tgen = get_topogen()
	126
	127	# This function tears down the whole topology.
	128	tgen.stop_topology()
	129
	130
	131	def print_cmd_result(rname, command):
	132	print(get_topogen().gears[rname].vtysh_cmd(command, isjson=False))
	133
	134
8ff24fd2	135	def router_compare_json_output(rname, command, reference, count=40, wait=2):
da78f4fe G	136	"Compare router JSON output"
	137
	138	logger.info('Comparing router "%s" "%s" output', rname, command)
	139
	140	tgen = get_topogen()
	141	filename = "{}/{}/{}".format(CWD, rname, reference)
	142	expected = json.loads(open(filename).read())
	143
8ff24fd2	144	# Run test function until we get an result. Wait at most 80 seconds.
da78f4fe G	145	test_func = partial(topotest.router_json_cmp, tgen.gears[rname], command, expected)
	146	_, diff = topotest.run_and_expect(test_func, None, count=count, wait=wait)
	147	assertmsg = '"{}" JSON output mismatches the expected result'.format(rname)
	148	assert diff is None, assertmsg
	149
	150
	151	## TEST STEPS
	152
	153
	154	def test_rib_ospf_step1():
	155	logger.info("Test (step 1): verify RIB for OSPF")
	156	tgen = get_topogen()
	157
	158	# Skip if previous fatal error condition is raised
	159	if tgen.routers_have_failure():
	160	pytest.skip(tgen.errors)
	161
	162	router_compare_json_output(
	163	"rt1", "show ip route ospf json", "step1/show_ip_route.ref"
	164	)
	165	router_compare_json_output(
	166	"rt1", "show ipv6 route ospf json", "step1/show_ipv6_route.ref"
	167	)
	168
	169
	170	def test_bfd_ospf_sessions_step2():
	171	logger.info("Test (step 2): verify BFD peers for OSPF")
	172	tgen = get_topogen()
	173
	174	# Skip if previous fatal error condition is raised
	175	if tgen.routers_have_failure():
	176	pytest.skip(tgen.errors)
	177
	178	# BFD is just used on three routers
	179	for rt in ["rt1", "rt2", "rt3"]:
	180	router_compare_json_output(
	181	rt, "show bfd peers json", "step2/show_bfd_peers.ref"
	182	)
	183
	184
	185	def test_bfd_ospf_interface_failure_rt2_step3():
	186	logger.info("Test (step 3): Check failover handling with RT2 down")
	187	tgen = get_topogen()
	188
	189	# Skip if previous fatal error condition is raised
	190	if tgen.routers_have_failure():
	191	pytest.skip(tgen.errors)
	192
	193	# Let's kill the interface on rt2 and see what happens with the RIB and BFD on rt1
	194	tgen.gears["rt2"].link_enable("eth-rt1", enabled=False)
	195
	196	# By default BFD provides a recovery time of 900ms plus jitter, so let's wait
	197	# initial 2 seconds to let the CI not suffer.
2d28cbe6 RZ	198	topotest.sleep(2, 'Wait for BFD down notification')
2d28cbe6 RZ	199
da78f4fe	200	router_compare_json_output(
f786c3ee	201	"rt1", "show ip route ospf json", "step3/show_ip_route_rt2_down.ref", 10, 2
da78f4fe G	202	)
	203	router_compare_json_output(
	204	"rt1", "show ipv6 route ospf json", "step3/show_ipv6_route_rt2_down.ref", 1, 0
	205	)
	206	router_compare_json_output(
	207	"rt1", "show bfd peers json", "step3/show_bfd_peers_rt2_down.ref", 1, 0
	208	)
	209
	210	# Check recovery, this can take some time
	211	tgen.gears["rt2"].link_enable("eth-rt1", enabled=True)
	212
	213	router_compare_json_output(
	214	"rt1", "show ip route ospf json", "step3/show_ip_route_healthy.ref"
	215	)
	216	router_compare_json_output(
	217	"rt1", "show ipv6 route ospf json", "step3/show_ipv6_route_healthy.ref"
	218	)
	219	router_compare_json_output(
	220	"rt1", "show bfd peers json", "step3/show_bfd_peers_healthy.ref"
	221	)
	222
	223
	224	def test_bfd_ospf_interface_failure_rt3_step3():
	225	logger.info("Test (step 3): Check failover handling with RT3 down")
	226	tgen = get_topogen()
	227
	228	# Skip if previous fatal error condition is raised
	229	if tgen.routers_have_failure():
	230	pytest.skip(tgen.errors)
	231
	232	# Let's kill the interface on rt3 and see what happens with the RIB and BFD on rt1
	233	tgen.gears["rt3"].link_enable("eth-rt1", enabled=False)
	234
	235	# By default BFD provides a recovery time of 900ms plus jitter, so let's wait
	236	# initial 2 seconds to let the CI not suffer.
2d28cbe6	237	topotest.sleep(2, 'Wait for BFD down notification')
da78f4fe	238	router_compare_json_output(
f786c3ee	239	"rt1", "show ip route ospf json", "step3/show_ip_route_rt3_down.ref", 10, 2
da78f4fe G	240	)
	241	router_compare_json_output(
	242	"rt1", "show ipv6 route ospf json", "step3/show_ipv6_route_rt3_down.ref", 1, 0
	243	)
	244	router_compare_json_output(
	245	"rt1", "show bfd peers json", "step3/show_bfd_peers_rt3_down.ref", 1, 0
	246	)
	247
	248	# Check recovery, this can take some time
	249	tgen.gears["rt3"].link_enable("eth-rt1", enabled=True)
	250
	251	router_compare_json_output(
	252	"rt1", "show ip route ospf json", "step3/show_ip_route_healthy.ref"
	253	)
	254	router_compare_json_output(
	255	"rt1", "show ipv6 route ospf json", "step3/show_ipv6_route_healthy.ref"
	256	)
	257	router_compare_json_output(
	258	"rt1", "show bfd peers json", "step3/show_bfd_peers_healthy.ref"
	259	)
	260
	261
	262	def test_memory_leak():
	263	"Run the memory leak test and report results."
	264	tgen = get_topogen()
	265	if not tgen.is_memleak_enabled():
	266	pytest.skip("Memory leak test/report is disabled")
	267
	268	tgen.report_memory_leaks()
	269
	270
	271	if __name__ == "__main__":
	272	args = ["-s"] + sys.argv[1:]
	273	sys.exit(pytest.main(args))