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