tests/topotests/ospf_topo2/test_ospf_topo2.py

   1 #!/usr/bin/env python
   2
   3 #
   4 # test_ospf_topo2.py
   5 #
   6 # Copyright (c) 2019 by
   7 # Cumulus Networks, Inc
   8 # Donald Sharp
   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_ospf_topo2.py: Test the OSPF unnumbered.
  27 """
  28
  29 import os
  30 import sys
  31 from functools import partial
  32 import pytest
  33 import json
  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
  45 # Required to instantiate the topology builder class.
  46
  47 pytestmark = [pytest.mark.ospfd]
  48
  49
  50 CWD = os.path.dirname(os.path.realpath(__file__))
  51
  52
  53 def build_topo(tgen):
  54     "Build function"
  55
  56     # Create 4 routers
  57     for routern in range(1, 3):
  58         tgen.add_router("r{}".format(routern))
  59
  60     # Create a empty network for router 1
  61     switch = tgen.add_switch("s1")
  62     switch.add_link(tgen.gears["r1"])
  63
  64     # Create a empty network for router 2
  65     switch = tgen.add_switch("s2")
  66     switch.add_link(tgen.gears["r2"])
  67
  68     # Interconect router 1, 2
  69     switch = tgen.add_switch("s3")
  70     switch.add_link(tgen.gears["r1"])
  71     switch.add_link(tgen.gears["r2"])
  72
  73
  74 def setup_module(mod):
  75     "Sets up the pytest environment"
  76     tgen = Topogen(build_topo, mod.__name__)
  77     tgen.start_topology()
  78
  79     router_list = tgen.routers()
  80     for rname, router in router_list.items():
  81         router.load_config(
  82             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
  83         )
  84         router.load_config(
  85             TopoRouter.RD_OSPF, os.path.join(CWD, "{}/ospfd.conf".format(rname))
  86         )
  87
  88         # What is this?  OSPF Unnumbered depends on the rp_filter
  89         # being set appropriately( HA! )
  90         # Effectively we are putting different /32's on the interface
  91         # the multicast packet delivery is somewhat controlled by
  92         # the rp_filter.  Setting it to '0' allows the OS to pass
  93         # up the mcast packet not destined for the local routers
  94         # network.
  95         topotest.sysctl_assure(tgen.net["r1"], "net.ipv4.conf.r1-eth1.rp_filter", 0)
  96         topotest.sysctl_assure(tgen.net["r1"], "net.ipv4.conf.all.rp_filter", 0)
  97         topotest.sysctl_assure(tgen.net["r2"], "net.ipv4.conf.r2-eth1.rp_filter", 0)
  98         topotest.sysctl_assure(tgen.net["r2"], "net.ipv4.conf.all.rp_filter", 0)
  99
 100     # Initialize all routers.
 101     tgen.start_router()
 102     # tgen.mininet_cli()
 103
 104
 105 def teardown_module(mod):
 106     "Teardown the pytest environment"
 107     tgen = get_topogen()
 108     tgen.stop_topology()
 109
 110
 111 def test_ospf_convergence():
 112     "Test OSPF daemon convergence and that we have received the ospf routes"
 113     tgen = get_topogen()
 114     if tgen.routers_have_failure():
 115         pytest.skip("skipped because of router(s) failure")
 116
 117     for router, rnode in tgen.routers().items():
 118         logger.info('Waiting for router "%s" convergence', router)
 119
 120         json_file = "{}/{}/ospf-route.json".format(CWD, router)
 121         expected = json.loads(open(json_file).read())
 122
 123         test_func = partial(
 124             topotest.router_json_cmp, rnode, "show ip ospf route json", expected
 125         )
 126         _, result = topotest.run_and_expect(test_func, None, count=160, wait=0.5)
 127         assertmsg = '"{}" JSON output mismatches'.format(router)
 128         assert result is None, assertmsg
 129     # tgen.mininet_cli()
 130
 131
 132 def test_ospf_kernel_route():
 133     "Test OSPF kernel route installation and we have the onlink success"
 134     tgen = get_topogen()
 135     if tgen.routers_have_failure():
 136         pytest.skip("skipped because of router(s) failure")
 137
 138     rlist = tgen.routers().values()
 139     for router in rlist:
 140         logger.info('Checking OSPF IPv4 kernel routes in "%s"', router.name)
 141
 142         json_file = "{}/{}/v4_route.json".format(CWD, router.name)
 143         expected = json.loads(open(json_file).read())
 144
 145         test_func = partial(
 146             topotest.router_json_cmp, router, "show ip route json", expected
 147         )
 148         _, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
 149         assertmsg = '"{}" JSON output mistmatches'.format(router)
 150         assert result is None, assertmsg
 151     # tgen.mininet_cli()
 152
 153
 154 def test_memory_leak():
 155     "Run the memory leak test and report results."
 156     tgen = get_topogen()
 157     if not tgen.is_memleak_enabled():
 158         pytest.skip("Memory leak test/report is disabled")
 159
 160     tgen.report_memory_leaks()
 161
 162
 163 if __name__ == "__main__":
 164     args = ["-s"] + sys.argv[1:]
 165     sys.exit(pytest.main(args))