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