path: root/tests/topotests/ospf_topo2/test_ospf_topo2.py

#!/usr/bin/env python
# -*- coding: utf-8 eval: (blacken-mode 1) -*-
# SPDX-License-Identifier: ISC
#
# test_ospf_topo2.py
# Part of NetDEF Topology Tests
#
# Copyright (c) 2017 by
# Network Device Education Foundation, Inc. ("NetDEF")
#

"""
test_ospf_topo2.py: Test correct route removal.

Proofs the following issue:
https://github.com/FRRouting/frr/issues/14488

"""

import ipaddress
import json
import pytest
import sys
import time

from lib.topogen import Topogen


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


def build_topo(tgen):
    """Build the topology used by all tests below."""

    # Create 4 routers
    r1 = tgen.add_router("r1")
    r2 = tgen.add_router("r2")
    r3 = tgen.add_router("r3")
    r4 = tgen.add_router("r4")

    # The r1/r2 and r3/r4 router pairs have two connections each
    tgen.add_link(r1, r2, ifname1="eth1", ifname2="eth1")
    tgen.add_link(r1, r2, ifname1="eth2", ifname2="eth2")
    tgen.add_link(r3, r4, ifname1="eth2", ifname2="eth2")
    tgen.add_link(r3, r4, ifname1="eth3", ifname2="eth3")

    # The r1/r4 and r2/r3 router pairs have one connection each
    tgen.add_link(r1, r4, ifname1="eth3", ifname2="eth1")
    tgen.add_link(r2, r3, ifname1="eth3", ifname2="eth1")


@pytest.fixture(scope="function")
def tgen(request):
    """Setup/Teardown the environment and provide tgen argument to tests.

    Do this once per function as some of the tests will leave the router
    in an unclean state.

    """

    tgen = Topogen(build_topo, request.module.__name__)
    tgen.start_topology()

    router_list = tgen.routers()

    for rname, router in router_list.items():
        router.load_frr_config("frr.conf")

    tgen.start_router()

    yield tgen

    tgen.stop_topology()


def ospf_neighbors(router, ip_version):
    """List the OSPF neighbors for the given router and IP version."""

    if ip_version == 4:
        cmd = "show ip ospf neighbor json"
    else:
        cmd = "show ipv6 ospf neighbor json"

    output = router.vtysh_cmd(cmd)

    if ip_version == 4:
        return [v for n in json.loads(output)["neighbors"].values() for v in n]
    else:
        return json.loads(output)["neighbors"]


def ospf_neighbor_uptime(router, interface, ip_version):
    """Uptime of the neighbor with the given interface name in seconds."""

    for neighbor in ospf_neighbors(router, ip_version):
        if ip_version == 4:
            if not neighbor["ifaceName"].startswith("{}:".format(interface)):
                continue

            return neighbor["upTimeInMsec"] / 1000
        else:
            if neighbor["interfaceName"] != interface:
                continue

            h, m, s = [int(d) for d in neighbor["duration"].split(":")]
            return h * 3600 + m * 60 + s

    raise KeyError(
        "No IPv{} neighbor with interface name {} on {}".format(
            ip_version, interface, router.name
        )
    )


def ospf_routes(router, prefix):
    """List the OSPF routes for the given router and prefix."""

    if ipaddress.ip_interface(prefix).ip.version == 4:
        cmd = "show ip route {} json"
    else:
        cmd = "show ipv6 route {} json"

    output = router.vtysh_cmd(cmd.format(prefix))
    return json.loads(output)[prefix]


def ospf_nexthops(router, prefix, protocol):
    """List the OSPF nexthops for the given prefix."""

    for route in ospf_routes(router, prefix):
        if route["protocol"] != protocol:
            continue

        for nexthop in route["nexthops"]:
            yield nexthop


def ospf_directly_connected_interfaces(router, ip_version):
    """The names of the directly connected interfaces, as discovered
    through the OSPF nexthops.

    """

    if ip_version == 4:
        prefix = "192.0.2.{}/32".format(router.name.strip("r"))
    else:
        prefix = "fe80::/64"

    hops = ospf_nexthops(router, prefix, protocol="connected")
    return sorted([n["interfaceName"] for n in hops if n["directlyConnected"]])


def wait_for_ospf(router, ip_version, neighbors, timeout=60):
    """Wait until the router has the given number of neighbors that are
    fully converged.

    Note that this checks for the exact number of neighbors, so if one neighbor
    is requested and three are converged, the wait continues.

    """

    until = time.monotonic() + timeout

    if ip_version == 4:
        filter = {"converged": "Full"}
    else:
        filter = {"state": "Full"}

    def is_match(neighbor):
        for k, v in filter.items():
            if neighbor[k] != v:
                return False

        return True

    while time.monotonic() < until:
        found = sum(1 for n in ospf_neighbors(router, ip_version) if is_match(n))

        if neighbors == found:
            return

    raise TimeoutError(
        "Waited over {}s for {} neighbors to reach {}".format(
            timeout, neighbors, filter
        )
    )


