tests/topotests/ospf_multi_vrf_bgp_route_leak/test_ospf_multi_vrf_bgp_route_leak.py

   1 #!/usr/bin/env python
   2
   3 #
   4 # test_ospf_multi_vrf_bgp_route_leak.py
   5 #
   6 # Copyright (c) 2022 ATCorp
   7 # Jafar Al-Gharaibeh
   8 #
   9 # Permission to use, copy, modify, and/or distribute this software
  10 # for any purpose with or without fee is hereby granted, provided
  11 # that the above copyright notice and this permission notice appear
  12 # in all copies.
  13 #
  14 # THE SOFTWARE IS PROVIDED "AS IS" AND NETDEF DISCLAIMS ALL WARRANTIES
  15 # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
  16 # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NETDEF BE LIABLE FOR
  17 # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
  18 # DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
  19 # WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
  20 # ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
  21 # OF THIS SOFTWARE.
  22 #
  23
  24 import os
  25 import sys
  26 from functools import partial
  27 import pytest
  28
  29 # pylint: disable=C0413
  30 # Import topogen and topotest helpers
  31 from lib import topotest
  32 from lib.topogen import Topogen, TopoRouter, get_topogen
  33 from lib.topolog import logger
  34
  35
  36 """
  37 test_ospf_multi_vrf_bgp_route_leak.py: Test OSPF with multi vrf setup and route leaking.
  38 """
  39
  40 TOPOLOGY = """
  41    bgp route leaking (connected/ospf), vrfs:  neno <==> default <==> ray
  42    routes leaking to vrfs are limited to neno and ray routes.
  43
  44                       10.0.1.1/24
  45                             ^
  46                             |vrf:default
  47                         +---+---+
  48       10.0.30.0/24    .1|       |.1
  49       +-----------------+  R1   +
  50       |        vrf:neno |       |
  51       |                 +---+---+                ^
  52       |.3                 .1|vrf:default         | 10.0.4.4/24
  53   +---+---+                 |                +---+---+
  54   |       |            10.0.20.0/24          |       |
  55   |  R3   |                 |                |  R4   |
  56   |       |                 |.2              |       |
  57   +---+---+             +---+---+            +---+---+
  58       |                 |       | vrf:ray        |.4
  59       v                 |  R2   +----------------+
  60 10.0.3.3/24             |       |.2         10.0.40.0/24
  61                         +---+---+
  62                             |
  63                             v
  64                       10.0.2.2/24
  65
  66 """
  67
  68 # Save the Current Working Directory to find configuration files.
  69 CWD = os.path.dirname(os.path.realpath(__file__))
  70 sys.path.append(os.path.join(CWD, "../"))
  71
  72 # Required to instantiate the topology builder class.
  73
  74 pytestmark = [pytest.mark.ospfd, pytest.mark.bgpd]
  75
  76
  77 def build_topo(tgen):
  78     "Build function"
  79
  80     # Create 4 routers
  81     for routern in range(1, 5):
  82         tgen.add_router("r{}".format(routern))
  83
  84     # Create a empty network for router 1
  85     switch = tgen.add_switch("s1")
  86     switch.add_link(tgen.gears["r1"])
  87
  88     # Create a empty network for router 2
  89     switch = tgen.add_switch("s2")
  90     switch.add_link(tgen.gears["r2"])
  91
  92     # Create a empty network for router 3
  93     switch = tgen.add_switch("s3")
  94     switch.add_link(tgen.gears["r3"])
  95
  96     # Create a empty network for router 4
  97     switch = tgen.add_switch("s4")
  98     switch.add_link(tgen.gears["r4"])
  99
 100     # Interconect router 1, 2
 101     switch = tgen.add_switch("s1-2")
 102     switch.add_link(tgen.gears["r1"])
 103     switch.add_link(tgen.gears["r2"])
 104
 105     # Interconect router 1, 3
 106     switch = tgen.add_switch("s1-3")
 107     switch.add_link(tgen.gears["r1"])
 108     switch.add_link(tgen.gears["r3"])
 109
 110     # Interconect router 2, 4
 111     switch = tgen.add_switch("s2-4")
 112     switch.add_link(tgen.gears["r2"])
 113     switch.add_link(tgen.gears["r4"])
 114
 115
 116 def setup_module(mod):
 117     logger.info("OSPF Multi VRF Topology with BGP route leaking:\n {}".format(TOPOLOGY))
 118
 119     tgen = Topogen(build_topo, mod.__name__)
 120     tgen.start_topology()
 121
 122     r1_vrf_setup_cmds = [
 123         "ip link add name neno type vrf table 11",
 124         "ip link set dev neno up",
 125         "ip link set dev r1-eth2 vrf neno up",
 126     ]
 127     r2_vrf_setup_cmds = [
 128         "ip link add name ray type vrf table 11",
 129         "ip link set dev ray up",
 130         "ip link set dev r2-eth2 vrf ray up",
 131     ]
 132
 133     # Starting Routers
 134     router_list = tgen.routers()
 135
 136     # Create VRFs on r1/r2 and bind to interfaces
 137     for cmd in r1_vrf_setup_cmds:
 138         tgen.net["r1"].cmd(cmd)
 139     for cmd in r2_vrf_setup_cmds:
 140         tgen.net["r2"].cmd(cmd)
 141
 142     logger.info("Testing OSPF VRF support")
 143
 144     for rname, router in router_list.items():
 145         logger.info("Loading router %s" % rname)
 146         router.load_frr_config(os.path.join(CWD, "{}/frr.conf".format(rname)))
 147
 148     # Initialize all routers.
 149     tgen.start_router()
 150     for router in router_list.values():
 151         if router.has_version("<", "4.0"):
 152             tgen.set_error("unsupported version")
 153
 154
 155 def teardown_module(mod):
 156     "Teardown the pytest environment"
 157     tgen = get_topogen()
 158     tgen.stop_topology()
 159
 160
 161 # Shared test function to validate expected output.
 162 def compare_show_ip_route_vrf(rname, expected, vrf_name):
 163     """
 164     Calls 'show ip route vrf [vrf_name] route' and compare the obtained
 165     result with the expected output.
 166     """
 167     tgen = get_topogen()
 168     current = topotest.ip4_route_zebra(tgen.gears[rname], vrf_name)
 169     ret = topotest.difflines(
 170         current, expected, title1="Current output", title2="Expected output"
 171     )
 172     return ret
 173
 174
 175 def test_ospf_convergence():
 176     "Test OSPF daemon convergence"
 177     tgen = get_topogen()
 178
 179     if tgen.routers_have_failure():
 180         pytest.skip("skipped because of router(s) failure")
 181
 182     for rname, router in tgen.routers().items():
 183         logger.info('Waiting for router "%s" convergence', rname)
 184
 185         for vrf in ["default", "neno", "ray"]:
 186             # Load expected results from the command
 187             reffile = os.path.join(CWD, "{}/ospf-vrf-{}.txt".format(rname, vrf))
 188             if vrf == "default" or os.path.exists(reffile):
 189                 expected = open(reffile).read()
 190
 191                 # Run test function until we get an result. Wait at most 80 seconds.
 192                 test_func = partial(
 193                     topotest.router_output_cmp,
 194                     router,
 195                     "show ip ospf vrf {} route".format(vrf),
 196                     expected,
 197                 )
 198                 result, diff = topotest.run_and_expect(
 199                     test_func, "", count=80, wait=1
 200                 )
 201                 assertmsg = "OSPF did not converge on {}:\n{}".format(rname, diff)
 202                 assert result, assertmsg
 203
 204
 205 def test_ospf_kernel_route():
 206     "Test OSPF kernel route installation"
 207     tgen = get_topogen()
 208
 209     if tgen.routers_have_failure():
 210         pytest.skip("skipped because of router(s) failure")
 211
 212     rlist = tgen.routers().values()
 213     for router in rlist:
 214         logger.info('Checking OSPF IPv4 kernel routes in "%s"', router.name)
 215         for vrf in ["default", "neno", "ray"]:
 216             reffile = os.path.join(CWD, "{}/zebra-vrf-{}.txt".format(router.name, vrf))
 217             if vrf == "default" or os.path.exists(reffile):
 218                 expected = open(reffile).read()
 219                 # Run test function until we get an result. Wait at most 80 seconds.
 220                 test_func = partial(
 221                     compare_show_ip_route_vrf, router.name, expected, vrf
 222                 )
 223                 result, diff = topotest.run_and_expect(
 224                     test_func, "", count=80, wait=1
 225                 )
 226                 assertmsg = 'OSPF IPv4 route mismatch in router "{}": {}'.format(
 227                     router.name, diff
 228                 )
 229                 assert result, assertmsg
 230
 231
 232 def test_memory_leak():
 233     "Run the memory leak test and report results."
 234     tgen = get_topogen()
 235     if not tgen.is_memleak_enabled():
 236         pytest.skip("Memory leak test/report is disabled")
 237
 238     tgen.report_memory_leaks()
 239
 240
 241 if __name__ == "__main__":
 242     args = ["-s"] + sys.argv[1:]
 243     sys.exit(pytest.main(args))