tests/topotests/bgp_confed1/test_bgp_confed1.py

   1 #!/usr/bin/env python
   2 # SPDX-License-Identifier: ISC
   3
   4 #
   5 # test_bgp_confed1.py
   6 #
   7 # Copyright 2022 6WIND S.A.
   8 #
   9
  10 """
  11 test_bgp_confed1.py: Test the FRR BGP confederations with AS member
  12 same as confederation Id, verify BGP prefixes and path distribution
  13 """
  14
  15 import os
  16 import sys
  17 import json
  18 from functools import partial
  19 import pytest
  20
  21 # Save the Current Working Directory to find configuration files.
  22 CWD = os.path.dirname(os.path.realpath(__file__))
  23 sys.path.append(os.path.join(CWD, "../"))
  24
  25 # pylint: disable=C0413
  26 # Import topogen and topotest helpers
  27 from lib import topotest
  28 from lib.topogen import Topogen, TopoRouter, get_topogen
  29 from lib.topolog import logger
  30
  31 pytestmark = [pytest.mark.bgpd]
  32
  33
  34 def build_topo(tgen):
  35     for routern in range(1, 5):
  36         tgen.add_router("r{}".format(routern))
  37
  38     switch = tgen.add_switch("s1")
  39     switch.add_link(tgen.gears["r1"])
  40     switch.add_link(tgen.gears["r2"])
  41
  42     switch = tgen.add_switch("s3")
  43     switch.add_link(tgen.gears["r3"])
  44     switch.add_link(tgen.gears["r4"])
  45
  46     switch = tgen.add_switch("s2")
  47     switch.add_link(tgen.gears["r2"])
  48     switch.add_link(tgen.gears["r3"])
  49
  50 def setup_module(mod):
  51
  52     tgen = Topogen(build_topo, mod.__name__)
  53     tgen.start_topology()
  54
  55     router_list = tgen.routers()
  56     for rname, router in router_list.items():
  57         router.load_config(
  58             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
  59         )
  60         router.load_config(
  61             TopoRouter.RD_ISIS, os.path.join(CWD, "{}/isisd.conf".format(rname))
  62         )
  63         router.load_config(
  64             TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format(rname))
  65         )
  66
  67     # Initialize all routers.
  68     tgen.start_router()
  69
  70 def teardown_module(_mod):
  71     "Teardown the pytest environment"
  72     tgen = get_topogen()
  73     tgen.stop_topology()
  74
  75
  76 def test_bgp_convergence():
  77     "Assert that BGP is converging."
  78     tgen = get_topogen()
  79     if tgen.routers_have_failure():
  80         pytest.skip(tgen.errors)
  81
  82     logger.info("waiting for bgp peers to go up")
  83
  84     for router in tgen.routers().values():
  85         ref_file = "{}/{}/bgp_summary.json".format(CWD, router.name)
  86         expected = json.loads(open(ref_file).read())
  87         test_func = partial(
  88             topotest.router_json_cmp, router, "show ip bgp summary json", expected
  89         )
  90         _, res = topotest.run_and_expect(test_func, None, count=125, wait=2.0)
  91         assertmsg = "{}: bgp did not converge".format(router.name)
  92         assert res is None, assertmsg
  93
  94
  95 def test_bgp_confed_ipv4_unicast():
  96     "Assert that BGP is exchanging BGP route."
  97     tgen = get_topogen()
  98     if tgen.routers_have_failure():
  99         pytest.skip(tgen.errors)
 100
 101     logger.info("waiting for bgp peers exchanging UPDATES")
 102
 103     for router in tgen.routers().values():
 104         ref_file = "{}/{}/bgp_ipv4_unicast.json".format(CWD, router.name)
 105         expected = json.loads(open(ref_file).read())
 106         test_func = partial(
 107             topotest.router_json_cmp, router, "show bgp ipv4 unicast json", expected
 108         )
 109         _, res = topotest.run_and_expect(test_func, None, count=40, wait=2.5)
 110         assertmsg = "{}: BGP UPDATE exchange failure".format(router.name)
 111         assert res is None, assertmsg
 112
 113
 114 if __name__ == "__main__":
 115     args = ["-s"] + sys.argv[1:]
 116     sys.exit(pytest.main(args))