[mirror_frr.git] / tests / topotests / ospf6_topo1_vrf / test_ospf6_topo1_vrf.py

#!/usr/bin/env python
# SPDX-License-Identifier: ISC

#
# test_ospf6_topo1_vrf.py
# Part of NetDEF Topology Tests
#
# Copyright (c) 2021 by Niral Networks, Inc. ("Niral Networks")
# Used Copyright (c) 2016 by Network Device Education Foundation,
# Inc. ("NetDEF") in this file.
#

r"""
test_ospf6_topo1_vrf.py:

                                                  -----\
  SW1 - Stub Net 1            SW2 - Stub Net 2          \
  fc00:1:1:1::/64             fc00:2:2:2::/64            \
\___________________/      \___________________/          |
          |                          |                    |
          |                          |                    |
          | ::1                      | ::2                |
+---------+---------+      +---------+---------+          |
|        R1         |      |        R2         |          |
|     FRRouting     |      |     FRRouting     |          |
| Rtr-ID: 10.0.0.1  |      | Rtr-ID: 10.0.0.2  |          |
+---------+---------+      +---------+---------+          |
          | ::1                      | ::2                 \
           \______        ___________/                      OSPFv3
                  \      /                               Area 0.0.0.0
                   \    /                                  /
             ~~~~~~~~~~~~~~~~~~                           |
           ~~       SW5        ~~                         |
         ~~       Switch         ~~                       |
           ~~  fc00:A:A:A::/64 ~~                         |
             ~~~~~~~~~~~~~~~~~~                           |
                     |                 /----              |
                     | ::3            | SW3 - Stub Net 3  | 
           +---------+---------+    /-+ fc00:3:3:3::/64   |
           |        R3         |   /  |                  /
           |     FRRouting     +--/    \----            /
           | Rtr-ID: 10.0.0.3  | ::3        ___________/
           +---------+---------+                       \
                     | ::3                              \
                     |                                   \
             ~~~~~~~~~~~~~~~~~~                           |
           ~~       SW6        ~~                         |
         ~~       Switch         ~~                       |
           ~~  fc00:B:B:B::/64 ~~                          \
             ~~~~~~~~~~~~~~~~~~                             OSPFv3
                     |                                   Area 0.0.0.1
                     | ::4                                 /
           +---------+---------+       /----              |
           |        R4         |      | SW4 - Stub Net 4  |
           |     FRRouting     +------+ fc00:4:4:4::/64   |
           | Rtr-ID: 10.0.0.4  | ::4  |                   /
           +-------------------+       \----             /
                                                   -----/
"""

import os
import re
import sys
import pytest

from functools import partial


# Save the Current Working Directory to find configuration files later.
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
from lib.topotest import iproute2_is_vrf_capable
from lib.common_config import required_linux_kernel_version


pytestmark = [pytest.mark.ospfd]


def build_topo(tgen):
    "Build function"

    # Create 4 routers
    for routern in range(1, 5):
        tgen.add_router("r{}".format(routern))

    #
    # Wire up the switches and routers
    # Note that we specify the link names so we match the config files
    #

    # Create a empty network for router 1
    switch = tgen.add_switch("s1")
    switch.add_link(tgen.gears["r1"], nodeif="r1-stubnet")

    # Create a empty network for router 2
    switch = tgen.add_switch("s2")
    switch.add_link(tgen.gears["r2"], nodeif="r2-stubnet")

    # Create a empty network for router 3
    switch = tgen.add_switch("s3")
    switch.add_link(tgen.gears["r3"], nodeif="r3-stubnet")

    # Create a empty network for router 4
    switch = tgen.add_switch("s4")
    switch.add_link(tgen.gears["r4"], nodeif="r4-stubnet")

    # Interconnect routers 1, 2, and 3
    switch = tgen.add_switch("s5")
    switch.add_link(tgen.gears["r1"], nodeif="r1-sw5")
    switch.add_link(tgen.gears["r2"], nodeif="r2-sw5")
    switch.add_link(tgen.gears["r3"], nodeif="r3-sw5")

    # Interconnect routers 3 and 4
    switch = tgen.add_switch("s6")
    switch.add_link(tgen.gears["r3"], nodeif="r3-sw6")
    switch.add_link(tgen.gears["r4"], nodeif="r4-sw6")


