tests/topotests/bgp_ecmp_topo1/test_bgp_ecmp_topo1.py

   1 #!/usr/bin/env python2
   2 # SPDX-License-Identifier: ISC
   3
   4 #
   5 # test_bgp_ecmp_topo1.py
   6 # Part of NetDEF Topology Tests
   7 #
   8 # Copyright (c) 2017 by
   9 # Network Device Education Foundation, Inc. ("NetDEF")
  10 #
  11
  12 """
  13 test_bgp_ecmp_topo1.py: Test BGP topology with ECMP (Equal Cost MultiPath).
  14 """
  15
  16 import json
  17 import functools
  18 import os
  19 import sys
  20 import pytest
  21
  22 # Save the Current Working Directory to find configuration files.
  23 CWD = os.path.dirname(os.path.realpath(__file__))
  24 sys.path.append(os.path.join(CWD, "../"))
  25
  26 # pylint: disable=C0413
  27 # Import topogen and topotest helpers
  28 from lib import topotest
  29 from lib.topogen import Topogen, TopoRouter, get_topogen
  30 from lib.topolog import logger
  31
  32 # Required to instantiate the topology builder class.
  33
  34
  35 pytestmark = [pytest.mark.bgpd]
  36
  37
  38 total_ebgp_peers = 20
  39
  40 #####################################################
  41 #
  42 #   Network Topology Definition
  43 #
  44 #####################################################
  45
  46
  47 def build_topo(tgen):
  48     router = tgen.add_router("r1")
  49
  50     # Setup Switches - 1 switch per 5 peering routers
  51     for swNum in range(1, (total_ebgp_peers + 4) // 5 + 1):
  52         switch = tgen.add_switch("s{}".format(swNum))
  53         switch.add_link(router)
  54
  55     # Add 'total_ebgp_peers' number of eBGP ExaBGP neighbors
  56     for peerNum in range(1, total_ebgp_peers + 1):
  57         swNum = (peerNum - 1) // 5 + 1
  58
  59         peer_ip = "10.0.{}.{}".format(swNum, peerNum + 100)
  60         peer_route = "via 10.0.{}.1".format(swNum)
  61         peer = tgen.add_exabgp_peer(
  62             "peer{}".format(peerNum), ip=peer_ip, defaultRoute=peer_route
  63         )
  64
  65         switch = tgen.gears["s{}".format(swNum)]
  66         switch.add_link(peer)
  67
  68
  69 #####################################################
  70 #
  71 #   Tests starting
  72 #
  73 #####################################################
  74
  75
  76 def setup_module(module):
  77     tgen = Topogen(build_topo, module.__name__)
  78     tgen.start_topology()
  79
  80     # Starting Routers
  81     router_list = tgen.routers()
  82     for rname, router in router_list.items():
  83         router.load_config(
  84             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
  85         )
  86         router.load_config(
  87             TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format(rname))
  88         )
  89         router.start()
  90
  91     # Starting Hosts and init ExaBGP on each of them
  92     topotest.sleep(10, "starting BGP on all {} peers".format(total_ebgp_peers))
  93     peer_list = tgen.exabgp_peers()
  94     for pname, peer in peer_list.items():
  95         peer_dir = os.path.join(CWD, pname)
  96         env_file = os.path.join(CWD, "exabgp.env")
  97         peer.start(peer_dir, env_file)
  98         logger.info(pname)
  99
 100
 101 def teardown_module(module):
 102     del module
 103     tgen = get_topogen()
 104     tgen.stop_topology()
 105
 106
 107 def test_bgp_convergence():
 108     "Test for BGP topology convergence"
 109     tgen = get_topogen()
 110
 111     # Skip if previous fatal error condition is raised
 112     if tgen.routers_have_failure():
 113         pytest.skip(tgen.errors)
 114
 115     # Expected result
 116     router = tgen.gears["r1"]
 117     if router.has_version("<", "3.0"):
 118         reffile = os.path.join(CWD, "r1/summary20.txt")
 119     else:
 120         reffile = os.path.join(CWD, "r1/summary.txt")
 121
 122     expected = json.loads(open(reffile).read())
 123
 124     def _output_summary_cmp(router, cmd, data):
 125         """
 126         Runs `cmd` that returns JSON data (normally the command ends
 127         with 'json') and compare with `data` contents.
 128         """
 129         output = router.vtysh_cmd(cmd, isjson=True)
 130         return topotest.json_cmp(output, data)
 131
 132     test_func = functools.partial(
 133         _output_summary_cmp, router, "show ip bgp summary json", expected
 134     )
 135     _, res = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
 136     assertmsg = "BGP router network did not converge"
 137     assert res is None, assertmsg
 138
 139
 140 def test_bgp_ecmp():
 141     tgen = get_topogen()
 142
 143     # Skip if previous fatal error condition is raised
 144     if tgen.routers_have_failure():
 145         pytest.skip(tgen.errors)
 146
 147     expect = {
 148         "routerId": "10.0.255.1",
 149         "routes": {},
 150     }
 151
 152     for net in range(1, 5):
 153         for subnet in range(0, 10):
 154             netkey = "10.20{}.{}.0/24".format(net, subnet)
 155             expect["routes"][netkey] = []
 156             for _ in range(0, 10):
 157                 peer = {"multipath": True, "valid": True}
 158                 expect["routes"][netkey].append(peer)
 159
 160     test_func = functools.partial(
 161         topotest.router_json_cmp, tgen.gears["r1"], "show ip bgp json", expect
 162     )
 163     _, res = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
 164     assertmsg = 'expected multipath routes in "show ip bgp" output'
 165     assert res is None, assertmsg
 166
 167
 168 if __name__ == "__main__":
 169     args = ["-s"] + sys.argv[1:]
 170     sys.exit(pytest.main(args))