tests/topotests/bfd_topo2/test_bfd_topo2.py

   1 #!/usr/bin/env python
   2
   3 #
   4 # test_bfd_topo2.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_bfd_topo2.py: Test the FRR BFD daemon with multihop and BGP
  27 unnumbered.
  28 """
  29
  30 import os
  31 import sys
  32 import json
  33 from functools import partial
  34 import pytest
  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 pytestmark = [pytest.mark.bfdd, pytest.mark.bgpd, pytest.mark.ospfd]
  47
  48
  49 def setup_module(mod):
  50     "Sets up the pytest environment"
  51     topodef = {
  52         "s1": ("r1", "r2"),
  53         "s2": ("r2", "r3"),
  54         "s3": ("r2", "r4"),
  55     }
  56     tgen = Topogen(topodef, mod.__name__)
  57     tgen.start_topology()
  58
  59     router_list = tgen.routers()
  60     for rname, router in router_list.items():
  61         daemon_file = "{}/{}/zebra.conf".format(CWD, rname)
  62         router.load_config(TopoRouter.RD_ZEBRA, daemon_file)
  63
  64         daemon_file = "{}/{}/bfdd.conf".format(CWD, rname)
  65         if os.path.isfile(daemon_file):
  66             router.load_config(TopoRouter.RD_BFD, daemon_file)
  67
  68         daemon_file = "{}/{}/bgpd.conf".format(CWD, rname)
  69         if os.path.isfile(daemon_file):
  70             router.load_config(TopoRouter.RD_BGP, daemon_file)
  71
  72         daemon_file = "{}/{}/ospfd.conf".format(CWD, rname)
  73         if os.path.isfile(daemon_file):
  74             router.load_config(TopoRouter.RD_OSPF, daemon_file)
  75
  76         daemon_file = "{}/{}/ospf6d.conf".format(CWD, rname)
  77         if os.path.isfile(daemon_file):
  78             router.load_config(TopoRouter.RD_OSPF6, daemon_file)
  79
  80     # Initialize all routers.
  81     tgen.start_router()
  82
  83
  84 def teardown_module(_mod):
  85     "Teardown the pytest environment"
  86     tgen = get_topogen()
  87     tgen.stop_topology()
  88
  89
  90 def test_protocols_convergence():
  91     """
  92     Assert that all protocols have converged before checking for the BFD
  93     statuses as they depend on it.
  94     """
  95     tgen = get_topogen()
  96     if tgen.routers_have_failure():
  97         pytest.skip(tgen.errors)
  98
  99     # Check IPv4 routing tables.
 100     logger.info("Checking IPv4 routes for convergence")
 101     for router in tgen.routers().values():
 102         json_file = "{}/{}/ipv4_routes.json".format(CWD, router.name)
 103         if not os.path.isfile(json_file):
 104             logger.info("skipping file {}".format(json_file))
 105             continue
 106
 107         expected = json.loads(open(json_file).read())
 108         test_func = partial(
 109             topotest.router_json_cmp, router, "show ip route json", expected
 110         )
 111         _, result = topotest.run_and_expect(test_func, None, count=40, wait=2)
 112         assertmsg = '"{}" JSON output mismatches'.format(router.name)
 113         assert result is None, assertmsg
 114
 115     # Check IPv6 routing tables.
 116     logger.info("Checking IPv6 routes for convergence")
 117     for router in tgen.routers().values():
 118         json_file = "{}/{}/ipv6_routes.json".format(CWD, router.name)
 119         if not os.path.isfile(json_file):
 120             logger.info("skipping file {}".format(json_file))
 121             continue
 122
 123         expected = json.loads(open(json_file).read())
 124         test_func = partial(
 125             topotest.router_json_cmp, router, "show ipv6 route json", expected
 126         )
 127         _, result = topotest.run_and_expect(test_func, None, count=40, wait=2)
 128         assertmsg = '"{}" JSON output mismatches'.format(router.name)
 129         assert result is None, assertmsg
 130
 131
 132 def test_bfd_connection():
 133     "Assert that the BFD peers can find themselves."
 134     tgen = get_topogen()
 135     if tgen.routers_have_failure():
 136         pytest.skip(tgen.errors)
 137
 138     logger.info("waiting for bfd peers to go up")
 139
 140     for router in tgen.routers().values():
 141         json_file = "{}/{}/peers.json".format(CWD, router.name)
 142         expected = json.loads(open(json_file).read())
 143
 144         test_func = partial(
 145             topotest.router_json_cmp, router, "show bfd peers json", expected
 146         )
 147         _, result = topotest.run_and_expect(test_func, None, count=30, wait=0.5)
 148         assertmsg = '"{}" JSON output mismatches'.format(router.name)
 149         assert result is None, assertmsg
 150
 151
 152 def test_memory_leak():
 153     "Run the memory leak test and report results."
 154     tgen = get_topogen()
 155     if not tgen.is_memleak_enabled():
 156         pytest.skip("Memory leak test/report is disabled")
 157
 158     tgen.report_memory_leaks()
 159
 160
 161 if __name__ == "__main__":
 162     args = ["-s"] + sys.argv[1:]
 163     sys.exit(pytest.main(args))