#####################################################
##
##   Tests starting
##
#####################################################


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

    # Required linux kernel version for this suite to run.
    result = required_linux_kernel_version("5.0")
    if result is not True:
        pytest.skip("Kernel requirements are not met")

    tgen = Topogen(build_topo, mod.__name__)
    tgen.start_topology()

    logger.info("** %s: Setup Topology" % mod.__name__)
    logger.info("******************************************")

    # For debugging after starting net, but before starting FRR,
    # uncomment the next line
    # tgen.mininet_cli()

    logger.info("Testing with VRF Lite support")

    cmds = [
        "ip link add {0}-cust1 type vrf table 1001",
        "ip link add loop1 type dummy",
        "ip link set {0}-stubnet master {0}-cust1",
    ]

    cmds1 = ["ip link set {0}-sw5 master {0}-cust1"]

    cmds2 = ["ip link set {0}-sw6 master {0}-cust1"]

    # For all registered routers, load the zebra configuration file
    for rname, router in tgen.routers().items():
        # create VRF rx-cust1 and link rx-eth0 to rx-cust1
        for cmd in cmds:
            output = tgen.net[rname].cmd(cmd.format(rname))
        if rname == "r1" or rname == "r2" or rname == "r3":
            for cmd in cmds1:
                output = tgen.net[rname].cmd(cmd.format(rname))
        if rname == "r3" or rname == "r4":
            for cmd in cmds2:
                output = tgen.net[rname].cmd(cmd.format(rname))

    for rname, router in tgen.routers().items():
        router.load_config(
            TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
        )
        router.load_config(
            TopoRouter.RD_OSPF6, os.path.join(CWD, "{}/ospf6d.conf".format(rname))
        )

    # Initialize all routers.
    tgen.start_router()

    # For debugging after starting FRR daemons, uncomment the next line
    # tgen.mininet_cli()


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


def test_wait_protocol_convergence():
    "Wait for OSPFv3 to converge"
    tgen = get_topogen()
    if tgen.routers_have_failure():
        pytest.skip(tgen.errors)

    logger.info("waiting for protocols to converge")

    def expect_neighbor_full(router, neighbor):
        "Wait until OSPFv3 convergence."
        logger.info("waiting OSPFv3 router '{}'".format(router))
        test_func = partial(
            topotest.router_json_cmp,
            tgen.gears[router],
            "show ipv6 ospf6 vrf {0}-cust1 neighbor json".format(router),
            {"neighbors": [{"neighborId": neighbor, "state": "Full"}]},
        )
        _, result = topotest.run_and_expect(test_func, None, count=130, wait=1)
        assertmsg = '"{}" convergence failure'.format(router)
        assert result is None, assertmsg

    expect_neighbor_full("r1", "10.0.0.2")
    expect_neighbor_full("r1", "10.0.0.3")

    expect_neighbor_full("r2", "10.0.0.1")
    expect_neighbor_full("r2", "10.0.0.3")

    expect_neighbor_full("r3", "10.0.0.1")
    expect_neighbor_full("r3", "10.0.0.2")
    expect_neighbor_full("r3", "10.0.0.4")


def compare_show_ipv6_vrf(rname, expected):
    """
    Calls 'show ipv6 route' for router `rname` and compare the obtained
    result with the expected output.
    """
    tgen = get_topogen()

    # Use the vtysh output, with some masking to make comparison easy
    vrf_name = "{0}-cust1".format(rname)
    current = topotest.ip6_route_zebra(tgen.gears[rname], vrf_name)

    # Use just the 'O'spf lines of the output
    linearr = []
    for line in current.splitlines():
        if re.match("^O", line):
            linearr.append(line)

    current = "\n".join(linearr)

    return topotest.difflines(
        topotest.normalize_text(current),
        topotest.normalize_text(expected),
        title1="Current output",
        title2="Expected output",
    )


def test_ospfv3_routingTable():

    tgen = get_topogen()
    if tgen.routers_have_failure():
        pytest.skip("skipped because of router(s) failure")

    # For debugging, uncomment the next line
    # tgen.mininet_cli()
    # Verify OSPFv3 Routing Table
    for router, rnode in tgen.routers().items():
        logger.info('Waiting for router "%s" convergence', router)

        # Load expected results from the command
        reffile = os.path.join(CWD, "{}/show_ipv6_vrf_route.ref".format(router))
        expected = open(reffile).read()

        # Run test function until we get an result. Wait at most 60 seconds.
        test_func = partial(compare_show_ipv6_vrf, router, expected)
        result, diff = topotest.run_and_expect(test_func, "", count=120, wait=0.5)
        assert result, "OSPFv3 did not converge on {}:\n{}".format(router, diff)


