tests/topotests/bgp_srv6l3vpn_to_bgp_vrf3/test_bgp_srv6l3vpn_to_bgp_vrf3.py

   1 #!/usr/bin/env python
   2
   3 # Copyright (c) 2022, University of Rome Tor Vergata
   4 # Authored by Carmine Scarpitta <carmine.scarpitta@uniroma2.it>
   5 #
   6 # Permission to use, copy, modify, and/or distribute this software
   7 # for any purpose with or without fee is hereby granted, provided
   8 # that the above copyright notice and this permission notice appear
   9 # in all copies.
  10 #
  11 # THE SOFTWARE IS PROVIDED "AS IS" AND NETDEF DISCLAIMS ALL WARRANTIES
  12 # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
  13 # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NETDEF BE LIABLE FOR
  14 # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
  15 # DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
  16 # WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
  17 # ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
  18 # OF THIS SOFTWARE.
  19 #
  20
  21 import os
  22 import re
  23 import sys
  24 import json
  25 import functools
  26 import pytest
  27
  28 CWD = os.path.dirname(os.path.realpath(__file__))
  29 sys.path.append(os.path.join(CWD, "../"))
  30
  31 # pylint: disable=C0413
  32 # Import topogen and topotest helpers
  33 from lib import topotest
  34 from lib.topogen import Topogen, TopoRouter, get_topogen
  35 from lib.topolog import logger
  36 from lib.common_config import required_linux_kernel_version
  37
  38 pytestmark = [pytest.mark.bgpd]
  39
  40
  41 def build_topo(tgen):
  42     tgen.add_router("r1")
  43     tgen.add_router("r2")
  44     tgen.add_router("ce1")
  45     tgen.add_router("ce2")
  46     tgen.add_router("ce3")
  47     tgen.add_router("ce4")
  48     tgen.add_router("ce5")
  49     tgen.add_router("ce6")
  50
  51     tgen.add_link(tgen.gears["r1"], tgen.gears["r2"], "eth0", "eth0")
  52     tgen.add_link(tgen.gears["ce1"], tgen.gears["r1"], "eth0", "eth1")
  53     tgen.add_link(tgen.gears["ce2"], tgen.gears["r2"], "eth0", "eth1")
  54     tgen.add_link(tgen.gears["ce3"], tgen.gears["r1"], "eth0", "eth2")
  55     tgen.add_link(tgen.gears["ce4"], tgen.gears["r2"], "eth0", "eth2")
  56     tgen.add_link(tgen.gears["ce5"], tgen.gears["r1"], "eth0", "eth3")
  57     tgen.add_link(tgen.gears["ce6"], tgen.gears["r2"], "eth0", "eth3")
  58
  59
  60 def setup_module(mod):
  61     result = required_linux_kernel_version("5.14")
  62     if result is not True:
  63         pytest.skip("Kernel requirements are not met")
  64
  65     tgen = Topogen(build_topo, mod.__name__)
  66     tgen.start_topology()
  67     for rname, router in tgen.routers().items():
  68         router.run("/bin/bash {}/{}/setup.sh".format(CWD, rname))
  69         router.load_config(
  70             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
  71         )
  72         router.load_config(
  73             TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format(rname))
  74         )
  75
  76     tgen.gears["r1"].run("sysctl net.vrf.strict_mode=1")
  77     tgen.gears["r1"].run("ip link add vrf10 type vrf table 10")
  78     tgen.gears["r1"].run("ip link set vrf10 up")
  79     tgen.gears["r1"].run("ip link add vrf20 type vrf table 20")
  80     tgen.gears["r1"].run("ip link set vrf20 up")
  81     tgen.gears["r1"].run("ip link set eth1 master vrf10")
  82     tgen.gears["r1"].run("ip link set eth2 master vrf10")
  83     tgen.gears["r1"].run("ip link set eth3 master vrf20")
  84
  85     tgen.gears["r2"].run("sysctl net.vrf.strict_mode=1")
  86     tgen.gears["r2"].run("ip link add vrf10 type vrf table 10")
  87     tgen.gears["r2"].run("ip link set vrf10 up")
  88     tgen.gears["r2"].run("ip link add vrf20 type vrf table 20")
  89     tgen.gears["r2"].run("ip link set vrf20 up")
  90     tgen.gears["r2"].run("ip link set eth1 master vrf10")
  91     tgen.gears["r2"].run("ip link set eth2 master vrf20")
  92     tgen.gears["r2"].run("ip link set eth3 master vrf20")
  93     tgen.start_router()
  94
  95
  96 def teardown_module(mod):
  97     tgen = get_topogen()
  98     tgen.stop_topology()
  99
 100
 101 def open_json_file(filename):
 102     try:
 103         with open(filename, "r") as f:
 104             return json.load(f)
 105     except IOError:
 106         assert False, "Could not read file {}".format(filename)
 107
 108
 109 def check_ping4(name, dest_addr, expect_connected):
 110     def _check(name, dest_addr, match):
 111         tgen = get_topogen()
 112         output = tgen.gears[name].run("ping {} -c 1 -w 1".format(dest_addr))
 113         logger.info(output)
 114         assert match in output, "ping fail"
 115
 116     match = ", {} packet loss".format("0%" if expect_connected else "100%")
 117     logger.info("[+] check {} {} {}".format(name, dest_addr, match))
 118     tgen = get_topogen()
 119     func = functools.partial(_check, name, dest_addr, match)
 120     success, result = topotest.run_and_expect(func, None, count=10, wait=0.5)
 121     assert result is None, "Failed"
 122
 123
 124 def check_ping6(name, dest_addr, expect_connected):
 125     def _check(name, dest_addr, match):
 126         tgen = get_topogen()
 127         output = tgen.gears[name].run("ping6 {} -c 1 -w 1".format(dest_addr))
 128         logger.info(output)
 129         if match not in output:
 130             return "ping fail"
 131
 132     match = "{} packet loss".format("0%" if expect_connected else "100%")
 133     logger.info("[+] check {} {} {}".format(name, dest_addr, match))
 134     tgen = get_topogen()
 135     func = functools.partial(_check, name, dest_addr, match)
 136     success, result = topotest.run_and_expect(func, None, count=10, wait=1)
 137     assert result is None, "Failed"
 138
 139
 140 def check_rib(name, cmd, expected_file):
 141     def _check(name, dest_addr, match):
 142         logger.info("polling")
 143         tgen = get_topogen()
 144         router = tgen.gears[name]
 145         output = json.loads(router.vtysh_cmd(cmd))
 146         expected = open_json_file("{}/{}".format(CWD, expected_file))
 147         return topotest.json_cmp(output, expected)
 148
 149     logger.info('[+] check {} "{}" {}'.format(name, cmd, expected_file))
 150     tgen = get_topogen()
 151     func = functools.partial(_check, name, cmd, expected_file)
 152     success, result = topotest.run_and_expect(func, None, count=10, wait=0.5)
 153     assert result is None, "Failed"
 154
 155
 156 def test_rib():
 157     check_rib("r1", "show bgp ipv4 vpn json", "r1/vpnv4_rib.json")
 158     check_rib("r2", "show bgp ipv4 vpn json", "r2/vpnv4_rib.json")
 159     check_rib("r1", "show ip route vrf vrf10 json", "r1/vrf10v4_rib.json")
 160     check_rib("r1", "show ip route vrf vrf20 json", "r1/vrf20v4_rib.json")
 161     check_rib("r2", "show ip route vrf vrf10 json", "r2/vrf10v4_rib.json")
 162     check_rib("r2", "show ip route vrf vrf20 json", "r2/vrf20v4_rib.json")
 163     check_rib("ce1", "show ip route json", "ce1/ip_rib.json")
 164     check_rib("ce2", "show ip route json", "ce2/ip_rib.json")
 165     check_rib("ce3", "show ip route json", "ce3/ip_rib.json")
 166     check_rib("ce4", "show ip route json", "ce4/ip_rib.json")
 167     check_rib("ce5", "show ip route json", "ce5/ip_rib.json")
 168     check_rib("ce6", "show ip route json", "ce6/ip_rib.json")
 169
 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/vrf10v6_rib.json")
 173     check_rib("r1", "show ipv6 route vrf vrf20 json", "r1/vrf20v6_rib.json")
 174     check_rib("r2", "show ipv6 route vrf vrf10 json", "r2/vrf10v6_rib.json")
 175     check_rib("r2", "show ipv6 route vrf vrf20 json", "r2/vrf20v6_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_ping4("ce1", "192.168.2.2", True)
 186     check_ping4("ce1", "192.168.3.2", True)
 187     check_ping4("ce1", "192.168.4.2", False)
 188     check_ping4("ce1", "192.168.5.2", False)
 189     check_ping4("ce1", "192.168.6.2", False)
 190     check_ping4("ce4", "192.168.1.2", False)
 191     check_ping4("ce4", "192.168.2.2", False)
 192     check_ping4("ce4", "192.168.3.2", False)
 193     check_ping4("ce4", "192.168.5.2", True)
 194     check_ping4("ce4", "192.168.6.2", True)
 195
 196     check_ping6("ce1", "2001:2::2", True)
 197     check_ping6("ce1", "2001:3::2", True)
 198     check_ping6("ce1", "2001:4::2", False)
 199     check_ping6("ce1", "2001:5::2", False)
 200     check_ping6("ce1", "2001:6::2", False)
 201     check_ping6("ce4", "2001:1::2", False)
 202     check_ping6("ce4", "2001:2::2", False)
 203     check_ping6("ce4", "2001:3::2", False)
 204     check_ping6("ce4", "2001:5::2", True)
 205     check_ping6("ce4", "2001:6::2", True)
 206
 207
 208 def test_bgp_sid_vpn_export_disable():
 209     check_ping4("ce1", "192.168.2.2", True)
 210     check_ping6("ce1", "2001:2::2", True)
 211     get_topogen().gears["r1"].vtysh_cmd(
 212         """
 213         configure terminal
 214          router bgp 1 vrf vrf10
 215           segment-routing srv6
 216            no sid vpn per-vrf export
 217         """
 218     )
 219     check_rib(
 220         "r1", "show bgp ipv4 vpn json", "r1/vpnv4_rib_sid_vpn_export_disabled.json"
 221     )
 222     check_rib(
 223         "r2", "show bgp ipv4 vpn json", "r2/vpnv4_rib_sid_vpn_export_disabled.json"
 224     )
 225     check_rib(
 226         "r1", "show bgp ipv6 vpn json", "r1/vpnv6_rib_sid_vpn_export_disabled.json"
 227     )
 228     check_rib(
 229         "r2", "show bgp ipv6 vpn json", "r2/vpnv6_rib_sid_vpn_export_disabled.json"
 230     )
 231     check_ping4("ce1", "192.168.2.2", False)
 232     check_ping6("ce1", "2001:2::2", False)
 233
 234
 235 def test_bgp_sid_vpn_export_reenable():
 236     check_ping4("ce1", "192.168.2.2", False)
 237     check_ping6("ce1", "2001:2::2", False)
 238     get_topogen().gears["r1"].vtysh_cmd(
 239         """
 240         configure terminal
 241          router bgp 1 vrf vrf10
 242           segment-routing srv6
 243            sid vpn per-vrf export auto
 244         """
 245     )
 246     check_rib(
 247         "r1", "show bgp ipv4 vpn json", "r1/vpnv4_rib_sid_vpn_export_reenabled.json"
 248     )
 249     check_rib(
 250         "r2", "show bgp ipv4 vpn json", "r2/vpnv4_rib_sid_vpn_export_reenabled.json"
 251     )
 252     check_rib(
 253         "r1", "show bgp ipv6 vpn json", "r1/vpnv6_rib_sid_vpn_export_reenabled.json"
 254     )
 255     check_rib(
 256         "r2", "show bgp ipv6 vpn json", "r2/vpnv6_rib_sid_vpn_export_reenabled.json"
 257     )
 258     check_ping4("ce1", "192.168.2.2", True)
 259     check_ping6("ce1", "2001:2::2", True)
 260
 261
 262 def test_locator_delete():
 263     check_ping4("ce1", "192.168.2.2", True)
 264     check_ping6("ce1", "2001:2::2", True)
 265     get_topogen().gears["r1"].vtysh_cmd(
 266         """
 267         configure terminal
 268          segment-routing
 269           srv6
 270            locators
 271             no locator loc1
 272         """
 273     )
 274     check_rib("r1", "show bgp ipv4 vpn json", "r1/vpnv4_rib_locator_deleted.json")
 275     check_rib("r2", "show bgp ipv4 vpn json", "r2/vpnv4_rib_locator_deleted.json")
 276     check_rib("r1", "show bgp ipv6 vpn json", "r1/vpnv6_rib_locator_deleted.json")
 277     check_rib("r2", "show bgp ipv6 vpn json", "r2/vpnv6_rib_locator_deleted.json")
 278     check_ping4("ce1", "192.168.2.2", False)
 279     check_ping6("ce1", "2001:2::2", False)
 280
 281
 282 def test_locator_recreate():
 283     check_ping4("ce1", "192.168.2.2", False)
 284     check_ping6("ce1", "2001:2::2", False)
 285     get_topogen().gears["r1"].vtysh_cmd(
 286         """
 287         configure terminal
 288          segment-routing
 289           srv6
 290            locators
 291             locator loc1
 292              prefix 2001:db8:1:1::/64
 293         """
 294     )
 295     check_rib("r1", "show bgp ipv4 vpn json", "r1/vpnv4_rib_locator_recreated.json")
 296     check_rib("r2", "show bgp ipv4 vpn json", "r2/vpnv4_rib_locator_recreated.json")
 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_ping4("ce1", "192.168.2.2", True)
 300     check_ping6("ce1", "2001:2::2", True)
 301
 302
 303 def test_bgp_locator_unset():
 304     check_ping4("ce1", "192.168.2.2", True)
 305     check_ping6("ce1", "2001:2::2", True)
 306     get_topogen().gears["r1"].vtysh_cmd(
 307         """
 308         configure terminal
 309          router bgp 1
 310           segment-routing srv6
 311            no locator loc1
 312         """
 313     )
 314     check_rib("r1", "show bgp ipv4 vpn json", "r1/vpnv4_rib_locator_deleted.json")
 315     check_rib("r2", "show bgp ipv4 vpn json", "r2/vpnv4_rib_locator_deleted.json")
 316     check_rib("r1", "show bgp ipv6 vpn json", "r1/vpnv6_rib_locator_deleted.json")
 317     check_rib("r2", "show bgp ipv6 vpn json", "r2/vpnv6_rib_locator_deleted.json")
 318     check_ping4("ce1", "192.168.2.2", False)
 319     check_ping6("ce1", "2001:2::2", False)
 320
 321
 322 def test_bgp_locator_reset():
 323     check_ping4("ce1", "192.168.2.2", False)
 324     check_ping6("ce1", "2001:2::2", False)
 325     get_topogen().gears["r1"].vtysh_cmd(
 326         """
 327         configure terminal
 328          router bgp 1
 329           segment-routing srv6
 330            locator loc1
 331         """
 332     )
 333     check_rib("r1", "show bgp ipv4 vpn json", "r1/vpnv4_rib_locator_recreated.json")
 334     check_rib("r2", "show bgp ipv4 vpn json", "r2/vpnv4_rib_locator_recreated.json")
 335     check_rib("r1", "show bgp ipv6 vpn json", "r1/vpnv6_rib_locator_recreated.json")
 336     check_rib("r2", "show bgp ipv6 vpn json", "r2/vpnv6_rib_locator_recreated.json")
 337     check_ping4("ce1", "192.168.2.2", True)
 338     check_ping6("ce1", "2001:2::2", True)
 339
 340
 341 def test_bgp_srv6_unset():
 342     check_ping4("ce1", "192.168.2.2", True)
 343     check_ping6("ce1", "2001:2::2", True)
 344     get_topogen().gears["r1"].vtysh_cmd(
 345         """
 346         configure terminal
 347          router bgp 1
 348           no segment-routing srv6
 349         """
 350     )
 351     check_rib("r1", "show bgp ipv4 vpn json", "r1/vpnv4_rib_locator_deleted.json")
 352     check_rib("r2", "show bgp ipv4 vpn json", "r2/vpnv4_rib_locator_deleted.json")
 353     check_rib("r1", "show bgp ipv6 vpn json", "r1/vpnv6_rib_locator_deleted.json")
 354     check_rib("r2", "show bgp ipv6 vpn json", "r2/vpnv6_rib_locator_deleted.json")
 355     check_ping4("ce1", "192.168.2.2", False)
 356     check_ping6("ce1", "2001:2::2", False)
 357
 358
 359 def test_bgp_srv6_reset():
 360     check_ping4("ce1", "192.168.2.2", False)
 361     check_ping6("ce1", "2001:2::2", False)
 362     get_topogen().gears["r1"].vtysh_cmd(
 363         """
 364         configure terminal
 365          router bgp 1
 366           segment-routing srv6
 367            locator loc1
 368         """
 369     )
 370     check_rib("r1", "show bgp ipv4 vpn json", "r1/vpnv4_rib_locator_recreated.json")
 371     check_rib("r2", "show bgp ipv4 vpn json", "r2/vpnv4_rib_locator_recreated.json")
 372     check_rib("r1", "show bgp ipv6 vpn json", "r1/vpnv6_rib_locator_recreated.json")
 373     check_rib("r2", "show bgp ipv6 vpn json", "r2/vpnv6_rib_locator_recreated.json")
 374     check_ping4("ce1", "192.168.2.2", True)
 375     check_ping6("ce1", "2001:2::2", True)
 376
 377
 378 if __name__ == "__main__":
 379     args = ["-s"] + sys.argv[1:]
 380     sys.exit(pytest.main(args))