tests/topotests/bfd_topo1/test_bfd_topo1.py

   1 #!/usr/bin/env python
   2
   3 #
   4 # test_bfd_topo1.py
   5 # Part of NetDEF Topology Tests
   6 #
   7 # Copyright (c) 2018 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_bfd_topo1.py: Test the FRR BFD 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
  45 # Required to instantiate the topology builder class.
  46 from mininet.topo import Topo
  47
  48 pytestmark = [pytest.mark.bfdd, pytest.mark.bgpd]
  49
  50
  51 class BFDTopo(Topo):
  52     "Test topology builder"
  53
  54     def build(self, *_args, **_opts):
  55         "Build function"
  56         tgen = get_topogen(self)
  57
  58         # Create 4 routers
  59         for routern in range(1, 5):
  60             tgen.add_router("r{}".format(routern))
  61
  62         switch = tgen.add_switch("s1")
  63         switch.add_link(tgen.gears["r1"])
  64         switch.add_link(tgen.gears["r2"])
  65
  66         switch = tgen.add_switch("s2")
  67         switch.add_link(tgen.gears["r2"])
  68         switch.add_link(tgen.gears["r3"])
  69
  70         switch = tgen.add_switch("s3")
  71         switch.add_link(tgen.gears["r2"])
  72         switch.add_link(tgen.gears["r4"])
  73
  74
  75 def setup_module(mod):
  76     "Sets up the pytest environment"
  77     tgen = Topogen(BFDTopo, mod.__name__)
  78     tgen.start_topology()
  79
  80     router_list = tgen.routers()
  81     for rname, router in router_list.items():
  82         router.load_config(
  83             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
  84         )
  85         router.load_config(
  86             TopoRouter.RD_BFD, os.path.join(CWD, "{}/bfdd.conf".format(rname))
  87         )
  88         router.load_config(
  89             TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format(rname))
  90         )
  91
  92     # Initialize all routers.
  93     tgen.start_router()
  94
  95     # Verify that we are using the proper version and that the BFD
  96     # daemon exists.
  97     for router in router_list.values():
  98         # Check for Version
  99         if router.has_version("<", "5.1"):
 100             tgen.set_error("Unsupported FRR version")
 101             break
 102
 103
 104 def teardown_module(_mod):
 105     "Teardown the pytest environment"
 106     tgen = get_topogen()
 107     tgen.stop_topology()
 108
 109
 110 def test_bfd_connection():
 111     "Assert that the BFD peers can find themselves."
 112     tgen = get_topogen()
 113     if tgen.routers_have_failure():
 114         pytest.skip(tgen.errors)
 115
 116     logger.info("waiting for bfd peers to go up")
 117
 118     for router in tgen.routers().values():
 119         json_file = "{}/{}/peers.json".format(CWD, router.name)
 120         expected = json.loads(open(json_file).read())
 121
 122         test_func = partial(
 123             topotest.router_json_cmp, router, "show bfd peers json", expected
 124         )
 125         _, result = topotest.run_and_expect(test_func, None, count=8, wait=0.5)
 126         assertmsg = '"{}" JSON output mismatches'.format(router.name)
 127         assert result is None, assertmsg
 128
 129
 130 def test_bgp_convergence():
 131     "Assert that BGP is converging."
 132     tgen = get_topogen()
 133     if tgen.routers_have_failure():
 134         pytest.skip(tgen.errors)
 135
 136     logger.info("waiting for bgp peers to go up")
 137
 138     for router in tgen.routers().values():
 139         ref_file = "{}/{}/bgp_summary.json".format(CWD, router.name)
 140         expected = json.loads(open(ref_file).read())
 141         test_func = partial(
 142             topotest.router_json_cmp, router, "show ip bgp summary json", expected
 143         )
 144         _, res = topotest.run_and_expect(test_func, None, count=125, wait=1.0)
 145         assertmsg = "{}: bgp did not converge".format(router.name)
 146         assert res is None, assertmsg
 147
 148
 149 def test_bgp_fast_convergence():
 150     "Assert that BGP is converging before setting a link down."
 151     tgen = get_topogen()
 152     if tgen.routers_have_failure():
 153         pytest.skip(tgen.errors)
 154
 155     logger.info("waiting for bgp peers converge")
 156
 157     for router in tgen.routers().values():
 158         ref_file = "{}/{}/bgp_prefixes.json".format(CWD, router.name)
 159         expected = json.loads(open(ref_file).read())
 160         test_func = partial(
 161             topotest.router_json_cmp, router, "show ip bgp json", expected
 162         )
 163         _, res = topotest.run_and_expect(test_func, None, count=40, wait=0.5)
 164         assertmsg = "{}: bgp did not converge".format(router.name)
 165         assert res is None, assertmsg
 166
 167
 168 def test_bfd_fast_convergence():
 169     """
 170     Assert that BFD notices the link down after simulating network
 171     failure.
 172     """
 173     tgen = get_topogen()
 174     if tgen.routers_have_failure():
 175         pytest.skip(tgen.errors)
 176
 177     # Disable r1-eth0 link.
 178     tgen.gears["r1"].link_enable("r1-eth0", enabled=False)
 179
 180     # Wait the minimum time we can before checking that BGP/BFD
 181     # converged.
 182     logger.info("waiting for BFD converge")
 183
 184     # Check that BGP converged quickly.
 185     for router in tgen.routers().values():
 186         json_file = "{}/{}/peers.json".format(CWD, router.name)
 187         expected = json.loads(open(json_file).read())
 188
 189         # Load the same file as previous test, but expect R1 to be down.
 190         if router.name == "r1":
 191             for peer in expected:
 192                 if peer["peer"] == "192.168.0.2":
 193                     peer["status"] = "down"
 194         else:
 195             for peer in expected:
 196                 if peer["peer"] == "192.168.0.1":
 197                     peer["status"] = "down"
 198
 199         test_func = partial(
 200             topotest.router_json_cmp, router, "show bfd peers json", expected
 201         )
 202         _, res = topotest.run_and_expect(test_func, None, count=20, wait=0.5)
 203         assertmsg = '"{}" JSON output mismatches'.format(router.name)
 204         assert res is None, assertmsg
 205
 206
 207 def test_bgp_fast_reconvergence():
 208     "Assert that BGP is converging after setting a link down."
 209     tgen = get_topogen()
 210     if tgen.routers_have_failure():
 211         pytest.skip(tgen.errors)
 212
 213     logger.info("waiting for BGP re convergence")
 214
 215     # Check that BGP converged quickly.
 216     for router in tgen.routers().values():
 217         ref_file = "{}/{}/bgp_prefixes.json".format(CWD, router.name)
 218         expected = json.loads(open(ref_file).read())
 219
 220         # Load the same file as previous test, but set networks to None
 221         # to test absence.
 222         if router.name == "r1":
 223             expected["routes"]["10.254.254.2/32"] = None
 224             expected["routes"]["10.254.254.3/32"] = None
 225             expected["routes"]["10.254.254.4/32"] = None
 226         else:
 227             expected["routes"]["10.254.254.1/32"] = None
 228
 229         test_func = partial(
 230             topotest.router_json_cmp, router, "show ip bgp json", expected
 231         )
 232         _, res = topotest.run_and_expect(test_func, None, count=3, wait=1)
 233         assertmsg = "{}: bgp did not converge".format(router.name)
 234         assert res is None, assertmsg
 235
 236
 237 def test_memory_leak():
 238     "Run the memory leak test and report results."
 239     tgen = get_topogen()
 240     if not tgen.is_memleak_enabled():
 241         pytest.skip("Memory leak test/report is disabled")
 242
 243     tgen.report_memory_leaks()
 244
 245
 246 if __name__ == "__main__":
 247     args = ["-s"] + sys.argv[1:]
 248     sys.exit(pytest.main(args))