def test_linux_ipv6_kernel_routingTable():

    # Required linux kernel version for this suite to run.
    result = required_linux_kernel_version("4.15")
    if result is not True:
        pytest.skip("Kernel requirements are not met")

    # iproute2 needs to support VRFs for this suite to run.
    if not iproute2_is_vrf_capable():
        pytest.skip("Installed iproute2 version does not support VRFs")

    tgen = get_topogen()

    if tgen.routers_have_failure():
        pytest.skip("skipped because of router(s) failure")

    # Verify Linux Kernel Routing Table
    logger.info("Verifying Linux IPv6 Kernel Routing Table")

    failures = 0

    # Get a list of all current link-local addresses first as they change for
    # each run and we need to translate them
    linklocals = []
    for i in range(1, 5):
        linklocals += tgen.net["r{}".format(i)].get_ipv6_linklocal()

    # Now compare the routing tables (after substituting link-local addresses)

    for i in range(1, 5):
        # Actual output from router
        actual = (
            tgen.gears["r{}".format(i)]
            .run("ip -6 route show vrf r{}-cust1".format(i))
            .rstrip()
        )
        if "nhid" in actual:
            refTableFile = os.path.join(CWD, "r{}/ip_6_address.nhg.ref".format(i))
        else:
            refTableFile = os.path.join(CWD, "r{}/ip_6_address.ref".format(i))

        if os.path.isfile(refTableFile):
            expected = open(refTableFile).read().rstrip()
            # Fix newlines (make them all the same)
            expected = ("\n".join(expected.splitlines())).splitlines(1)

            # Mask out Link-Local mac addresses
            for ll in linklocals:
                actual = actual.replace(ll[1], "fe80::__(%s)__" % ll[0])
            # Mask out protocol name or number
            actual = re.sub(r"[ ]+proto [0-9a-z]+ +", "  proto XXXX ", actual)
            actual = re.sub(r"[ ]+nhid [0-9]+ +", " nhid XXXX ", actual)
            # Remove ff00::/8 routes (seen on some kernels - not from FRR)
            actual = re.sub(r"ff00::/8.*", "", actual)

            # Strip empty lines
            actual = actual.lstrip()
            actual = actual.rstrip()
            actual = re.sub(r"  +", " ", actual)

            filtered_lines = []
            for line in sorted(actual.splitlines()):
                if line.startswith("fe80::/64 ") or line.startswith(
                    "unreachable fe80::/64 "
                ):
                    continue
                if "anycast" in line:
                    continue
                if "multicast" in line:
                    continue
                filtered_lines.append(line)
            actual = "\n".join(filtered_lines).splitlines(1)

            # Print Actual table
            # logger.info("Router r%s table" % i)
            # for line in actual:
            #     logger.info(line.rstrip())

            # Generate Diff
            diff = topotest.get_textdiff(
                actual,
                expected,
                title1="actual OSPFv3 IPv6 routing table",
                title2="expected OSPFv3 IPv6 routing table",
            )

            # Empty string if it matches, otherwise diff contains unified diff
            if diff:
                sys.stderr.write(
                    "r%s failed Linux IPv6 Kernel Routing Table Check:\n%s\n"
                    % (i, diff)
                )
                failures += 1
            else:
                logger.info("r%s ok" % i)

            assert failures == 0, (
                "Linux Kernel IPv6 Routing Table verification failed for router r%s:\n%s"
                % (i, diff)
            )


def test_ospfv3_routingTable_write_multiplier():

    tgen = get_topogen()
    if tgen.routers_have_failure():
        pytest.skip("skipped because of router(s) failure")

    # For debugging, uncomment the next line
    # tgen.mininet_cli()
    # Modify R1 write muliplier and reset the interfaces
    r1 = tgen.gears["r1"]

    r1.vtysh_cmd("conf t\nrouter ospf6 vrf r1-cust1 \n write-multiplier 100")
    r1.vtysh_cmd("clear ipv6 ospf interface r1-stubnet")
    r1.vtysh_cmd("clear ipv6 ospf interface r1-sw5")

    # Verify OSPFv3 Routing Table
    for router, rnode in tgen.routers().items():
        logger.info('Waiting for router "%s" convergence', router)

        # Load expected results from the command
        reffile = os.path.join(CWD, "{}/show_ipv6_vrf_route.ref".format(router))
        expected = open(reffile).read()

        # Run test function until we get an result. Wait at most 60 seconds.
        test_func = partial(compare_show_ipv6_vrf, router, expected)
        result, diff = topotest.run_and_expect(test_func, "", count=120, wait=0.5)
        assert result, "OSPFv3 did not converge on {}:\n{}".format(router, diff)


