path: root/tests/topotests/ospf_multi_instance/test_ospf_multi_instance.py

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

#
# test_ospf_multi_instance.py
#
# Copyright (c) 2024 LabN Consulting
# Acee Lindem
#

import os
import sys
from functools import partial
import pytest

# 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.common_config import (
    step,
    create_interface_in_kernel,
)


"""
test_ospf_metric_propagation.py: Test OSPF/BGP metric propagation
"""

TOPOLOGY = """

 +---------+             +--------------------+             +---------+
 |   r1    |             |  r2      |   r2    |             |   r3    |
 |         |             | ospf 1   | ospf 2  |             |        |
 | 1.1.1.1 | eth0    eth0| 2.2.2.1  | 2.2.2.2 |eth1     eth0| 3.3.3.1 |
 |         +-------------+          |         +-------------+         |
 |         | 10.1.1.0/24 |          |         | 10.1.2.0/24 |         |
 +---------+             +--------------------+             +---------+


"""

# Save the Current Working Directory to find configuration files.
CWD = os.path.dirname(os.path.realpath(__file__))
sys.path.append(os.path.join(CWD, "../"))

# Required to instantiate the topology builder class.

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


def build_topo(tgen):
    "Build function"

    # Create 3 routers
    tgen.add_router("r1")
    tgen.add_router("r2")
    tgen.add_router("r3")

    # Interconect router 1, 2 (0)
    switch = tgen.add_switch("s1-1-2")
    switch.add_link(tgen.gears["r1"])
    switch.add_link(tgen.gears["r2"])

    # Interconect router 2, 3 (1)
    switch = tgen.add_switch("s2-2-3")
    switch.add_link(tgen.gears["r2"])
    switch.add_link(tgen.gears["r3"])

    # Add more loopbacks to r2
    create_interface_in_kernel(
        tgen, "r2", "lo1", "2.2.2.1", netmask="255.255.255.255", create=True
    )
    create_interface_in_kernel(
        tgen, "r2", "lo2", "2.2.2.2", netmask="255.255.255.255", create=True
    )


def setup_module(mod):
    logger.info("OSPF Multi-Instance:\n {}".format(TOPOLOGY))

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

    # Starting Routers
    router_list = tgen.routers()

    for rname, router in router_list.items():
        logger.info("Loading router %s" % rname)
        router.load_frr_config(os.path.join(CWD, "{}/frr.conf".format(rname)))

    # Initialize all routers.
    tgen.start_router()


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


