tests/topotests/bgp_vpnv4_noretain/test_bgp_vpnv4_noretain.py

   1 #!/usr/bin/env python
   2
   3 #
   4 # test_bgp_vpnv4_noretain.py
   5 # Part of NetDEF Topology Tests
   6 #
   7 # Copyright 2022 6WIND S.A.
   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_noretain.py: Do not keep the VPNvx entries when no
  26  VRF matches incoming VPNVx entries
  27 """
  28
  29 import os
  30 import sys
  31 import json
  32 from functools import partial
  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
  47
  48 pytestmark = [pytest.mark.bgpd]
  49
  50 def build_topo(tgen):
  51     "Build function"
  52
  53     tgen.add_router("r1")
  54     tgen.add_router("r2")
  55
  56     switch = tgen.add_switch("s1")
  57     switch.add_link(tgen.gears["r1"])
  58     switch.add_link(tgen.gears["r2"])
  59
  60     switch = tgen.add_switch("s2")
  61     switch.add_link(tgen.gears["r1"])
  62
  63     switch = tgen.add_switch("s3")
  64     switch.add_link(tgen.gears["r2"])
  65
  66     switch = tgen.add_switch("s4")
  67     switch.add_link(tgen.gears["r2"])
  68
  69
  70 def _populate_iface():
  71     tgen = get_topogen()
  72     cmds_list = [
  73         'modprobe mpls_router',
  74         'echo 100000 > /proc/sys/net/mpls/platform_labels',
  75         'ip link add vrf1 type vrf table 10',
  76         'ip link set dev vrf1 up',
  77         'ip link set dev {0}-eth1 master vrf1',
  78         'echo 1 > /proc/sys/net/mpls/conf/vrf1/input',
  79     ]
  80     cmds_list_extra = [
  81         'ip link add vrf2 type vrf table 20',
  82         'ip link set dev vrf2 up',
  83         'ip link set dev {0}-eth2 master vrf2',
  84         'echo 1 > /proc/sys/net/mpls/conf/vrf2/input',
  85     ]
  86
  87     for cmd in cmds_list:
  88         input = cmd.format('r1', '1', '2')
  89         logger.info('input: ' + cmd)
  90         output = tgen.net['r1'].cmd(cmd.format('r1', '1', '2'))
  91         logger.info('output: ' + output)
  92
  93     for cmd in cmds_list:
  94         input = cmd.format('r2', '2', '1')
  95         logger.info('input: ' + cmd)
  96         output = tgen.net['r2'].cmd(cmd.format('r2', '2', '1'))
  97         logger.info('output: ' + output)
  98
  99     for cmd in cmds_list_extra:
 100         input = cmd.format('r2', '2', '1')
 101         logger.info('input: ' + cmd)
 102         output = tgen.net['r2'].cmd(cmd.format('r2', '2', '1'))
 103         logger.info('output: ' + output)
 104
 105 def setup_module(mod):
 106     "Sets up the pytest environment"
 107
 108     tgen = Topogen(build_topo, mod.__name__)
 109     tgen.start_topology()
 110
 111     router_list = tgen.routers()
 112     _populate_iface()
 113
 114     for rname, router in router_list.items():
 115         router.load_config(
 116             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
 117         )
 118         router.load_config(
 119             TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format(rname))
 120         )
 121         router.load_config(
 122             TopoRouter.RD_ISIS, os.path.join(CWD, "{}/bgpd.conf".format(rname))
 123         )
 124
 125     # Initialize all routers.
 126     tgen.start_router()
 127
 128
 129 def teardown_module(_mod):
 130     "Teardown the pytest environment"
 131     tgen = get_topogen()
 132
 133     tgen.stop_topology()
 134
 135
 136 def router_json_cmp_exact_filter(router, cmd, expected):
 137     # filter out tableVersion, version and nhVrfID
 138     output = router.cmd('vtysh -c "{}" | grep -v ersion | grep -v nhVrfId'.format(cmd))
 139     logger.info("{}: {}\n{}".format(router.name, cmd, output))
 140
 141     json_output = json.loads(output)
 142
 143     return topotest.json_cmp(json_output, expected, exact=True)
 144
 145
 146 def test_bgp_no_retain():
 147     """
 148     Check bgp no retain route-target all on r1
 149     """
 150
 151     tgen = get_topogen()
 152     if tgen.routers_have_failure():
 153         pytest.skip(tgen.errors)
 154
 155     # Check IPv4 VPN routing tables on r1
 156     logger.info("Checking VPNv4 routes for convergence on r1")
 157     router = tgen.gears["r1"]
 158     json_file = "{}/{}/ipv4_vpn_routes.json".format(CWD, router.name)
 159     if not os.path.isfile(json_file):
 160         logger.info("skipping file {}".format(json_file))
 161         assert 0, "{} file not found".format(json_file)
 162         return
 163
 164     expected = json.loads(open(json_file).read())
 165     test_func = partial(
 166         router_json_cmp_exact_filter,
 167         router,
 168         "show bgp ipv4 vpn json",
 169         expected,
 170     )
 171     _, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
 172     assertmsg = '"{}" JSON output mismatches'.format(router.name)
 173     assert result is None, assertmsg
 174
 175
 176 def test_bgp_retain():
 177     """
 178     Apply and check bgp retain route-target all on r1
 179     """
 180
 181     tgen = get_topogen()
 182     if tgen.routers_have_failure():
 183         pytest.skip(tgen.errors)
 184
 185     # Check IPv4 VPN routing tables on r1
 186     logger.info("Checking VPNv4 routes on r1 after bgp no retain")
 187     router = tgen.gears["r1"]
 188     router.vtysh_cmd(
 189         "configure\nrouter bgp 65500\naddress-family ipv4 vpn\nbgp retain route-target all\n"
 190     )
 191     json_file = "{}/{}/ipv4_vpn_routes_unfiltered.json".format(CWD, router.name)
 192     if not os.path.isfile(json_file):
 193         logger.info("skipping file {}".format(json_file))
 194         assert 0, "{} file not found".format(json_file)
 195         return
 196
 197     expected = json.loads(open(json_file).read())
 198     test_func = partial(
 199         router_json_cmp_exact_filter,
 200         router,
 201         "show bgp ipv4 vpn json",
 202         expected,
 203     )
 204     _, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
 205     assertmsg = '"{}" JSON output mismatches'.format(router.name)
 206     assert result is None, assertmsg
 207
 208
 209 def test_memory_leak():
 210     "Run the memory leak test and report results."
 211     tgen = get_topogen()
 212     if not tgen.is_memleak_enabled():
 213         pytest.skip("Memory leak test/report is disabled")
 214
 215     tgen.report_memory_leaks()
 216
 217
 218 if __name__ == "__main__":
 219     args = ["-s"] + sys.argv[1:]
 220     sys.exit(pytest.main(args))