tests/topotests/nhrp_topo/test_nhrp_topo.py

   1 #!/usr/bin/env python
   2
   3 #
   4 # test_nhrp_topo.py
   5 # Part of NetDEF Topology Tests
   6 #
   7 # Copyright (c) 2019 by
   8 # Network Device Education Foundation, Inc. ("NetDEF")
   9 #
  10 # Permission to use, copy, modify, and/or distribute this software
  11 # for any purpose with or without fee is hereby granted, provided
  12 # that the above copyright notice and this permission notice appear
  13 # in all copies.
  14 #
  15 # THE SOFTWARE IS PROVIDED "AS IS" AND NETDEF DISCLAIMS ALL WARRANTIES
  16 # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
  17 # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NETDEF BE LIABLE FOR
  18 # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
  19 # DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
  20 # WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
  21 # ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
  22 # OF THIS SOFTWARE.
  23 #
  24
  25 """
  26 test_nhrp_topo.py: Test the FRR/Quagga NHRP daemon
  27 """
  28
  29 import os
  30 import sys
  31 import json
  32 from functools import partial
  33 import pytest
  34
  35 # Save the Current Working Directory to find configuration files.
  36 CWD = os.path.dirname(os.path.realpath(__file__))
  37 sys.path.append(os.path.join(CWD, "../"))
  38
  39 # pylint: disable=C0413
  40 # Import topogen and topotest helpers
  41 from lib import topotest
  42 from lib.topogen import Topogen, TopoRouter, get_topogen
  43 from lib.topolog import logger
  44 from lib.common_config import required_linux_kernel_version
  45
  46 # Required to instantiate the topology builder class.
  47
  48 pytestmark = [pytest.mark.nhrpd]
  49
  50
  51 def build_topo(tgen):
  52     "Build function"
  53
  54     # Create 3 routers.
  55     for routern in range(1, 4):
  56         tgen.add_router("r{}".format(routern))
  57
  58     switch = tgen.add_switch("s1")
  59     switch.add_link(tgen.gears["r1"])
  60     switch.add_link(tgen.gears["r3"])
  61     switch = tgen.add_switch("s2")
  62     switch.add_link(tgen.gears["r2"])
  63     switch.add_link(tgen.gears["r3"])
  64     switch = tgen.add_switch("s3")
  65     switch.add_link(tgen.gears["r2"])
  66     switch = tgen.add_switch("s4")
  67     switch.add_link(tgen.gears["r1"])
  68
  69
  70 def _populate_iface():
  71     tgen = get_topogen()
  72     cmds_tot_hub = [
  73         "ip tunnel add {0}-gre0 mode gre ttl 64 key 42 dev {0}-eth0 local 10.2.1.{1} remote 0.0.0.0",
  74         "ip link set dev {0}-gre0 up",
  75         "echo 0 > /proc/sys/net/ipv4/ip_forward_use_pmtu",
  76         "echo 1 > /proc/sys/net/ipv6/conf/{0}-eth0/disable_ipv6",
  77         "echo 1 > /proc/sys/net/ipv6/conf/{0}-gre0/disable_ipv6",
  78     ]
  79
  80     cmds_tot = [
  81         "ip tunnel add {0}-gre0 mode gre ttl 64 key 42 dev {0}-eth0 local 10.1.1.{1} remote 0.0.0.0",
  82         "ip link set dev {0}-gre0 up",
  83         "echo 0 > /proc/sys/net/ipv4/ip_forward_use_pmtu",
  84         "echo 1 > /proc/sys/net/ipv6/conf/{0}-eth0/disable_ipv6",
  85         "echo 1 > /proc/sys/net/ipv6/conf/{0}-gre0/disable_ipv6",
  86     ]
  87
  88     for cmd in cmds_tot_hub:
  89         input = cmd.format("r2", "2")
  90         logger.info("input: " + input)
  91         output = tgen.net["r2"].cmd(input)
  92         logger.info("output: " + output)
  93
  94     for cmd in cmds_tot:
  95         input = cmd.format("r1", "1")
  96         logger.info("input: " + input)
  97         output = tgen.net["r1"].cmd(input)
  98         logger.info("output: " + output)
  99
 100
 101 def setup_module(mod):
 102     "Sets up the pytest environment"
 103
 104     result = required_linux_kernel_version("5.0")
 105     if result is not True:
 106         pytest.skip("Kernel requirements are not met")
 107
 108     tgen = Topogen(build_topo, mod.__name__)
 109     tgen.start_topology()
 110
 111     router_list = tgen.routers()
 112     _populate_iface()
 113
 114     for rname, router in router_list.items():
 115         router.load_config(
 116             TopoRouter.RD_ZEBRA,
 117             os.path.join(CWD, "{}/zebra.conf".format(rname)),
 118         )
 119         if rname in ("r1", "r2"):
 120             router.load_config(
 121                 TopoRouter.RD_NHRP, os.path.join(CWD, "{}/nhrpd.conf".format(rname))
 122             )
 123
 124     # Initialize all routers.
 125     logger.info("Launching NHRP")
 126     for name in router_list:
 127         router = tgen.gears[name]
 128         router.start()
 129
 130
 131 def teardown_module(_mod):
 132     "Teardown the pytest environment"
 133     tgen = get_topogen()
 134     tgen.stop_topology()
 135
 136
 137 def test_protocols_convergence():
 138     """
 139     Assert that all protocols have converged before checking for the NHRP
 140     statuses as they depend on it.
 141     """
 142     tgen = get_topogen()
 143     if tgen.routers_have_failure():
 144         pytest.skip(tgen.errors)
 145
 146     # Check IPv4 routing tables.
 147     logger.info("Checking NHRP cache and IPv4 routes for convergence")
 148     router_list = tgen.routers()
 149
 150     for rname, router in router_list.items():
 151         if rname == "r3":
 152             continue
 153
 154         json_file = "{}/{}/nhrp4_cache.json".format(CWD, router.name)
 155         if not os.path.isfile(json_file):
 156             logger.info("skipping file {}".format(json_file))
 157             continue
 158
 159         expected = json.loads(open(json_file).read())
 160         test_func = partial(
 161             topotest.router_json_cmp, router, "show ip nhrp cache json", expected
 162         )
 163         _, result = topotest.run_and_expect(test_func, None, count=40, wait=0.5)
 164
 165         output = router.vtysh_cmd("show ip nhrp cache")
 166         logger.info(output)
 167
 168         assertmsg = '"{}" JSON output mismatches'.format(router.name)
 169         assert result is None, assertmsg
 170
 171     for rname, router in router_list.items():
 172         if rname == "r3":
 173             continue
 174
 175         json_file = "{}/{}/nhrp_route4.json".format(CWD, router.name)
 176         if not os.path.isfile(json_file):
 177             logger.info("skipping file {}".format(json_file))
 178             continue
 179
 180         expected = json.loads(open(json_file).read())
 181         test_func = partial(
 182             topotest.router_json_cmp, router, "show ip route nhrp json", expected
 183         )
 184         _, result = topotest.run_and_expect(test_func, None, count=40, wait=0.5)
 185
 186         output = router.vtysh_cmd("show ip route nhrp")
 187         logger.info(output)
 188
 189         assertmsg = '"{}" JSON output mismatches'.format(router.name)
 190         assert result is None, assertmsg
 191
 192     for rname, router in router_list.items():
 193         if rname == "r3":
 194             continue
 195         logger.info("Dump neighbor information on {}-gre0".format(rname))
 196         output = router.run("ip neigh show")
 197         logger.info(output)
 198
 199
 200 def test_nhrp_connection():
 201     "Assert that the NHRP peers can find themselves."
 202     tgen = get_topogen()
 203     if tgen.routers_have_failure():
 204         pytest.skip(tgen.errors)
 205
 206     pingrouter = tgen.gears["r1"]
 207     logger.info("Check Ping IPv4 from  R1 to R2 = 10.255.255.2)")
 208     output = pingrouter.run("ping 10.255.255.2 -f -c 1000")
 209     logger.info(output)
 210     if "1000 packets transmitted, 1000 received" not in output:
 211         assertmsg = "expected ping IPv4 from R1 to R2 should be ok"
 212         assert 0, assertmsg
 213     else:
 214         logger.info("Check Ping IPv4 from R1 to R2 OK")
 215
 216
 217 def test_memory_leak():
 218     "Run the memory leak test and report results."
 219     tgen = get_topogen()
 220     if not tgen.is_memleak_enabled():
 221         pytest.skip("Memory leak test/report is disabled")
 222
 223     tgen.report_memory_leaks()
 224
 225
 226 if __name__ == "__main__":
 227     args = ["-s"] + sys.argv[1:]
 228     sys.exit(pytest.main(args))