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