tests/topotests/bgp_vrf_netns/test_bgp_vrf_netns_topo.py

   1 #!/usr/bin/env python
   2 # SPDX-License-Identifier: ISC
   3
   4 #
   5 # test_bgp_vrf_netns_topo1.py
   6 # Part of NetDEF Topology Tests
   7 #
   8 # Copyright (c) 2018 by 6WIND
   9 #
  10
  11 """
  12 test_bgp_vrf_netns_topo1.py: Test BGP topology with EBGP on NETNS VRF
  13 """
  14
  15 import json
  16 import os
  17 import sys
  18 import functools
  19 import pytest
  20
  21 # Save the Current Working Directory to find configuration files.
  22 CWD = os.path.dirname(os.path.realpath(__file__))
  23 sys.path.append(os.path.join(CWD, "../"))
  24
  25 # pylint: disable=C0413
  26 # Import topogen and topotest helpers
  27 from lib import topotest
  28 from lib.topogen import Topogen, TopoRouter, get_topogen
  29 from lib.topolog import logger
  30
  31 # Required to instantiate the topology builder class.
  32
  33 pytestmark = [pytest.mark.bgpd]
  34
  35
  36 total_ebgp_peers = 1
  37 CustomizeVrfWithNetns = True
  38
  39 #####################################################
  40 ##
  41 ##   Network Topology Definition
  42 ##
  43 #####################################################
  44
  45
  46 def build_topo(tgen):
  47     tgen.add_router("r1")
  48
  49     # Setup Switches
  50     switch = tgen.add_switch("s1")
  51     switch.add_link(tgen.gears["r1"])
  52
  53     # Add eBGP ExaBGP neighbors
  54     peer_ip = "10.0.1.101"
  55     peer_route = "via 10.0.1.1"
  56     peer = tgen.add_exabgp_peer("peer1", ip=peer_ip, defaultRoute=peer_route)
  57     switch = tgen.gears["s1"]
  58     switch.add_link(peer)
  59
  60
  61 #####################################################
  62 ##
  63 ##   Tests starting
  64 ##
  65 #####################################################
  66
  67
  68 def setup_module(module):
  69     tgen = Topogen(build_topo, module.__name__)
  70     tgen.start_topology()
  71
  72     # Get r1 reference
  73     router = tgen.gears["r1"]
  74
  75     # check for zebra capability
  76     if CustomizeVrfWithNetns == True:
  77         if router.check_capability(TopoRouter.RD_ZEBRA, "--vrfwnetns") == False:
  78             return pytest.skip(
  79                 "Skipping BGP VRF NETNS Test. VRF NETNS backend not available on FRR"
  80             )
  81         if os.system("ip netns list") != 0:
  82             return pytest.skip(
  83                 "Skipping BGP VRF NETNS Test. NETNS not available on System"
  84             )
  85     # retrieve VRF backend kind
  86     if CustomizeVrfWithNetns == True:
  87         logger.info("Testing with VRF Namespace support")
  88
  89     # create VRF r1-bgp-cust1
  90     # move r1-eth0 to VRF r1-bgp-cust1
  91
  92     ns = "{}-bgp-cust1".format("r1")
  93     router.net.add_netns(ns)
  94     router.net.set_intf_netns("r1-eth0", ns, up=True)
  95
  96     # run daemons
  97     router.load_config(
  98         TopoRouter.RD_ZEBRA,
  99         os.path.join(CWD, "{}/zebra.conf".format("r1")),
 100         "--vrfwnetns",
 101     )
 102     router.load_config(
 103         TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format("r1"))
 104     )
 105
 106     logger.info("Launching BGP and ZEBRA")
 107     # BGP and ZEBRA start without underlying VRF
 108     router.start()
 109     # Starting Hosts and init ExaBGP on each of them
 110     logger.info("starting exaBGP on peer1")
 111     peer_list = tgen.exabgp_peers()
 112     for pname, peer in peer_list.items():
 113         peer_dir = os.path.join(CWD, pname)
 114         env_file = os.path.join(CWD, "exabgp.env")
 115         logger.info("Running ExaBGP peer")
 116         peer.start(peer_dir, env_file)
 117         logger.info(pname)
 118
 119
 120 def teardown_module(module):
 121     tgen = get_topogen()
 122
 123     # Move interfaces out of vrf namespace and delete the namespace
 124     tgen.net["r1"].reset_intf_netns("r1-eth0")
 125     tgen.net["r1"].delete_netns("r1-bgp-cust1")
 126
 127     tgen.stop_topology()
 128
 129
 130 def test_bgp_vrf_learn():
 131     "Test daemon learnt VRF context"
 132     tgen = get_topogen()
 133
 134     # Skip if previous fatal error condition is raised
 135     if tgen.routers_have_failure():
 136         pytest.skip(tgen.errors)
 137
 138     # Expected result
 139     output = tgen.gears["r1"].vtysh_cmd("show vrf", isjson=False)
 140     logger.info("output is: {}".format(output))
 141
 142     output = tgen.gears["r1"].vtysh_cmd("show bgp vrfs", isjson=False)
 143     logger.info("output is: {}".format(output))
 144
 145
 146 def test_bgp_convergence():
 147     "Test for BGP topology convergence"
 148     tgen = get_topogen()
 149
 150     # uncomment if you want to troubleshoot
 151     # tgen.mininet_cli()
 152     # Skip if previous fatal error condition is raised
 153     if tgen.routers_have_failure():
 154         pytest.skip(tgen.errors)
 155
 156     logger.info("waiting for bgp convergence")
 157
 158     # Expected result
 159     router = tgen.gears["r1"]
 160     if router.has_version("<", "3.0"):
 161         reffile = os.path.join(CWD, "r1/summary20.txt")
 162     else:
 163         reffile = os.path.join(CWD, "r1/summary.txt")
 164
 165     expected = json.loads(open(reffile).read())
 166
 167     test_func = functools.partial(
 168         topotest.router_json_cmp,
 169         router,
 170         "show bgp vrf r1-bgp-cust1 summary json",
 171         expected,
 172     )
 173     _, res = topotest.run_and_expect(test_func, None, count=90, wait=0.5)
 174     assertmsg = "BGP router network did not converge"
 175     assert res is None, assertmsg
 176
 177
 178 def test_bgp_vrf_netns():
 179     tgen = get_topogen()
 180
 181     # Skip if previous fatal error condition is raised
 182     if tgen.routers_have_failure():
 183         pytest.skip(tgen.errors)
 184
 185     expect = {
 186         "routerId": "10.0.1.1",
 187         "routes": {},
 188     }
 189
 190     for subnet in range(0, 10):
 191         netkey = "10.201.{}.0/24".format(subnet)
 192         expect["routes"][netkey] = []
 193         peer = {"valid": True}
 194         expect["routes"][netkey].append(peer)
 195
 196     test_func = functools.partial(
 197         topotest.router_json_cmp,
 198         tgen.gears["r1"],
 199         "show ip bgp vrf r1-bgp-cust1 ipv4 json",
 200         expect,
 201     )
 202     _, res = topotest.run_and_expect(test_func, None, count=12, wait=0.5)
 203     assertmsg = 'expected routes in "show ip bgp vrf r1-bgp-cust1 ipv4" output'
 204     assert res is None, assertmsg
 205
 206
 207 if __name__ == "__main__":
 208
 209     args = ["-s"] + sys.argv[1:]
 210     ret = pytest.main(args)
 211
 212     sys.exit(ret)