tests/topotests/bgp_ecmp_topo1/test_bgp_ecmp_topo1.py

   1 #!/usr/bin/env python
   2
   3 #
   4 # test_bgp_ecmp_topo1.py
   5 # Part of NetDEF Topology Tests
   6 #
   7 # Copyright (c) 2017 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_bgp_ecmp_topo1.py: Test BGP topology with ECMP (Equal Cost MultiPath).
  27 """
  28
  29 import json
  30 import functools
  31 import os
  32 import sys
  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 # Required to instantiate the topology builder class.
  46
  47
  48 pytestmark = [pytest.mark.bgpd]
  49
  50
  51 total_ebgp_peers = 20
  52
  53 #####################################################
  54 #
  55 #   Network Topology Definition
  56 #
  57 #####################################################
  58
  59
  60 def build_topo(tgen):
  61     router = tgen.add_router("r1")
  62
  63     # Setup Switches - 1 switch per 5 peering routers
  64     for swNum in range(1, (total_ebgp_peers + 4) // 5 + 1):
  65         switch = tgen.add_switch("s{}".format(swNum))
  66         switch.add_link(router)
  67
  68     # Add 'total_ebgp_peers' number of eBGP ExaBGP neighbors
  69     for peerNum in range(1, total_ebgp_peers + 1):
  70         swNum = (peerNum - 1) // 5 + 1
  71
  72         peer_ip = "10.0.{}.{}".format(swNum, peerNum + 100)
  73         peer_route = "via 10.0.{}.1".format(swNum)
  74         peer = tgen.add_exabgp_peer(
  75             "peer{}".format(peerNum), ip=peer_ip, defaultRoute=peer_route
  76         )
  77
  78         switch = tgen.gears["s{}".format(swNum)]
  79         switch.add_link(peer)
  80
  81
  82 #####################################################
  83 #
  84 #   Tests starting
  85 #
  86 #####################################################
  87
  88
  89 def setup_module(module):
  90     tgen = Topogen(build_topo, module.__name__)
  91     tgen.start_topology()
  92
  93     # Starting Routers
  94     router_list = tgen.routers()
  95     for rname, router in router_list.items():
  96         router.load_config(
  97             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
  98         )
  99         router.load_config(
 100             TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format(rname))
 101         )
 102         router.start()
 103
 104     # Starting Hosts and init ExaBGP on each of them
 105     topotest.sleep(10, "starting BGP on all {} peers".format(total_ebgp_peers))
 106     peer_list = tgen.exabgp_peers()
 107     for pname, peer in peer_list.items():
 108         peer_dir = os.path.join(CWD, pname)
 109         env_file = os.path.join(CWD, "exabgp.env")
 110         peer.start(peer_dir, env_file)
 111         logger.info(pname)
 112
 113
 114 def teardown_module(module):
 115     del module
 116     tgen = get_topogen()
 117     tgen.stop_topology()
 118
 119
 120 def test_bgp_convergence():
 121     "Test for BGP topology convergence"
 122     tgen = get_topogen()
 123
 124     # Skip if previous fatal error condition is raised
 125     if tgen.routers_have_failure():
 126         pytest.skip(tgen.errors)
 127
 128     # Expected result
 129     router = tgen.gears["r1"]
 130     if router.has_version("<", "3.0"):
 131         reffile = os.path.join(CWD, "r1/summary20.txt")
 132     else:
 133         reffile = os.path.join(CWD, "r1/summary.txt")
 134
 135     expected = json.loads(open(reffile).read())
 136
 137     def _output_summary_cmp(router, cmd, data):
 138         """
 139         Runs `cmd` that returns JSON data (normally the command ends
 140         with 'json') and compare with `data` contents.
 141         """
 142         output = router.vtysh_cmd(cmd, isjson=True)
 143         if "ipv4Unicast" in output:
 144             output["ipv4Unicast"]["vrfName"] = output["ipv4Unicast"]["vrfName"].replace(
 145                 "default", "Default"
 146             )
 147         elif "vrfName" in output:
 148             output["vrfName"] = output["vrfName"].replace("default", "Default")
 149         return topotest.json_cmp(output, data)
 150
 151     test_func = functools.partial(
 152         _output_summary_cmp, router, "show ip bgp summary json", expected
 153     )
 154     _, res = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
 155     assertmsg = "BGP router network did not converge"
 156     assert res is None, assertmsg
 157
 158
 159 def test_bgp_ecmp():
 160     tgen = get_topogen()
 161
 162     # Skip if previous fatal error condition is raised
 163     if tgen.routers_have_failure():
 164         pytest.skip(tgen.errors)
 165
 166     expect = {
 167         "routerId": "10.0.255.1",
 168         "routes": {},
 169     }
 170
 171     for net in range(1, 5):
 172         for subnet in range(0, 10):
 173             netkey = "10.20{}.{}.0/24".format(net, subnet)
 174             expect["routes"][netkey] = []
 175             for _ in range(0, 10):
 176                 peer = {"multipath": True, "valid": True}
 177                 expect["routes"][netkey].append(peer)
 178
 179     test_func = functools.partial(
 180         topotest.router_json_cmp, tgen.gears["r1"], "show ip bgp json", expect
 181     )
 182     _, res = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
 183     assertmsg = 'expected multipath routes in "show ip bgp" output'
 184     assert res is None, assertmsg
 185
 186
 187 if __name__ == "__main__":
 188     args = ["-s"] + sys.argv[1:]
 189     sys.exit(pytest.main(args))