tests/topotests/bgp_confed1/test_bgp_confed1.py

   1 #!/usr/bin/env python
   2
   3 #
   4 # test_bgp_confed1.py
   5 #
   6 # Copyright 2022 6WIND S.A.
   7 #
   8 # Permission to use, copy, modify, and/or distribute this software
   9 # for any purpose with or without fee is hereby granted, provided
  10 # that the above copyright notice and this permission notice appear
  11 # in all copies.
  12 #
  13 # THE SOFTWARE IS PROVIDED "AS IS" AND 6WIND DISCLAIMS ALL WARRANTIES
  14 # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
  15 # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL 6WIND BE LIABLE FOR
  16 # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
  17 # DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
  18 # WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
  19 # ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
  20 # OF THIS SOFTWARE.
  21 #
  22
  23 """
  24 test_bgp_confed1.py: Test the FRR BGP confederations with AS member
  25 same as confederation Id, verify BGP prefixes and path distribution
  26 """
  27
  28 import os
  29 import sys
  30 import json
  31 from functools import partial
  32 import pytest
  33
  34 # Save the Current Working Directory to find configuration files.
  35 CWD = os.path.dirname(os.path.realpath(__file__))
  36 sys.path.append(os.path.join(CWD, "../"))
  37
  38 # pylint: disable=C0413
  39 # Import topogen and topotest helpers
  40 from lib import topotest
  41 from lib.topogen import Topogen, TopoRouter, get_topogen
  42 from lib.topolog import logger
  43
  44 pytestmark = [pytest.mark.bgpd]
  45
  46
  47 def build_topo(tgen):
  48     for routern in range(1, 5):
  49         tgen.add_router("r{}".format(routern))
  50
  51     switch = tgen.add_switch("s1")
  52     switch.add_link(tgen.gears["r1"])
  53     switch.add_link(tgen.gears["r2"])
  54
  55     switch = tgen.add_switch("s3")
  56     switch.add_link(tgen.gears["r3"])
  57     switch.add_link(tgen.gears["r4"])
  58
  59     switch = tgen.add_switch("s2")
  60     switch.add_link(tgen.gears["r2"])
  61     switch.add_link(tgen.gears["r3"])
  62
  63 def setup_module(mod):
  64
  65     tgen = Topogen(build_topo, mod.__name__)
  66     tgen.start_topology()
  67
  68     router_list = tgen.routers()
  69     for rname, router in router_list.items():
  70         router.load_config(
  71             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
  72         )
  73         router.load_config(
  74             TopoRouter.RD_ISIS, os.path.join(CWD, "{}/isisd.conf".format(rname))
  75         )
  76         router.load_config(
  77             TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format(rname))
  78         )
  79
  80     # Initialize all routers.
  81     tgen.start_router()
  82
  83 def teardown_module(_mod):
  84     "Teardown the pytest environment"
  85     tgen = get_topogen()
  86     tgen.stop_topology()
  87
  88
  89 def test_bgp_convergence():
  90     "Assert that BGP is converging."
  91     tgen = get_topogen()
  92     if tgen.routers_have_failure():
  93         pytest.skip(tgen.errors)
  94
  95     logger.info("waiting for bgp peers to go up")
  96
  97     for router in tgen.routers().values():
  98         ref_file = "{}/{}/bgp_summary.json".format(CWD, router.name)
  99         expected = json.loads(open(ref_file).read())
 100         test_func = partial(
 101             topotest.router_json_cmp, router, "show ip bgp summary json", expected
 102         )
 103         _, res = topotest.run_and_expect(test_func, None, count=125, wait=2.0)
 104         assertmsg = "{}: bgp did not converge".format(router.name)
 105         assert res is None, assertmsg
 106
 107
 108 def test_bgp_confed_ipv4_unicast():
 109     "Assert that BGP is exchanging BGP route."
 110     tgen = get_topogen()
 111     if tgen.routers_have_failure():
 112         pytest.skip(tgen.errors)
 113
 114     logger.info("waiting for bgp peers exchanging UPDATES")
 115
 116     for router in tgen.routers().values():
 117         ref_file = "{}/{}/bgp_ipv4_unicast.json".format(CWD, router.name)
 118         expected = json.loads(open(ref_file).read())
 119         test_func = partial(
 120             topotest.router_json_cmp, router, "show bgp ipv4 unicast json", expected
 121         )
 122         _, res = topotest.run_and_expect(test_func, None, count=40, wait=2.5)
 123         assertmsg = "{}: BGP UPDATE exchange failure".format(router.name)
 124         assert res is None, assertmsg
 125
 126
 127 if __name__ == "__main__":
 128     args = ["-s"] + sys.argv[1:]
 129     sys.exit(pytest.main(args))