tests/topotests/bgp_prefix_sid2/test_bgp_prefix_sid2.py

   1 #!/usr/bin/env python
   2
   3 #
   4 # test_bgp_prefix_sid2.py
   5 # Part of NetDEF Topology Tests
   6 #
   7 # Copyright (c) 2020 by LINE Corporation
   8 # Copyright (c) 2020 by Hiroki Shirokura <slank.dev@gmail.com>
   9 #
  10 # Permission to use, copy, modify, and/or distribute this software
  11 # for any purpose with or without fee is hereby granted, provided
  12 # that the above copyright notice and this permission notice appear
  13 # in all copies.
  14 #
  15 # THE SOFTWARE IS PROVIDED "AS IS" AND NETDEF DISCLAIMS ALL WARRANTIES
  16 # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
  17 # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NETDEF BE LIABLE FOR
  18 # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
  19 # DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
  20 # WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
  21 # ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
  22 # OF THIS SOFTWARE.
  23 #
  24
  25 """
  26 test_bgp_prefix_sid2.py: Test BGP topology with EBGP on prefix-sid
  27 """
  28
  29 import json
  30 import os
  31 import sys
  32 import functools
  33 import pytest
  34
  35 CWD = os.path.dirname(os.path.realpath(__file__))
  36 sys.path.append(os.path.join(CWD, "../"))
  37
  38 # pylint: disable=C0413
  39 from lib import topotest
  40 from lib.topogen import Topogen, TopoRouter, get_topogen
  41 from lib.topolog import logger
  42
  43 pytestmark = [pytest.mark.bgpd]
  44
  45
  46 def build_topo(tgen):
  47     router = tgen.add_router("r1")
  48     switch = tgen.add_switch("s1")
  49     switch.add_link(router)
  50
  51     switch = tgen.gears["s1"]
  52     peer1 = tgen.add_exabgp_peer(
  53         "peer1", ip="10.0.0.101", defaultRoute="via 10.0.0.1"
  54     )
  55     switch.add_link(peer1)
  56
  57
  58 def setup_module(module):
  59     tgen = Topogen(build_topo, module.__name__)
  60     tgen.start_topology()
  61
  62     router = tgen.gears["r1"]
  63     router.load_config(
  64         TopoRouter.RD_ZEBRA,
  65         os.path.join(CWD, "{}/zebra.conf".format("r1"))
  66     )
  67     router.load_config(
  68         TopoRouter.RD_BGP,
  69         os.path.join(CWD, "{}/bgpd.conf".format("r1"))
  70     )
  71     router.start()
  72
  73     logger.info("starting exaBGP")
  74     peer_list = tgen.exabgp_peers()
  75     for pname, peer in peer_list.items():
  76         logger.info("starting exaBGP on {}".format(pname))
  77         peer_dir = os.path.join(CWD, pname)
  78         env_file = os.path.join(CWD, pname, "exabgp.env")
  79         logger.info("Running ExaBGP peer on {}".format(pname))
  80         peer.start(peer_dir, env_file)
  81         logger.info(pname)
  82
  83
  84 def teardown_module(module):
  85     tgen = get_topogen()
  86     tgen.stop_topology()
  87
  88
  89 def open_json_file(filename):
  90     try:
  91         with open(filename, "r") as f:
  92             return json.load(f)
  93     except IOError:
  94         assert False, "Could not read file {}".format(filename)
  95
  96
  97 def test_r1_rib():
  98     def _check(name, cmd, expected_file):
  99         logger.info("polling")
 100         tgen = get_topogen()
 101         router = tgen.gears[name]
 102         output = json.loads(router.vtysh_cmd(cmd))
 103         expected = open_json_file("{}/{}".format(CWD, expected_file))
 104         return topotest.json_cmp(output, expected)
 105
 106     def check(name, cmd, expected_file):
 107         logger.info("[+] check {} \"{}\" {}".format(name, cmd, expected_file))
 108         tgen = get_topogen()
 109         func = functools.partial(_check, name, cmd, expected_file)
 110         success, result = topotest.run_and_expect(func, None, count=10, wait=0.5)
 111         assert result is None, 'Failed'
 112
 113     check("r1", "show bgp ipv6 vpn 2001:1::/64 json", "r1/vpnv6_rib_entry1.json")
 114     check("r1", "show bgp ipv6 vpn 2001:2::/64 json", "r1/vpnv6_rib_entry2.json")
 115
 116
 117 if __name__ == "__main__":
 118     args = ["-s"] + sys.argv[1:]
 119     ret = pytest.main(args)
 120     sys.exit(ret)