def test_shutdown_check_stderr():

    tgen = get_topogen()

    if tgen.routers_have_failure():
        pytest.skip("skipped because of router(s) failure")

    if os.environ.get("TOPOTESTS_CHECK_STDERR") is None:
        logger.info(
            "SKIPPED final check on StdErr output: Disabled (TOPOTESTS_CHECK_STDERR undefined)\n"
        )
        pytest.skip("Skipping test for Stderr output")

    net = tgen.net

    logger.info("\n\n** Verifying unexpected STDERR output from daemons")
    logger.info("******************************************")

    for i in range(1, 5):
        net["r%s" % i].stopRouter()
        log = net["r%s" % i].getStdErr("ospf6d")
        if log:
            logger.info("\nRouter r%s OSPF6d StdErr Log:\n%s" % (i, log))
        log = net["r%s" % i].getStdErr("zebra")
        if log:
            logger.info("\nRouter r%s Zebra StdErr Log:\n%s" % (i, log))


def test_shutdown_check_memleak():
    "Run the memory leak test and report results."

    if os.environ.get("TOPOTESTS_CHECK_MEMLEAK") is None:
        logger.info(
            "SKIPPED final check on Memory leaks: Disabled (TOPOTESTS_CHECK_MEMLEAK undefined)"
        )
        pytest.skip("Skipping test for memory leaks")

    tgen = get_topogen()

    net = tgen.net

    for i in range(1, 5):
        net["r%s" % i].stopRouter()
        net["r%s" % i].report_memory_leaks(
            os.environ.get("TOPOTESTS_CHECK_MEMLEAK"), os.path.basename(__file__)
        )


if __name__ == "__main__":

    # To suppress tracebacks, either use the following pytest call or
    # add "--tb=no" to cli
    # retval = pytest.main(["-s", "--tb=no"])

    retval = pytest.main(["-s"])
    sys.exit(retval)
Commit	Line	Data
0681195e	1	#!/usr/bin/env python
acddc0ed	2	# SPDX-License-Identifier: ISC
0681195e	3
	4	#
	5	# test_ospf6_topo1_vrf.py
	6	# Part of NetDEF Topology Tests
	7	#
	8	# Copyright (c) 2021 by Niral Networks, Inc. ("Niral Networks")
d01673b4	9	# Used Copyright (c) 2016 by Network Device Education Foundation,
0681195e	10	# Inc. ("NetDEF") in this file.
0681195e	11	#
0681195e	12
21b5cd1d	13	r"""
0681195e	14	test_ospf6_topo1_vrf.py:
	15
	16	-----\
	17	SW1 - Stub Net 1 SW2 - Stub Net 2 \
	18	fc00:1:1:1::/64 fc00:2:2:2::/64 \
	19	\___________________/ \___________________/ \|
	20	\| \| \|
	21	\| \| \|
	22	\| ::1 \| ::2 \|
	23	+---------+---------+ +---------+---------+ \|
	24	\| R1 \| \| R2 \| \|
	25	\| FRRouting \| \| FRRouting \| \|
	26	\| Rtr-ID: 10.0.0.1 \| \| Rtr-ID: 10.0.0.2 \| \|
	27	+---------+---------+ +---------+---------+ \|
	28	\| ::1 \| ::2 \
	29	\______ ___________/ OSPFv3
	30	\ / Area 0.0.0.0
	31	\ / /
	32	~~~~~~~~~~~~~~~~~~ \|
	33	~~ SW5 ~~ \|
	34	~~ Switch ~~ \|
	35	~~ fc00:A:A:A::/64 ~~ \|
	36	~~~~~~~~~~~~~~~~~~ \|
	37	\| /---- \|
	38	\| ::3 \| SW3 - Stub Net 3 \|
	39	+---------+---------+ /-+ fc00:3:3:3::/64 \|
	40	\| R3 \| / \| /
	41	\| FRRouting +--/ \---- /
	42	\| Rtr-ID: 10.0.0.3 \| ::3 ___________/
	43	+---------+---------+ \
	44	\| ::3 \
	45	\| \
	46	~~~~~~~~~~~~~~~~~~ \|
	47	~~ SW6 ~~ \|
	48	~~ Switch ~~ \|
	49	~~ fc00:B:B:B::/64 ~~ \
	50	~~~~~~~~~~~~~~~~~~ OSPFv3
	51	\| Area 0.0.0.1
	52	\| ::4 /
	53	+---------+---------+ /---- \|
	54	\| R4 \| \| SW4 - Stub Net 4 \|
	55	\| FRRouting +------+ fc00:4:4:4::/64 \|
	56	\| Rtr-ID: 10.0.0.4 \| ::4 \| /
	57	+-------------------+ \---- /
	58	-----/
	59	"""
	60
	61	import os
	62	import re
	63	import sys
	64	import pytest
