tests/topotests/bgp_vpnv4_ebgp/test_bgp_vpnv4_ebgp.py

   1 #!/usr/bin/env python
   2
   3 #
   4 # test_bgp_vpnv4_ebgp.py
   5 # Part of NetDEF Topology Tests
   6 #
   7 # Copyright (c) 2022 by 6WIND
   8 #
   9 # Permission to use, copy, modify, and/or distribute this software
  10 # for any purpose with or without fee is hereby granted, provided
  11 # that the above copyright notice and this permission notice appear
  12 # in all copies.
  13 #
  14 # THE SOFTWARE IS PROVIDED "AS IS" AND NETDEF DISCLAIMS ALL WARRANTIES
  15 # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
  16 # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NETDEF BE LIABLE FOR
  17 # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
  18 # DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
  19 # WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
  20 # ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
  21 # OF THIS SOFTWARE.
  22 #
  23
  24 """
  25  test_bgp_vpnv4_ebgp.py: Test the FRR BGP daemon with EBGP direct connection
  26 """
  27
  28 import os
  29 import sys
  30 import json
  31 from functools import partial
  32 import pytest
  33
  34 # Save the Current Working Directory to find configuration files.
  35 CWD = os.path.dirname(os.path.realpath(__file__))
  36 sys.path.append(os.path.join(CWD, "../"))
  37
  38 # pylint: disable=C0413
  39 # Import topogen and topotest helpers
  40 from lib import topotest
  41 from lib.topogen import Topogen, TopoRouter, get_topogen
  42 from lib.topolog import logger
  43
  44 # Required to instantiate the topology builder class.
  45
  46
  47 pytestmark = [pytest.mark.bgpd]
  48
  49
  50 def build_topo(tgen):
  51     "Build function"
  52
  53     # Create 2 routers.
  54     tgen.add_router("r1")
  55     tgen.add_router("r2")
  56
  57     switch = tgen.add_switch("s1")
  58     switch.add_link(tgen.gears["r1"])
  59     switch.add_link(tgen.gears["r2"])
  60
  61     switch = tgen.add_switch("s2")
  62     switch.add_link(tgen.gears["r1"])
  63
  64     switch = tgen.add_switch("s3")
  65     switch.add_link(tgen.gears["r2"])
  66
  67 def _populate_iface():
  68     tgen = get_topogen()
  69     cmds_list = [
  70         'ip link add vrf1 type vrf table 10',
  71         'echo 100000 > /proc/sys/net/mpls/platform_labels',
  72         'ip link set dev vrf1 up',
  73         'ip link set dev {0}-eth1 master vrf1',
  74         'echo 1 > /proc/sys/net/mpls/conf/{0}-eth0/input',
  75     ]
  76
  77     for cmd in cmds_list:
  78         input = cmd.format('r1', '1', '2')
  79         logger.info('input: ' + cmd)
  80         output = tgen.net['r1'].cmd(cmd.format('r1', '1', '2'))
  81         logger.info('output: ' + output)
  82
  83     for cmd in cmds_list:
  84         input = cmd.format('r2', '2', '1')
  85         logger.info('input: ' + cmd)
  86         output = tgen.net['r2'].cmd(cmd.format('r2', '2', '1'))
  87         logger.info('output: ' + output)
  88
  89 def setup_module(mod):
  90     "Sets up the pytest environment"
  91     tgen = Topogen(build_topo, mod.__name__)
  92     tgen.start_topology()
  93
  94     router_list = tgen.routers()
  95     _populate_iface()
  96
  97     for rname, router in router_list.items():
  98         router.load_config(
  99             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
 100         )
 101         router.load_config(
 102             TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format(rname))
 103         )
 104
 105     # Initialize all routers.
 106     tgen.start_router()
 107
 108
 109 def teardown_module(_mod):
 110     "Teardown the pytest environment"
 111     tgen = get_topogen()
 112
 113     tgen.stop_topology()
 114
 115
 116 def test_protocols_convergence():
 117     """
 118     Assert that all protocols have converged
 119     statuses as they depend on it.
 120     """
 121     tgen = get_topogen()
 122     if tgen.routers_have_failure():
 123         pytest.skip(tgen.errors)
 124
 125     router = tgen.gears['r1']
 126     logger.info("Dump some context for r1")
 127     router.vtysh_cmd("show bgp ipv4 vpn")
 128     router.vtysh_cmd("show bgp summary")
 129     router.vtysh_cmd("show bgp vrf vrf1 ipv4")
 130     router.vtysh_cmd("show running-config")
 131     router = tgen.gears['r2']
 132     logger.info("Dump some context for r2")
 133     router.vtysh_cmd("show bgp ipv4 vpn")
 134     router.vtysh_cmd("show bgp summary")
 135     router.vtysh_cmd("show bgp vrf vrf1 ipv4")
 136     router.vtysh_cmd("show running-config")
 137
 138     # Check IPv4 routing tables on r1
 139     logger.info("Checking IPv4 routes for convergence on r1")
 140     router = tgen.gears['r1']
 141     json_file = "{}/{}/ipv4_routes.json".format(CWD, router.name)
 142     if not os.path.isfile(json_file):
 143         logger.info("skipping file {}".format(json_file))
 144         assert 0, 'ipv4_routes.json file not found'
 145         return
 146
 147     expected = json.loads(open(json_file).read())
 148     test_func = partial(
 149         topotest.router_json_cmp,
 150         router,
 151         "show ip route vrf vrf1 json",
 152         expected,
 153     )
 154     _, result = topotest.run_and_expect(test_func, None, count=40, wait=2)
 155     assertmsg = '"{}" JSON output mismatches'.format(router.name)
 156     assert result is None, assertmsg
 157
 158     # Check BGP IPv4 routing tables on r2 not installed
 159     logger.info("Checking BGP IPv4 routes for convergence on r2")
 160     router = tgen.gears['r2']
 161     json_file = "{}/{}/bgp_ipv4_routes.json".format(CWD, router.name)
 162     if not os.path.isfile(json_file):
 163         assert 0, 'bgp_ipv4_routes.json file not found'
 164
 165     expected = json.loads(open(json_file).read())
 166     test_func = partial(
 167         topotest.router_json_cmp,
 168         router,
 169         "show bgp vrf vrf1 ipv4 json",
 170         expected,
 171     )
 172     _, result = topotest.run_and_expect(test_func, None, count=40, wait=2)
 173     assertmsg = '"{}" JSON output mismatches'.format(router.name)
 174     assert result is None, assertmsg
 175
 176 def test_memory_leak():
 177     "Run the memory leak test and report results."
 178     tgen = get_topogen()
 179     if not tgen.is_memleak_enabled():
 180         pytest.skip("Memory leak test/report is disabled")
 181
 182     tgen.report_memory_leaks()
 183
 184
 185 if __name__ == "__main__":
 186     args = ["-s"] + sys.argv[1:]
 187     sys.exit(pytest.main(args))