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