0681195e	65
	66	from functools import partial
	67
0681195e	68
	69	# Save the Current Working Directory to find configuration files later.
	70	CWD = os.path.dirname(os.path.realpath(__file__))
	71	sys.path.append(os.path.join(CWD, "../"))
	72
	73	# pylint: disable=C0413
	74	# Import topogen and topotest helpers
	75	from lib import topotest
	76	from lib.topogen import Topogen, TopoRouter, get_topogen
	77	from lib.topolog import logger
	78	from lib.topotest import iproute2_is_vrf_capable
36eef858	79	from lib.common_config import required_linux_kernel_version
0681195e	80
0681195e	81
e82b531d	82	pytestmark = [pytest.mark.ospfd]
0681195e	83
0681195e	84
e82b531d CH	85	def build_topo(tgen):
e82b531d CH	86	"Build function"
0681195e	87
e82b531d CH	88	# Create 4 routers
	89	for routern in range(1, 5):
	90	tgen.add_router("r{}".format(routern))
0681195e	91
e82b531d CH	92	#
	93	# Wire up the switches and routers
	94	# Note that we specify the link names so we match the config files
	95	#
0681195e	96
e82b531d CH	97	# Create a empty network for router 1
	98	switch = tgen.add_switch("s1")
	99	switch.add_link(tgen.gears["r1"], nodeif="r1-stubnet")
0681195e	100
e82b531d CH	101	# Create a empty network for router 2
	102	switch = tgen.add_switch("s2")
	103	switch.add_link(tgen.gears["r2"], nodeif="r2-stubnet")
0681195e	104
e82b531d CH	105	# Create a empty network for router 3
	106	switch = tgen.add_switch("s3")
	107	switch.add_link(tgen.gears["r3"], nodeif="r3-stubnet")
0681195e	108
e82b531d CH	109	# Create a empty network for router 4
	110	switch = tgen.add_switch("s4")
	111	switch.add_link(tgen.gears["r4"], nodeif="r4-stubnet")
0681195e	112
e82b531d CH	113	# Interconnect routers 1, 2, and 3
	114	switch = tgen.add_switch("s5")
	115	switch.add_link(tgen.gears["r1"], nodeif="r1-sw5")
	116	switch.add_link(tgen.gears["r2"], nodeif="r2-sw5")
	117	switch.add_link(tgen.gears["r3"], nodeif="r3-sw5")
0681195e	118
e82b531d CH	119	# Interconnect routers 3 and 4
	120	switch = tgen.add_switch("s6")
	121	switch.add_link(tgen.gears["r3"], nodeif="r3-sw6")
	122	switch.add_link(tgen.gears["r4"], nodeif="r4-sw6")
0681195e	123
	124
	125	#####################################################
	126	##
	127	## Tests starting
	128	##
	129	#####################################################
	130
	131
	132	def setup_module(mod):
	133	"Sets up the pytest environment"
	134
36eef858 IR	135	# Required linux kernel version for this suite to run.
	136	result = required_linux_kernel_version("5.0")
	137	if result is not True:
	138	pytest.skip("Kernel requirements are not met")
	139
