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 test_protocols_convergence():
 137     """
 138     Assert that all protocols have converged
 139     statuses as they depend on it.
 140     """
 141     tgen = get_topogen()
 142     if tgen.routers_have_failure():
 143         pytest.skip(tgen.errors)
 144
 145     # Check IPv4 VPN routing tables on r1
 146     logger.info("Checking IPv4 routes for convergence on r1")
 147     router = tgen.gears['r1']
 148     json_file = "{}/{}/ipv4_vpn_routes.json".format(CWD, router.name)
 149     if not os.path.isfile(json_file):
 150         logger.info("skipping file {}".format(json_file))
 151         assert 0, 'ipv4_vpn_routes.json file not found'
 152         return
 153
 154     expected = json.loads(open(json_file).read())
 155     test_func = partial(
 156         topotest.router_json_cmp,
 157         router,
 158         "show bgp ipv4 vpn json",
 159         expected,
 160     )
 161     _, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
 162     assertmsg = '"{}" JSON output mismatches'.format(router.name)
 163     assert result is None, assertmsg
 164
 165     # Check BGP IPv4 routing tables after unsetting no retain flag
 166     logger.info("Checking BGP IPv4 routes for convergence on r2")
 167     router = tgen.gears['r1']
 168     router.vtysh_cmd("configure\nrouter bgp 65500\naddress-family ipv4 vpn\nbgp retain route-target all\n")
 169
 170     # Check IPv4 VPN routing tables on r1
 171     logger.info("Checking IPv4 routes for convergence on r1")
 172     router = tgen.gears['r1']
 173     json_file = "{}/{}/ipv4_vpn_routes_unfiltered.json".format(CWD, router.name)
 174     if not os.path.isfile(json_file):
 175         logger.info("skipping file {}".format(json_file))
 176         assert 0, 'ipv4_vpn_routes_unfiltered.json file not found'
 177         return
 178
 179     expected = json.loads(open(json_file).read())
 180     test_func = partial(
 181         topotest.router_json_cmp,
 182         router,
 183         "show bgp ipv4 vpn json",
 184         expected,
 185     )
 186     _, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
 187     assertmsg = '"{}" JSON output mismatches'.format(router.name)
 188     assert result is None, assertmsg
 189
 190 def test_memory_leak():
 191     "Run the memory leak test and report results."
 192     tgen = get_topogen()
 193     if not tgen.is_memleak_enabled():
 194         pytest.skip("Memory leak test/report is disabled")
 195
 196     tgen.report_memory_leaks()
 197
 198
 199 if __name__ == "__main__":
 200     args = ["-s"] + sys.argv[1:]
 201     sys.exit(pytest.main(args))