def test_multi_instance_default_origination():
    tgen = get_topogen()

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

    step("Configure a local default route")
    r1 = tgen.gears["r1"]
    r1.vtysh_cmd("conf t\nip route 0.0.0.0/0 Null0")

    step("Verify the R1 configuration and install of 'ip route 0.0.0.0/0 Null0'")
    prefix_suppression_cfg = (
        tgen.net["r1"]
        .cmd('vtysh -c "show running" | grep "^ip route 0.0.0.0/0 Null0"')
        .rstrip()
    )
    assertmsg = "'ip route 0.0.0.0/0 Null0' applied, but not present in configuration"
    assert prefix_suppression_cfg == "ip route 0.0.0.0/0 Null0", assertmsg

    input_dict = {
        "0.0.0.0/0": [
            {
                "prefix": "0.0.0.0/0",
                "prefixLen": 0,
                "protocol": "static",
                "nexthops": [
                    {
                        "blackhole": True,
                    }
                ],
            }
        ]
    }
    test_func = partial(
        topotest.router_json_cmp, r1, "show ip route 0.0.0.0/0 json", input_dict
    )
    _, result = topotest.run_and_expect(test_func, None, count=60, wait=1)
    assertmsg = "0.0.0.0/0 not installed on router r1"
    assert result is None, assertmsg

    step(
        "Verify the R1 configuration and advertisement of 'default-information originate'"
    )
    r1.vtysh_cmd("conf t\nrouter ospf\n default-information originate")

    input_dict = {
        "asExternalLinkStates": [
            {
                "lsaType": "AS-external-LSA",
                "linkStateId": "0.0.0.0",
                "advertisingRouter": "1.1.1.1",
                "networkMask": 0,
                "metricType": "E2 (Larger than any link state path)",
                "metric": 10,
                "forwardAddress": "0.0.0.0",
                "externalRouteTag": 0,
            }
        ]
    }
    test_func = partial(
        topotest.router_json_cmp, r1, "show ip ospf database json", input_dict
    )

    r2 = tgen.gears["r2"]
    step("Verify the OSPF instance 1 installation of default route on router 2")
    input_dict = {
        "0.0.0.0/0": [
            {
                "prefix": "0.0.0.0/0",
                "prefixLen": 0,
                "protocol": "ospf",
                "instance": 1,
                "nexthops": [
                    {
                        "ip": "10.1.1.1",
                        "interfaceName": "r2-eth0",
                    }
                ],
            }
        ]
    }
    test_func = partial(
        topotest.router_json_cmp, r2, "show ip route 0.0.0.0/0 json", input_dict
    )
    _, result = topotest.run_and_expect(test_func, None, count=60, wait=1)
    assertmsg = "0.0.0.0/0 not installed on router r2"
    assert result is None, assertmsg

    step("Configure OSPF 'default-intformation originate' on router r2 instance 2")
    r2.vtysh_cmd("conf t\nrouter ospf 2\n default-information originate")

    step("Verify r2 instance 2 AS-External default origination")
    input_dict = {
        "ospfInstance": 2,
        "routerId": "2.2.2.2",
        "asExternalLinkStates": [
            {
                "lsaType": "AS-external-LSA",
                "linkStateId": "0.0.0.0",
                "advertisingRouter": "2.2.2.2",
                "networkMask": 0,
                "metricType": "E2 (Larger than any link state path)",
                "tos": 0,
                "metric": 10,
                "forwardAddress": "0.0.0.0",
                "externalRouteTag": 0,
            }
        ],
    }
    test_func = partial(
        topotest.router_json_cmp,
        r2,
        "show ip ospf 2 database external json",
        input_dict,
    )
    _, result = topotest.run_and_expect(test_func, None, count=60, wait=1)
    assertmsg = "AS-External default not originated by router r2 OSPF instance 2"
    assert result is None, assertmsg

    step("Update the OSPF instance 2 distance so it will be preferred over instance 1")
    r2.vtysh_cmd("conf t\nrouter ospf 2\n distance 15")

    step("Generate a default route from OSPF on r3")
    r3 = tgen.gears["r3"]
    r3.vtysh_cmd("conf t\nrouter ospf\n default-information originate")
    r3.vtysh_cmd("conf t\nip route 0.0.0.0/0 Null0")

    step("Verify r3 AS-External default origination on r2")
    input_dict = {
        "ospfInstance": 2,
        "routerId": "2.2.2.2",
        "asExternalLinkStates": [
            {
                "lsaType": "AS-external-LSA",
                "linkStateId": "0.0.0.0",
                "advertisingRouter": "3.3.3.1",
                "length": 36,
                "networkMask": 0,
                "metricType": "E2 (Larger than any link state path)",
                "tos": 0,
                "metric": 10,
                "forwardAddress": "0.0.0.0",
                "externalRouteTag": 0,
            }
        ],
    }
    test_func = partial(
        topotest.router_json_cmp,
        r2,
        "show ip ospf 2 database external json",
        input_dict,
    )
    _, result = topotest.run_and_expect(test_func, None, count=60, wait=1)
    assertmsg = "AS-External default not originated by router r3 OSPF"
    assert result is None, assertmsg

    step("Verify r3's default installed by OSPF instance 2 is preferred on r2")
    input_dict = {
        "0.0.0.0/0": [
            {
                "prefix": "0.0.0.0/0",
                "prefixLen": 0,
                "protocol": "ospf",
                "instance": 2,
                "distance": 15,
                "nexthops": [
                    {
                        "ip": "10.1.2.3",
                        "interfaceName": "r2-eth1",
                    }
                ],
            }
        ]
    }
    test_func = partial(
        topotest.router_json_cmp, r2, "show ip route 0.0.0.0/0 json", input_dict
    )
    _, result = topotest.run_and_expect(test_func, None, count=60, wait=1)
    assertmsg = "0.0.0.0/0 from r3 not installed on router r2"
    assert result is None, assertmsg

    step(
        "Verify that r2's OSPF instance 2 AS-External LSA default is flushed due to default from r3"
    )
    input_dict = {
        "ospfInstance": 2,
        "routerId": "2.2.2.2",
        "asExternalLinkStates": [
            {
                "lsaAge": 3600,
                "lsaType": "AS-external-LSA",
                "linkStateId": "0.0.0.0",
                "advertisingRouter": "2.2.2.2",
                "networkMask": 0,
                "metricType": "E2 (Larger than any link state path)",
                "tos": 0,
                "metric": 10,
                "forwardAddress": "0.0.0.0",
                "externalRouteTag": 0,
            }
        ],
    }
    test_func = partial(
        topotest.router_json_cmp,
        r2,
        "show ip ospf 2 database external json",
        input_dict,
    )
    _, result = topotest.run_and_expect(test_func, None, count=60, wait=1)
    assertmsg = "AS-External default not flushed by router r2 OSPF instance 2"
    assert result is None, assertmsg

    step("Remove r3's default route and verify that its advertisement is flushed")
    r3.vtysh_cmd("conf t\nno ip route 0.0.0.0/0 Null0")
    input_dict = {
        "routerId": "3.3.3.1",
        "asExternalLinkStates": [
            {
                "lsaAge": 3600,
                "lsaType": "AS-external-LSA",
                "linkStateId": "0.0.0.0",
                "advertisingRouter": "3.3.3.1",
                "networkMask": 0,
                "metricType": "E2 (Larger than any link state path)",
                "tos": 0,
                "metric": 10,
                "forwardAddress": "0.0.0.0",
                "externalRouteTag": 0,
            }
        ],
    }
    test_func = partial(
        topotest.router_json_cmp, r3, "show ip ospf database external json", input_dict
    )
    _, result = topotest.run_and_expect(test_func, None, count=60, wait=1)
    assertmsg = "AS-External default not flushed by router r3 OSPF"
    assert result is None, assertmsg

    step(
        "Verify that r2's OSPF instance 2 AS-External default is advertised and installed by r3"
    )
    input_dict = {
        "routerId": "3.3.3.1",
        "asExternalLinkStates": [
            {
                "lsaType": "AS-external-LSA",
                "linkStateId": "0.0.0.0",
                "advertisingRouter": "2.2.2.2",
                "networkMask": 0,
                "metricType": "E2 (Larger than any link state path)",
                "tos": 0,
                "metric": 10,
                "forwardAddress": "0.0.0.0",
                "externalRouteTag": 0,
            }
        ],
    }
    test_func = partial(
        topotest.router_json_cmp, r3, "show ip ospf database external json", input_dict
    )
    assertmsg = "AS-External default not originated by r2 OSPF instance 2"
    assert result is None, assertmsg

    step("Verify r2's OSPF instance 2 is AS-External default is installed on r3")
    input_dict = {
        "0.0.0.0/0": [
            {
                "prefix": "0.0.0.0/0",
                "prefixLen": 0,
                "protocol": "ospf",
                "distance": 20,
                "nexthops": [
                    {
                        "ip": "10.1.2.2",
                        "interfaceName": "r3-eth0",
                    }
                ],
            }
        ]
    }
    test_func = partial(
        topotest.router_json_cmp, r3, "show ip route 0.0.0.0/0 json", input_dict
    )
    _, result = topotest.run_and_expect(test_func, None, count=60, wait=1)
    assertmsg = "0.0.0.0/0 from router r2 not installed on r3"
    assert result is None, assertmsg


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