e82b531d	140	tgen = Topogen(build_topo, mod.__name__)
0681195e	141	tgen.start_topology()
	142
	143	logger.info("** %s: Setup Topology" % mod.__name__)
	144	logger.info("******************************************")
	145
	146	# For debugging after starting net, but before starting FRR,
	147	# uncomment the next line
	148	# tgen.mininet_cli()
	149
	150	logger.info("Testing with VRF Lite support")
	151
	152	cmds = [
	153	"ip link add {0}-cust1 type vrf table 1001",
	154	"ip link add loop1 type dummy",
	155	"ip link set {0}-stubnet master {0}-cust1",
	156	]
	157
d01673b4	158	cmds1 = ["ip link set {0}-sw5 master {0}-cust1"]
0681195e	159
d01673b4	160	cmds2 = ["ip link set {0}-sw6 master {0}-cust1"]
0681195e	161
	162	# For all registered routers, load the zebra configuration file
	163	for rname, router in tgen.routers().items():
	164	# create VRF rx-cust1 and link rx-eth0 to rx-cust1
	165	for cmd in cmds:
	166	output = tgen.net[rname].cmd(cmd.format(rname))
	167	if rname == "r1" or rname == "r2" or rname == "r3":
	168	for cmd in cmds1:
	169	output = tgen.net[rname].cmd(cmd.format(rname))
	170	if rname == "r3" or rname == "r4":
	171	for cmd in cmds2:
	172	output = tgen.net[rname].cmd(cmd.format(rname))
	173
0681195e	174	for rname, router in tgen.routers().items():
	175	router.load_config(
	176	TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
	177	)
	178	router.load_config(
	179	TopoRouter.RD_OSPF6, os.path.join(CWD, "{}/ospf6d.conf".format(rname))
	180	)
	181
	182	# Initialize all routers.
	183	tgen.start_router()
	184
	185	# For debugging after starting FRR daemons, uncomment the next line
	186	# tgen.mininet_cli()
	187
	188
	189	def teardown_module(mod):
	190	"Teardown the pytest environment"
	191	tgen = get_topogen()
	192	tgen.stop_topology()
	193
d01673b4	194
0681195e	195	def test_wait_protocol_convergence():
	196	"Wait for OSPFv3 to converge"
	197	tgen = get_topogen()
	198	if tgen.routers_have_failure():
	199	pytest.skip(tgen.errors)
	200
	201	logger.info("waiting for protocols to converge")
	202
	203	def expect_neighbor_full(router, neighbor):
	204	"Wait until OSPFv3 convergence."
	205	logger.info("waiting OSPFv3 router '{}'".format(router))
	206	test_func = partial(
	207	topotest.router_json_cmp,
	208	tgen.gears[router],
	209	"show ipv6 ospf6 vrf {0}-cust1 neighbor json".format(router),
	210	{"neighbors": [{"neighborId": neighbor, "state": "Full"}]},
	211	)
	212	_, result = topotest.run_and_expect(test_func, None, count=130, wait=1)
	213	assertmsg = '"{}" convergence failure'.format(router)
	214	assert result is None, assertmsg
	215
	216	expect_neighbor_full("r1", "10.0.0.2")
	217	expect_neighbor_full("r1", "10.0.0.3")
	218
	219	expect_neighbor_full("r2", "10.0.0.1")
	220	expect_neighbor_full("r2", "10.0.0.3")
	221
	222	expect_neighbor_full("r3", "10.0.0.1")
	223	expect_neighbor_full("r3", "10.0.0.2")
	224	expect_neighbor_full("r3", "10.0.0.4")
	225
	226
	227	def compare_show_ipv6_vrf(rname, expected):
	228	"""
	229	Calls 'show ipv6 route' for router `rname` and compare the obtained
	230	result with the expected output.
	231	"""
	232	tgen = get_topogen()
	233
	234	# Use the vtysh output, with some masking to make comparison easy
	235	vrf_name = "{0}-cust1".format(rname)
	236	current = topotest.ip6_route_zebra(tgen.gears[rname], vrf_name)
d01673b4	237
0681195e	238	# Use just the 'O'spf lines of the output
	239	linearr = []
	240	for line in current.splitlines():
	241	if re.match("^O", line):
	242	linearr.append(line)
	243
	244	current = "\n".join(linearr)
	245
	246	return topotest.difflines(
	247	topotest.normalize_text(current),
	248	topotest.normalize_text(expected),
	249	title1="Current output",
	250	title2="Expected output",
	251	)
	252
	253
	254	def test_ospfv3_routingTable():
	255
	256	tgen = get_topogen()
	257	if tgen.routers_have_failure():
	258	pytest.skip("skipped because of router(s) failure")
	259
	260	# For debugging, uncomment the next line
	261	# tgen.mininet_cli()
	262	# Verify OSPFv3 Routing Table
