tests/topotests/bgp_srv6l3vpn_to_bgp_vrf/test_bgp_srv6l3vpn_to_bgp_vrf.py

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