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     output = router.vtysh_cmd(cmd)
 138     logger.info("{}: {}\n{}".format(router.name, cmd, output))
 139
 140     json_output = json.loads(output)
 141
 142     # filter out tableVersion, version and nhVrfID
 143     json_output.pop("tableVersion")
 144     for rd, data in json_output["routes"]["routeDistinguishers"].items():
 145         for prefix, attrs in data.items():
 146             for attr in attrs:
 147                 if "nhVrfId" in attr:
 148                     attr.pop("nhVrfId")
 149                 if "version" in attr:
 150                     attr.pop("version")
 151
 152     return topotest.json_cmp(json_output, expected, exact=True)
 153
 154
 155 def test_bgp_no_retain():
 156     """
 157     Check bgp no retain route-target all on r1
 158     """
 159
 160     tgen = get_topogen()
 161     if tgen.routers_have_failure():
 162         pytest.skip(tgen.errors)
 163
 164     # Check IPv4 VPN routing tables on r1
 165     logger.info("Checking VPNv4 routes for convergence on r1")
 166     router = tgen.gears["r1"]
 167     json_file = "{}/{}/ipv4_vpn_routes.json".format(CWD, router.name)
 168     if not os.path.isfile(json_file):
 169         logger.info("skipping file {}".format(json_file))
 170         assert 0, "{} file not found".format(json_file)
 171         return
 172
 173     expected = json.loads(open(json_file).read())
 174     test_func = partial(
 175         router_json_cmp_exact_filter,
 176         router,
 177         "show bgp ipv4 vpn json",
 178         expected,
 179     )
 180     _, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
 181     assertmsg = '"{}" JSON output mismatches'.format(router.name)
 182     assert result is None, assertmsg
 183
 184
 185 def test_bgp_retain():
 186     """
 187     Apply and check bgp retain route-target all on r1
 188     """
 189
 190     tgen = get_topogen()
 191     if tgen.routers_have_failure():
 192         pytest.skip(tgen.errors)
 193
 194     # Check IPv4 VPN routing tables on r1
 195     logger.info("Checking VPNv4 routes on r1 after bgp no retain")
 196     router = tgen.gears["r1"]
 197     router.vtysh_cmd(
 198         "configure\nrouter bgp 65500\naddress-family ipv4 vpn\nbgp retain route-target all\n"
 199     )
 200     json_file = "{}/{}/ipv4_vpn_routes_unfiltered.json".format(CWD, router.name)
 201     if not os.path.isfile(json_file):
 202         logger.info("skipping file {}".format(json_file))
 203         assert 0, "{} file not found".format(json_file)
 204         return
 205
 206     expected = json.loads(open(json_file).read())
 207     test_func = partial(
 208         router_json_cmp_exact_filter,
 209         router,
 210         "show bgp ipv4 vpn json",
 211         expected,
 212     )
 213     _, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
 214     assertmsg = '"{}" JSON output mismatches'.format(router.name)
 215     assert result is None, assertmsg
 216
 217
 218 def test_memory_leak():
 219     "Run the memory leak test and report results."
 220     tgen = get_topogen()
 221     if not tgen.is_memleak_enabled():
 222         pytest.skip("Memory leak test/report is disabled")
 223
 224     tgen.report_memory_leaks()
 225
 226
 227 if __name__ == "__main__":
 228     args = ["-s"] + sys.argv[1:]
 229     sys.exit(pytest.main(args))