d7d21c3a	263	for router, rnode in tgen.routers().items():
0681195e	264	logger.info('Waiting for router "%s" convergence', router)
	265
	266	# Load expected results from the command
	267	reffile = os.path.join(CWD, "{}/show_ipv6_vrf_route.ref".format(router))
	268	expected = open(reffile).read()
	269
	270	# Run test function until we get an result. Wait at most 60 seconds.
	271	test_func = partial(compare_show_ipv6_vrf, router, expected)
	272	result, diff = topotest.run_and_expect(test_func, "", count=120, wait=0.5)
	273	assert result, "OSPFv3 did not converge on {}:\n{}".format(router, diff)
	274
	275
	276	def test_linux_ipv6_kernel_routingTable():
	277
	278	# Required linux kernel version for this suite to run.
	279	result = required_linux_kernel_version("4.15")
	280	if result is not True:
	281	pytest.skip("Kernel requirements are not met")
	282
	283	# iproute2 needs to support VRFs for this suite to run.
	284	if not iproute2_is_vrf_capable():
	285	pytest.skip("Installed iproute2 version does not support VRFs")
	286
	287	tgen = get_topogen()
	288
	289	if tgen.routers_have_failure():
	290	pytest.skip("skipped because of router(s) failure")
	291
	292	# Verify Linux Kernel Routing Table
	293	logger.info("Verifying Linux IPv6 Kernel Routing Table")
	294
	295	failures = 0
	296
	297	# Get a list of all current link-local addresses first as they change for
	298	# each run and we need to translate them
	299	linklocals = []
	300	for i in range(1, 5):
	301	linklocals += tgen.net["r{}".format(i)].get_ipv6_linklocal()
	302
	303	# Now compare the routing tables (after substituting link-local addresses)
	304
	305	for i in range(1, 5):
	306	# Actual output from router
d01673b4 PR	307	actual = (
	308	tgen.gears["r{}".format(i)]
	309	.run("ip -6 route show vrf r{}-cust1".format(i))
	310	.rstrip()
	311	)
0681195e	312	if "nhid" in actual:
	313	refTableFile = os.path.join(CWD, "r{}/ip_6_address.nhg.ref".format(i))
	314	else:
	315	refTableFile = os.path.join(CWD, "r{}/ip_6_address.ref".format(i))
	316
	317	if os.path.isfile(refTableFile):
	318	expected = open(refTableFile).read().rstrip()
	319	# Fix newlines (make them all the same)
	320	expected = ("\n".join(expected.splitlines())).splitlines(1)
	321
	322	# Mask out Link-Local mac addresses
	323	for ll in linklocals:
	324	actual = actual.replace(ll[1], "fe80::__(%s)__" % ll[0])
	325	# Mask out protocol name or number
	326	actual = re.sub(r"[ ]+proto [0-9a-z]+ +", " proto XXXX ", actual)
	327	actual = re.sub(r"[ ]+nhid [0-9]+ +", " nhid XXXX ", actual)
	328	# Remove ff00::/8 routes (seen on some kernels - not from FRR)
	329	actual = re.sub(r"ff00::/8.*", "", actual)
	330
	331	# Strip empty lines
	332	actual = actual.lstrip()
	333	actual = actual.rstrip()
	334	actual = re.sub(r" +", " ", actual)
	335
	336	filtered_lines = []
	337	for line in sorted(actual.splitlines()):
	338	if line.startswith("fe80::/64 ") or line.startswith(
	339	"unreachable fe80::/64 "
	340	):
	341	continue
