tests/topotests/bgp_srv6l3vpn_to_bgp_vrf/test_bgp_srv6l3vpn_to_bgp_vrf.py

   1 #!/usr/bin/env python
   2
   3 #
   4 # Part of NetDEF Topology Tests
   5 #
   6 # Copyright (c) 2018, LabN Consulting, L.L.C.
   7 # Authored by Lou Berger <lberger@labn.net>
   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 import os
  25 import sys
  26 import json
  27 import functools
  28 import pytest
  29
  30 CWD = os.path.dirname(os.path.realpath(__file__))
  31 sys.path.append(os.path.join(CWD, "../"))
  32
  33 # pylint: disable=C0413
  34 # Import topogen and topotest helpers
  35 from lib import topotest
  36 from lib.topogen import Topogen, TopoRouter, get_topogen
  37 from lib.topolog import logger
  38 from lib.common_config import required_linux_kernel_version
  39
  40 pytestmark = [pytest.mark.bgpd]
  41
  42
  43 def build_topo(tgen):
  44     r"""
  45       CE1     CE3      CE5
  46     (eth0)  (eth0)   (eth0)
  47       :2      :2      :2
  48        |       |       |
  49      2001:   2001:   2001:
  50      1::/64  3::/64  5::/64
  51        |       |       |
  52       :1      :1      :1
  53    +-(eth1)--(eth2)---(eth3)-+
  54    |     \   /          |    |
  55    |    (vrf10)     (vrf20)  |
  56    |             R1          |
  57    +----------(eth0)---------+
  58                 :1
  59                 |
  60             2001::/64
  61                 |
  62                 :2
  63               (eth0)
  64     +----------(eth0)--------------+
  65     |            R2                |
  66     |   (vrf10)       (vrf20)      |
  67     |     /           /     \      |
  68     +-(eth1)-----(eth2)-----(eth3)-+
  69         :1         :1          :1
  70          |          |           |
  71       +------+   +------+   +------+
  72      /  2001: \ /  2001: \ /  2001: \
  73      \ 2::/64 / \ 4::/64 / \ 6::/64 /
  74       +------+   +------+   +------+
  75          |           |          |
  76         :2          :2         :2
  77       (eth0)      (eth0)      (eth0)
  78         CE2         CE4         CE6
  79     """
  80     tgen.add_router("r1")
  81     tgen.add_router("r2")
  82     tgen.add_router("ce1")
  83     tgen.add_router("ce2")
  84     tgen.add_router("ce3")
  85     tgen.add_router("ce4")
  86     tgen.add_router("ce5")
  87     tgen.add_router("ce6")
  88
  89     tgen.add_link(tgen.gears["r1"], tgen.gears["r2"], "eth0", "eth0")
  90     tgen.add_link(tgen.gears["ce1"], tgen.gears["r1"], "eth0", "eth1")
  91     tgen.add_link(tgen.gears["ce2"], tgen.gears["r2"], "eth0", "eth1")
  92     tgen.add_link(tgen.gears["ce3"], tgen.gears["r1"], "eth0", "eth2")
  93     tgen.add_link(tgen.gears["ce4"], tgen.gears["r2"], "eth0", "eth2")
  94     tgen.add_link(tgen.gears["ce5"], tgen.gears["r1"], "eth0", "eth3")
  95     tgen.add_link(tgen.gears["ce6"], tgen.gears["r2"], "eth0", "eth3")
  96
  97
  98 def setup_module(mod):
  99     result = required_linux_kernel_version("4.15")
 100     if result is not True:
 101         pytest.skip("Kernel requirements are not met")
 102
 103     tgen = Topogen(build_topo, mod.__name__)
 104     tgen.start_topology()
 105     router_list = tgen.routers()
 106     for rname, router in tgen.routers().items():
 107         router.run("/bin/bash {}/{}/setup.sh".format(CWD, rname))
 108         router.load_config(
 109             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
 110         )
 111         router.load_config(
 112             TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format(rname))
 113         )
 114
 115     tgen.gears["r1"].run("ip link add vrf10 type vrf table 10")
 116     tgen.gears["r1"].run("ip link set vrf10 up")
 117     tgen.gears["r1"].run("ip link add vrf20 type vrf table 20")
 118     tgen.gears["r1"].run("ip link set vrf20 up")
 119     tgen.gears["r1"].run("ip link set eth1 master vrf10")
 120     tgen.gears["r1"].run("ip link set eth2 master vrf10")
 121     tgen.gears["r1"].run("ip link set eth3 master vrf20")
 122
 123     tgen.gears["r2"].run("ip link add vrf10 type vrf table 10")
 124     tgen.gears["r2"].run("ip link set vrf10 up")
 125     tgen.gears["r2"].run("ip link add vrf20 type vrf table 20")
 126     tgen.gears["r2"].run("ip link set vrf20 up")
 127     tgen.gears["r2"].run("ip link set eth1 master vrf10")
 128     tgen.gears["r2"].run("ip link set eth2 master vrf20")
 129     tgen.gears["r2"].run("ip link set eth3 master vrf20")
 130     tgen.start_router()
 131
 132     # FOR DEVELOPER:
 133     # If you want to stop some specific line and start interactive shell,
 134     # please use tgen.mininet_cli() to start it.
 135
 136
 137 def teardown_module(mod):
 138     tgen = get_topogen()
 139     tgen.stop_topology()
 140
 141
 142 def open_json_file(filename):
 143     try:
 144         with open(filename, "r") as f:
 145             return json.load(f)
 146     except IOError:
 147         assert False, "Could not read file {}".format(filename)
 148
 149
 150 def check_ping(name, dest_addr, expect_connected):
 151     def _check(name, dest_addr, match):
 152         tgen = get_topogen()
 153         output = tgen.gears[name].run("ping6 {} -c 1 -w 1".format(dest_addr))
 154         logger.info(output)
 155         if match not in output:
 156             return "ping fail"
 157
 158     match = ", {} packet loss".format("0%" if expect_connected else "100%")
 159     logger.info("[+] check {} {} {}".format(name, dest_addr, match))
 160     tgen = get_topogen()
 161     func = functools.partial(_check, name, dest_addr, match)
 162     success, result = topotest.run_and_expect(func, None, count=10, wait=1)
 163     assert result is None, "Failed"
 164
 165
 166 def check_rib(name, cmd, expected_file):
 167     def _check(name, cmd, expected_file):
 168         logger.info("polling")
 169         tgen = get_topogen()
 170         router = tgen.gears[name]
 171         output = json.loads(router.vtysh_cmd(cmd))
 172         expected = open_json_file("{}/{}".format(CWD, expected_file))
 173         return topotest.json_cmp(output, expected)
 174
 175     logger.info('[+] check {} "{}" {}'.format(name, cmd, expected_file))
 176     tgen = get_topogen()
 177     func = functools.partial(_check, name, cmd, expected_file)
 178     success, result = topotest.run_and_expect(func, None, count=10, wait=0.5)
 179     assert result is None, "Failed"
 180
 181
 182 def test_rib():
 183     check_rib("r1", "show bgp ipv6 vpn json", "r1/vpnv6_rib.json")
 184     check_rib("r2", "show bgp ipv6 vpn json", "r2/vpnv6_rib.json")
 185     check_rib("r1", "show ipv6 route vrf vrf10 json", "r1/vrf10_rib.json")
 186     check_rib("r1", "show ipv6 route vrf vrf20 json", "r1/vrf20_rib.json")
 187     check_rib("r2", "show ipv6 route vrf vrf10 json", "r2/vrf10_rib.json")
 188     check_rib("r2", "show ipv6 route vrf vrf20 json", "r2/vrf20_rib.json")
 189     check_rib("ce1", "show ipv6 route json", "ce1/ipv6_rib.json")
 190     check_rib("ce2", "show ipv6 route json", "ce2/ipv6_rib.json")
 191     check_rib("ce3", "show ipv6 route json", "ce3/ipv6_rib.json")
 192     check_rib("ce4", "show ipv6 route json", "ce4/ipv6_rib.json")
 193     check_rib("ce5", "show ipv6 route json", "ce5/ipv6_rib.json")
 194     check_rib("ce6", "show ipv6 route json", "ce6/ipv6_rib.json")
 195
 196
 197 def test_ping():
 198     check_ping("ce1", "2001:2::2", True)
 199     check_ping("ce1", "2001:3::2", True)
 200     check_ping("ce1", "2001:4::2", False)
 201     check_ping("ce1", "2001:5::2", False)
 202     check_ping("ce1", "2001:6::2", False)
 203     check_ping("ce4", "2001:1::2", False)
 204     check_ping("ce4", "2001:2::2", False)
 205     check_ping("ce4", "2001:3::2", False)
 206     check_ping("ce4", "2001:5::2", True)
 207     check_ping("ce4", "2001:6::2", True)
 208
 209
 210 def test_locator_delete():
 211     check_ping("ce1", "2001:2::2", True)
 212     get_topogen().gears["r1"].vtysh_cmd(
 213         """
 214         configure terminal
 215          segment-routing
 216           srv6
 217            locators
 218             no locator loc1
 219         """
 220     )
 221     check_rib("r1", "show bgp ipv6 vpn json", "r1/vpnv6_rib_locator_deleted.json")
 222     check_rib("r2", "show bgp ipv6 vpn json", "r2/vpnv6_rib_locator_deleted.json")
 223     check_ping("ce1", "2001:2::2", False)
 224
 225
 226 def test_locator_recreate():
 227     check_ping("ce1", "2001:2::2", False)
 228     get_topogen().gears["r1"].vtysh_cmd(
 229         """
 230         configure terminal
 231          segment-routing
 232           srv6
 233            locators
 234             locator loc1
 235              prefix 2001:db8:1:1::/64
 236         """
 237     )
 238     check_rib("r1", "show bgp ipv6 vpn json", "r1/vpnv6_rib_locator_recreated.json")
 239     check_rib("r2", "show bgp ipv6 vpn json", "r2/vpnv6_rib_locator_recreated.json")
 240     check_ping("ce1", "2001:2::2", True)
 241
 242
 243 def test_bgp_locator_unset():
 244     check_ping("ce1", "2001:2::2", True)
 245     get_topogen().gears["r1"].vtysh_cmd(
 246         """
 247         configure terminal
 248          router bgp 1
 249           segment-routing srv6
 250            no locator loc1
 251         """
 252     )
 253     check_rib("r1", "show bgp ipv6 vpn json", "r1/vpnv6_rib_locator_deleted.json")
 254     check_rib("r2", "show bgp ipv6 vpn json", "r2/vpnv6_rib_locator_deleted.json")
 255     check_ping("ce1", "2001:2::2", False)
 256
 257
 258 def test_bgp_locator_reset():
 259     check_ping("ce1", "2001:2::2", False)
 260     get_topogen().gears["r1"].vtysh_cmd(
 261         """
 262         configure terminal
 263          router bgp 1
 264           segment-routing srv6
 265            locator loc1
 266         """
 267     )
 268     check_rib("r1", "show bgp ipv6 vpn json", "r1/vpnv6_rib_locator_recreated.json")
 269     check_rib("r2", "show bgp ipv6 vpn json", "r2/vpnv6_rib_locator_recreated.json")
 270     check_ping("ce1", "2001:2::2", True)
 271
 272
 273 def test_bgp_srv6_unset():
 274     check_ping("ce1", "2001:2::2", True)
 275     get_topogen().gears["r1"].vtysh_cmd(
 276         """
 277         configure terminal
 278          router bgp 1
 279           no segment-routing srv6
 280         """
 281     )
 282     check_rib("r1", "show bgp ipv6 vpn json", "r1/vpnv6_rib_locator_deleted.json")
 283     check_rib("r2", "show bgp ipv6 vpn json", "r2/vpnv6_rib_locator_deleted.json")
 284     check_ping("ce1", "2001:2::2", False)
 285
 286
 287 def test_bgp_srv6_reset():
 288     check_ping("ce1", "2001:2::2", False)
 289     get_topogen().gears["r1"].vtysh_cmd(
 290         """
 291         configure terminal
 292          router bgp 1
 293           segment-routing srv6
 294            locator loc1
 295         """
 296     )
 297     check_rib("r1", "show bgp ipv6 vpn json", "r1/vpnv6_rib_locator_recreated.json")
 298     check_rib("r2", "show bgp ipv6 vpn json", "r2/vpnv6_rib_locator_recreated.json")
 299     check_ping("ce1", "2001:2::2", True)
 300
 301
 302 if __name__ == "__main__":
 303     args = ["-s"] + sys.argv[1:]
 304     sys.exit(pytest.main(args))