@pytest.mark.parametrize("ip_version", [4, 6])
def test_interface_up(tgen, ip_version):
    """Verify the initial routing table, before any changes."""

    # Wait for the routers to be ready
    routers = {id: tgen.gears[id] for id in ("r1", "r2", "r3", "r4")}

    for router in routers.values():
        wait_for_ospf(router, ip_version=ip_version, neighbors=3)

    # Verify that the link-local routes are correct
    for router in routers.values():
        connected = ospf_directly_connected_interfaces(router, ip_version)

        if ip_version == 4:
            expected = ["eth1", "eth2", "eth3", "lo"]
        else:
            expected = ["eth1", "eth2", "eth3"]

        assert (
            connected == expected
        ), "Expected all interfaces to be connected on {}".format(router.name)


@pytest.mark.parametrize("ip_version", [4, 6])
def test_interface_down(tgen, ip_version):
    """Verify the routing table after taking interfaces down."""

    # Wait for the routers to be ready
    routers = {id: tgen.gears[id] for id in ("r1", "r2", "r3", "r4")}

    for id, router in routers.items():
        wait_for_ospf(router, ip_version=ip_version, neighbors=3)

    # Keep track of the uptime of the eth3 neighbor
    uptime = ospf_neighbor_uptime(routers["r1"], "eth3", ip_version)
    before = time.monotonic()

    # Take the links between r1 and r2 down
    routers["r1"].cmd_raises("ip link set down dev eth1")
    routers["r1"].cmd_raises("ip link set down dev eth2")

    # Wait for OSPF to converge
    wait_for_ospf(routers["r1"], ip_version=ip_version, neighbors=1)

    # The uptime of the unaffected eth3 neighbor should be monotonic
    new_uptime = ospf_neighbor_uptime(routers["r1"], "eth3", ip_version)
    took = round(time.monotonic() - before, 3)

    # IPv6 has a resolution of 1s, for IPv4 some slack is necesssary.
    if ip_version == 4:
        offset = 0.25
    else:
        offset = 1

    assert (
        new_uptime + offset >= uptime + took
    ), "The eth3 neighbor uptime must not decrease"

    # We should only find eth3 once OSPF has converged
    connected = ospf_directly_connected_interfaces(routers["r1"], ip_version)

    if ip_version == 4:
        expected = ["eth3", "lo"]
    else:
        expected = ["eth3"]

    assert connected == expected, "Expected only eth1 and eth2 to be disconnected"


@pytest.mark.parametrize("ip_version", [4, 6])
def test_interface_flap(tgen, ip_version):
    """Verify the routing table after enabling an interface that was down."""

    # Wait for the routers to be ready
    routers = {id: tgen.gears[id] for id in ("r1", "r2", "r3", "r4")}

    for id, router in routers.items():
        wait_for_ospf(router, ip_version=ip_version, neighbors=3)

    # Keep track of the uptime of the eth3 neighbor
    uptime = ospf_neighbor_uptime(routers["r1"], "eth3", ip_version)
    before = time.monotonic()

    # Take the links between r1 and r2 down
    routers["r1"].cmd_raises("ip link set down dev eth1")
    routers["r2"].cmd_raises("ip link set down dev eth2")

    # Wait for OSPF to converge
    wait_for_ospf(routers["r1"], ip_version=ip_version, neighbors=1)

    # Take the links between r1 and r2 up
    routers["r1"].cmd_raises("ip link set up dev eth1")
    routers["r2"].cmd_raises("ip link set up dev eth2")

    # Wait for OSPF to converge
    wait_for_ospf(routers["r1"], ip_version=ip_version, neighbors=3)

    # The uptime of the unaffected eth3 neighbor should be monotonic
    new_uptime = ospf_neighbor_uptime(routers["r1"], "eth3", ip_version)
    took = round(time.monotonic() - before, 3)

    # IPv6 has a resolution of 1s, for IPv4 some slack is necesssary.
    if ip_version == 4:
        offset = 0.25
    else:
        offset = 1

    assert (
        new_uptime + offset >= uptime + took
    ), "The eth3 neighbor uptime must not decrease"

    # We should find all interfaces again
    connected = ospf_directly_connected_interfaces(routers["r1"], ip_version)

    if ip_version == 4:
        expected = ["eth1", "eth2", "eth3", "lo"]
    else:
        expected = ["eth1", "eth2", "eth3"]

    assert connected == expected, "Expected all interfaces to be connected"


if __name__ == "__main__":
    args = ["-s"] + sys.argv[1:]
    sys.exit(pytest.main(args))