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

#!/usr/bin/env python

#
# 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.
#
# 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_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
import platform
from time import sleep

from functools import partial

from lib.micronet_compat import Topo

# 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]


#####################################################
##
##   Network Topology Definition
##
#####################################################


class NetworkTopo(Topo):
    "OSPFv3 (IPv6) Test Topology 1"

    def build(self, **_opts):
        "Build function"

        tgen = get_topogen(self)

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