tests/topotests/bgp_srv6l3vpn_route_leak/test_bgp_srv6l3vpn_route_leak.py

   1 #!/usr/bin/env python
   2
   3 # Copyright (c) 2022, LINE Corporation
   4 # Authored by Ryoga Saito <ryoga.saito@linecorp.com>
   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("pe1")
  43     tgen.add_router("ce1")
  44
  45     tgen.add_link(tgen.gears["pe1"], tgen.gears["ce1"], "eth0", "eth0")
  46
  47
  48 def setup_module(mod):
  49     tgen = Topogen(build_topo, mod.__name__)
  50     tgen.start_topology()
  51
  52     for rname, router in tgen.routers().items():
  53         router.load_config(TopoRouter.RD_ZEBRA,
  54                            os.path.join(CWD, '{}/zebra.conf'.format(rname)))
  55         router.load_config(TopoRouter.RD_BGP,
  56                            os.path.join(CWD, '{}/bgpd.conf'.format(rname)))
  57
  58     tgen.gears["pe1"].run("ip link add vrf10 type vrf table 10")
  59     tgen.gears["pe1"].run("ip link set vrf10 up")
  60     tgen.gears["pe1"].run("ip link add vrf20 type vrf table 20")
  61     tgen.gears["pe1"].run("ip link set vrf20 up")
  62     tgen.gears["pe1"].run("ip link set eth0 master vrf10")
  63
  64     tgen.start_router()
  65
  66
  67 def teardown_module(mod):
  68     tgen = get_topogen()
  69     tgen.stop_topology()
  70
  71
  72 def open_json_file(path):
  73     try:
  74         with open(path, "r") as f:
  75             return json.load(f)
  76     except IOError:
  77         assert False, "Could not read file {}".format(path)
  78
  79
  80 def check(name, command, checker):
  81     tgen = get_topogen()
  82     router = tgen.gears[name]
  83
  84     def _check():
  85         try:
  86             return checker(router.vtysh_cmd(command))
  87         except:
  88             return False
  89
  90     logger.info('[+] check {} "{}"'.format(name, command))
  91     _, result = topotest.run_and_expect(_check, None, count=10, wait=0.5)
  92     assert result is None, "Failed"
  93
  94
  95 def check_vrf10_bgp_rib(output):
  96     expected = open_json_file("%s/pe1/results/vrf10_ipv4_unicast.json" % CWD)
  97     actual = json.loads(output)
  98     return topotest.json_cmp(actual, expected)
  99
 100
 101 def check_default_bgp_vpn_rib(output):
 102     expected = open_json_file("%s/pe1/results/default_ipv4_vpn.json" % CWD)
 103     actual = json.loads(output)
 104     return topotest.json_cmp(actual, expected)
 105
 106
 107 def check_vrf20_bgp_rib(output):
 108     expected = open_json_file("%s/pe1/results/vrf20_ipv4_unicast.json" % CWD)
 109     actual = json.loads(output)
 110     return topotest.json_cmp(actual, expected)
 111
 112
 113 def check_vrf20_rib(output):
 114     expected = open_json_file("%s/pe1/results/vrf20_ipv4.json" % CWD)
 115     actual = json.loads(output)
 116     return topotest.json_cmp(actual, expected)
 117
 118
 119 def test_rib():
 120     check("pe1", "show bgp vrf vrf10 ipv4 unicast json", check_vrf10_bgp_rib)
 121     check("pe1", "show bgp ipv4 vpn json", check_default_bgp_vpn_rib)
 122     check("pe1", "show bgp vrf vrf20 ipv4 unicast json", check_vrf20_bgp_rib)
 123     check("pe1", "show ip route vrf vrf20 json", check_vrf20_rib)
 124
 125
 126 if __name__ == "__main__":
 127     args = ["-s"] + sys.argv[1:]
 128     sys.exit(pytest.main(args))