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