tests/topotests/bfd-vrf-topo1/test_bfd_vrf_topo1.py

   1 #!/usr/bin/env python
   2
   3 #
   4 # test_bfd_vrf_topo1.py
   5 # Part of NetDEF Topology Tests
   6 #
   7 # Copyright (c) 2018 by
   8 # Network Device Education Foundation, Inc. ("NetDEF")
   9 # Copyright (c) 2019 by 6WIND
  10 #
  11 # Permission to use, copy, modify, and/or distribute this software
  12 # for any purpose with or without fee is hereby granted, provided
  13 # that the above copyright notice and this permission notice appear
  14 # in all copies.
  15 #
  16 # THE SOFTWARE IS PROVIDED "AS IS" AND NETDEF DISCLAIMS ALL WARRANTIES
  17 # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
  18 # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NETDEF BE LIABLE FOR
  19 # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
  20 # DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
  21 # WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
  22 # ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
  23 # OF THIS SOFTWARE.
  24 #
  25
  26 """
  27 test_bfd_vrf_topo1.py: Test the FRR BFD daemon.
  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 # Required to instantiate the topology builder class.
  47 from mininet.topo import Topo
  48
  49
  50 class BFDTopo(Topo):
  51     "Test topology builder"
  52
  53     def build(self, *_args, **_opts):
  54         "Build function"
  55         tgen = get_topogen(self)
  56
  57         # Create 4 routers
  58         for routern in range(1, 5):
  59             tgen.add_router("r{}".format(routern))
  60
  61         switch = tgen.add_switch("s1")
  62         switch.add_link(tgen.gears["r1"])
  63         switch.add_link(tgen.gears["r2"])
  64
  65         switch = tgen.add_switch("s2")
  66         switch.add_link(tgen.gears["r2"])
  67         switch.add_link(tgen.gears["r3"])
  68
  69         switch = tgen.add_switch("s3")
  70         switch.add_link(tgen.gears["r2"])
  71         switch.add_link(tgen.gears["r4"])
  72
  73 @pytest.mark.bfd
  74 def setup_module(mod):
  75     "Sets up the pytest environment"
  76     tgen = Topogen(BFDTopo, mod.__name__)
  77     tgen.start_topology()
  78
  79     router_list = tgen.routers()
  80
  81     # check for zebra capability
  82     for rname, router in router_list.items():
  83         if router.check_capability(TopoRouter.RD_ZEBRA, "--vrfwnetns") == False:
  84             return pytest.skip(
  85                 "Skipping BFD Topo1 VRF NETNS feature. VRF NETNS backend not available on FRR"
  86             )
  87
  88     if os.system("ip netns list") != 0:
  89         return pytest.skip(
  90             "Skipping BFD Topo1 VRF NETNS Test. NETNS not available on System"
  91         )
  92
  93     logger.info("Testing with VRF Namespace support")
  94
  95     cmds = [
  96         "if [ -e /var/run/netns/{0}-cust1 ] ; then ip netns del {0}-cust1 ; fi",
  97         "ip netns add {0}-cust1",
  98         "ip link set dev {0}-eth0 netns {0}-cust1",
  99         "ip netns exec {0}-cust1 ifconfig {0}-eth0 up",
 100     ]
 101     cmds2 = [
 102         "ip link set dev {0}-eth1 netns {0}-cust1",
 103         "ip netns exec {0}-cust1 ifconfig {0}-eth1 up",
 104         "ip link set dev {0}-eth2 netns {0}-cust1",
 105         "ip netns exec {0}-cust1 ifconfig {0}-eth2 up",
 106     ]
 107
 108     for rname, router in router_list.items():
 109         # create VRF rx-cust1 and link rx-eth0 to rx-cust1
 110         for cmd in cmds:
 111             output = tgen.net[rname].cmd(cmd.format(rname))
 112         if rname == "r2":
 113             for cmd in cmds2:
 114                 output = tgen.net[rname].cmd(cmd.format(rname))
 115
 116     for rname, router in router_list.items():
 117         router.load_config(
 118             TopoRouter.RD_ZEBRA,
 119             os.path.join(CWD, "{}/zebra.conf".format(rname)),
 120             "--vrfwnetns",
 121         )
 122         router.load_config(
 123             TopoRouter.RD_BFD, os.path.join(CWD, "{}/bfdd.conf".format(rname))
 124         )
 125         router.load_config(
 126             TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format(rname))
 127         )
 128
 129     # Initialize all routers.
 130     tgen.start_router()
 131
 132
 133 def teardown_module(_mod):
 134     "Teardown the pytest environment"
 135     tgen = get_topogen()
 136     # move back rx-eth0 to default VRF
 137     # delete rx-vrf
 138     cmds = [
 139         "ip netns exec {0}-cust1 ip link set {0}-eth0 netns 1",
 140         "ip netns delete {0}-cust1",
 141     ]
 142     cmds2 = [
 143         "ip netns exec {0}-cust1 ip link set {0}-eth1 netns 1",
 144         "ip netns exec {0}-cust2 ip link set {0}-eth1 netns 1",
 145     ]
 146
 147     router_list = tgen.routers()
 148     for rname, router in router_list.items():
 149         if rname == "r2":
 150             for cmd in cmds2:
 151                 tgen.net[rname].cmd(cmd.format(rname))
 152         for cmd in cmds:
 153             tgen.net[rname].cmd(cmd.format(rname))
 154     tgen.stop_topology()
 155
 156
 157 def test_bfd_connection():
 158     "Assert that the BFD peers can find themselves."
 159     tgen = get_topogen()
 160     if tgen.routers_have_failure():
 161         pytest.skip(tgen.errors)
 162
 163     logger.info("waiting for bfd peers to go up")
 164     for router in tgen.routers().values():
 165         json_file = "{}/{}/peers.json".format(CWD, router.name)
 166         expected = json.loads(open(json_file).read())
 167
 168         test_func = partial(
 169             topotest.router_json_cmp, router, "show bfd peers json", expected
 170         )
 171         _, result = topotest.run_and_expect(test_func, None, count=16, wait=1)
 172         assertmsg = '"{}" JSON output mismatches'.format(router.name)
 173         assert result is None, assertmsg
 174
 175
 176 def test_bgp_convergence():
 177     "Assert that BGP is converging."
 178     tgen = get_topogen()
 179     if tgen.routers_have_failure():
 180         pytest.skip(tgen.errors)
 181
 182     logger.info("waiting for bgp peers to go up")
 183
 184     for router in tgen.routers().values():
 185         ref_file = "{}/{}/bgp_summary.json".format(CWD, router.name)
 186         expected = json.loads(open(ref_file).read())
 187         test_func = partial(
 188             topotest.router_json_cmp,
 189             router,
 190             "show ip bgp vrf {}-cust1 summary json".format(router.name),
 191             expected,
 192         )
 193         _, res = topotest.run_and_expect(test_func, None, count=125, wait=1.0)
 194         assertmsg = "{}: bgp did not converge".format(router.name)
 195         assert res is None, assertmsg
 196
 197
 198 def test_bgp_fast_convergence():
 199     "Assert that BGP is converging before setting a link down."
 200     tgen = get_topogen()
 201     if tgen.routers_have_failure():
 202         pytest.skip(tgen.errors)
 203
 204     logger.info("waiting for bgp peers converge")
 205
 206     for router in tgen.routers().values():
 207         ref_file = "{}/{}/bgp_prefixes.json".format(CWD, router.name)
 208         expected = json.loads(open(ref_file).read())
 209         test_func = partial(
 210             topotest.router_json_cmp,
 211             router,
 212             "show ip bgp vrf {}-cust1 json".format(router.name),
 213             expected,
 214         )
 215         _, res = topotest.run_and_expect(test_func, None, count=40, wait=1)
 216         assertmsg = "{}: bgp did not converge".format(router.name)
 217         assert res is None, assertmsg
 218
 219
 220 def test_bfd_fast_convergence():
 221     """
 222     Assert that BFD notices the link down after simulating network
 223     failure.
 224     """
 225     tgen = get_topogen()
 226     if tgen.routers_have_failure():
 227         pytest.skip(tgen.errors)
 228
 229     # Disable r2-eth0 link
 230     router2 = tgen.gears["r2"]
 231     topotest.interface_set_status(
 232         router2, "r2-eth0", ifaceaction=False, vrf_name="r2-cust1"
 233     )
 234
 235     # Wait the minimum time we can before checking that BGP/BFD
 236     # converged.
 237     logger.info("waiting for BFD converge")
 238
 239     # Check that BGP converged quickly.
 240     for router in tgen.routers().values():
 241         json_file = "{}/{}/peers.json".format(CWD, router.name)
 242         expected = json.loads(open(json_file).read())
 243
 244         # Load the same file as previous test, but expect R1 to be down.
 245         if router.name == "r1":
 246             for peer in expected:
 247                 if peer["peer"] == "192.168.0.2":
 248                     peer["status"] = "down"
 249         else:
 250             for peer in expected:
 251                 if peer["peer"] == "192.168.0.1":
 252                     peer["status"] = "down"
 253
 254         test_func = partial(
 255             topotest.router_json_cmp, router, "show bfd peers json", expected
 256         )
 257         _, res = topotest.run_and_expect(test_func, None, count=40, wait=1)
 258         assertmsg = '"{}" JSON output mismatches'.format(router.name)
 259         assert res is None, assertmsg
 260
 261
 262 def test_bgp_fast_reconvergence():
 263     "Assert that BGP is converging after setting a link down."
 264     tgen = get_topogen()
 265     if tgen.routers_have_failure():
 266         pytest.skip(tgen.errors)
 267
 268     logger.info("waiting for BGP re convergence")
 269
 270     # Check that BGP converged quickly.
 271     for router in tgen.routers().values():
 272         ref_file = "{}/{}/bgp_prefixes.json".format(CWD, router.name)
 273         expected = json.loads(open(ref_file).read())
 274
 275         # Load the same file as previous test, but set networks to None
 276         # to test absence.
 277         if router.name == "r1":
 278             expected["routes"]["10.254.254.2/32"] = None
 279             expected["routes"]["10.254.254.3/32"] = None
 280             expected["routes"]["10.254.254.4/32"] = None
 281         else:
 282             expected["routes"]["10.254.254.1/32"] = None
 283
 284         test_func = partial(
 285             topotest.router_json_cmp,
 286             router,
 287             "show ip bgp vrf {}-cust1 json".format(router.name),
 288             expected,
 289         )
 290         _, res = topotest.run_and_expect(test_func, None, count=16, wait=1)
 291         assertmsg = "{}: bgp did not converge".format(router.name)
 292         assert res is None, assertmsg
 293
 294
 295 def test_memory_leak():
 296     "Run the memory leak test and report results."
 297     tgen = get_topogen()
 298     if not tgen.is_memleak_enabled():
 299         pytest.skip("Memory leak test/report is disabled")
 300
 301     tgen.report_memory_leaks()
 302
 303
 304 if __name__ == "__main__":
 305     args = ["-s"] + sys.argv[1:]
 306     sys.exit(pytest.main(args))