d01673b4	342	if "anycast" in line:
0681195e	343	continue
d01673b4	344	if "multicast" in line:
0681195e	345	continue
	346	filtered_lines.append(line)
	347	actual = "\n".join(filtered_lines).splitlines(1)
	348
	349	# Print Actual table
	350	# logger.info("Router r%s table" % i)
	351	# for line in actual:
	352	# logger.info(line.rstrip())
	353
	354	# Generate Diff
	355	diff = topotest.get_textdiff(
	356	actual,
	357	expected,
	358	title1="actual OSPFv3 IPv6 routing table",
	359	title2="expected OSPFv3 IPv6 routing table",
	360	)
	361
	362	# Empty string if it matches, otherwise diff contains unified diff
	363	if diff:
	364	sys.stderr.write(
	365	"r%s failed Linux IPv6 Kernel Routing Table Check:\n%s\n"
	366	% (i, diff)
	367	)
	368	failures += 1
	369	else:
	370	logger.info("r%s ok" % i)
	371
	372	assert failures == 0, (
	373	"Linux Kernel IPv6 Routing Table verification failed for router r%s:\n%s"
	374	% (i, diff)
	375	)
	376
	377
d01673b4 PR	378	def test_ospfv3_routingTable_write_multiplier():
	379
	380	tgen = get_topogen()
	381	if tgen.routers_have_failure():
	382	pytest.skip("skipped because of router(s) failure")
	383
	384	# For debugging, uncomment the next line
	385	# tgen.mininet_cli()
	386	# Modify R1 write muliplier and reset the interfaces
	387	r1 = tgen.gears["r1"]
	388
	389	r1.vtysh_cmd("conf t\nrouter ospf6 vrf r1-cust1 \n write-multiplier 100")
	390	r1.vtysh_cmd("clear ipv6 ospf interface r1-stubnet")
	391	r1.vtysh_cmd("clear ipv6 ospf interface r1-sw5")
	392
	393	# Verify OSPFv3 Routing Table
d7d21c3a	394	for router, rnode in tgen.routers().items():
d01673b4 PR	395	logger.info('Waiting for router "%s" convergence', router)
	396
	397	# Load expected results from the command
	398	reffile = os.path.join(CWD, "{}/show_ipv6_vrf_route.ref".format(router))
	399	expected = open(reffile).read()
	400
	401	# Run test function until we get an result. Wait at most 60 seconds.
	402	test_func = partial(compare_show_ipv6_vrf, router, expected)
	403	result, diff = topotest.run_and_expect(test_func, "", count=120, wait=0.5)
	404	assert result, "OSPFv3 did not converge on {}:\n{}".format(router, diff)
	405
	406
0681195e	407	def test_shutdown_check_stderr():
	408
	409	tgen = get_topogen()
	410
	411	if tgen.routers_have_failure():
	412	pytest.skip("skipped because of router(s) failure")
	413
	414	if os.environ.get("TOPOTESTS_CHECK_STDERR") is None:
	415	logger.info(
	416	"SKIPPED final check on StdErr output: Disabled (TOPOTESTS_CHECK_STDERR undefined)\n"
	417	)
	418	pytest.skip("Skipping test for Stderr output")
	419
	420	net = tgen.net
	421
	422	logger.info("\n\n** Verifying unexpected STDERR output from daemons")
	423	logger.info("******************************************")
	424
	425	for i in range(1, 5):
	426	net["r%s" % i].stopRouter()
	427	log = net["r%s" % i].getStdErr("ospf6d")
	428	if log:
	429	logger.info("\nRouter r%s OSPF6d StdErr Log:\n%s" % (i, log))
	430	log = net["r%s" % i].getStdErr("zebra")
	431	if log:
	432	logger.info("\nRouter r%s Zebra StdErr Log:\n%s" % (i, log))
	433
	434
	435	def test_shutdown_check_memleak():
	436	"Run the memory leak test and report results."
	437
	438	if os.environ.get("TOPOTESTS_CHECK_MEMLEAK") is None:
	439	logger.info(
	440	"SKIPPED final check on Memory leaks: Disabled (TOPOTESTS_CHECK_MEMLEAK undefined)"
	441	)
	442	pytest.skip("Skipping test for memory leaks")
	443
	444	tgen = get_topogen()
	445
	446	net = tgen.net
	447
	448	for i in range(1, 5):
	449	net["r%s" % i].stopRouter()
	450	net["r%s" % i].report_memory_leaks(
	451	os.environ.get("TOPOTESTS_CHECK_MEMLEAK"), os.path.basename(__file__)
	452	)
	453
	454
	455	if __name__ == "__main__":
	456
	457	# To suppress tracebacks, either use the following pytest call or
	458	# add "--tb=no" to cli
	459	# retval = pytest.main(["-s", "--tb=no"])
	460
	461	retval = pytest.main(["-s"])
	462	sys.exit(retval)