tests/topotests/bgp_vpnv4_noretain/test_bgp_vpnv4_noretain.py

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