Merge pull request #12798 from donaldsharp/rib_match_multicast

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

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

#
# test_bfd_ospf_topo1.py
# Part of NetDEF Topology Tests
#
# Copyright (c) 2020 by
# Network Device Education Foundation, Inc. ("NetDEF")
#

"""
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", 10, 2
    )
    router_compare_json_output(
        "rt1", "show bfd peers json", "step3/show_bfd_peers_rt2_down.ref", 10, 2
    )

    # 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", 10, 2
    )
    router_compare_json_output(
        "rt1", "show bfd peers json", "step3/show_bfd_peers_rt3_down.ref", 10, 2
    )

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