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