+++ /dev/null
-#!/usr/bin/env python
-# SPDX-License-Identifier: ISC
-
-#
-# Copyright (c) 2023 by 6WIND
-#
-
-"""
-Test the FRR BGP 6VPE functionality
-"""
-
-import os
-import sys
-import json
-import functools
-from functools import partial
-import pytest
-
-# Save the Current Working Directory to find configuration files.
-CWD = os.path.dirname(os.path.realpath(__file__))
-sys.path.append(os.path.join(CWD, "../"))
-
-# pylint: disable=C0413
-# Import topogen and topotest helpers
-from lib import topotest
-from lib.topogen import Topogen, TopoRouter, get_topogen
-from lib.topolog import logger
-from lib.checkping import check_ping
-
-pytestmark = [pytest.mark.bgpd, pytest.mark.isisd]
-
-
-def build_topo(tgen):
- """
- +---+ +---+ +---+ +---+
- | h1|----|pe1|----|pe2|----| h2|
- +---+ +---+ +---+ +---+
- """
-
- def connect_routers(tgen, left, right):
- pe = None
- host = None
- for rname in [left, right]:
- if rname not in tgen.routers().keys():
- tgen.add_router(rname)
- if "pe" in rname:
- pe = tgen.gears[rname]
- if "h" in rname:
- host = tgen.gears[rname]
-
- switch = tgen.add_switch("s-{}-{}".format(left, right))
- switch.add_link(tgen.gears[left], nodeif="eth-{}".format(right))
- switch.add_link(tgen.gears[right], nodeif="eth-{}".format(left))
-
- if pe and host:
- pe.cmd("ip link add vrf1 type vrf table 10")
- pe.cmd("ip link set vrf1 up")
- pe.cmd("ip link set dev eth-{} master vrf1".format(host.name))
-
- if "p" in left and "p" in right:
- # PE <-> P or P <-> P
- tgen.gears[left].run("sysctl -w net.mpls.conf.eth-{}.input=1".format(right))
- tgen.gears[right].run("sysctl -w net.mpls.conf.eth-{}.input=1".format(left))
-
- connect_routers(tgen, "h1", "pe1")
- connect_routers(tgen, "pe1", "pe2")
- connect_routers(tgen, "pe2", "h2")
-
-
-def setup_module(mod):
- "Sets up the pytest environment"
-
- tgen = Topogen(build_topo, mod.__name__)
- tgen.start_topology()
- logger.info("setup_module")
-
- for rname, router in tgen.routers().items():
- router.load_config(
- TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
- )
- if "h" in rname:
- # hosts
- continue
-
- router.load_config(
- TopoRouter.RD_ISIS, os.path.join(CWD, "{}/isisd.conf".format(rname))
- )
-
- router.load_config(
- TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format(rname))
- )
-
- # Initialize all routers.
- tgen.start_router()
-
-
-def teardown_module(_mod):
- "Teardown the pytest environment"
- tgen = get_topogen()
- tgen.stop_topology()
-
-
-def test_bgp_convergence():
- "Assert that BGP is converging."
- tgen = get_topogen()
- if tgen.routers_have_failure():
- pytest.skip(tgen.errors)
-
- logger.info("waiting for bgp peers to go up")
-
- router_list = ["pe1", "pe2"]
-
- for name in router_list:
- router = tgen.gears[name]
- ref_file = "{}/{}/bgp_summary.json".format(CWD, router.name)
- expected = json.loads(open(ref_file).read())
- test_func = partial(
- topotest.router_json_cmp, router, "show bgp summary json", expected
- )
- _, res = topotest.run_and_expect(test_func, None, count=90, wait=1)
- assertmsg = "{}: bgp did not converge".format(router.name)
- assert res is None, assertmsg
-
-
-def test_bgp_ipv6_vpn():
- "Assert that BGP is exchanging BGP route."
- tgen = get_topogen()
- if tgen.routers_have_failure():
- pytest.skip(tgen.errors)
-
- logger.info("waiting for bgp peers exchanging UPDATES")
-
- router_list = ["pe1", "pe2"]
-
- for name in router_list:
- router = tgen.gears[name]
- ref_file = "{}/{}/bgp_vrf_ipv6.json".format(CWD, router.name)
- expected = json.loads(open(ref_file).read())
- test_func = partial(
- topotest.router_json_cmp,
- router,
- "show bgp vrf vrf1 ipv6 unicast json",
- expected,
- )
- _, res = topotest.run_and_expect(test_func, None, count=30, wait=1)
- assertmsg = "{}: BGP UPDATE exchange failure".format(router.name)
- assert res is None, assertmsg
-
-
-def test_zebra_ipv6_installed():
- "Assert that routes are installed."
- tgen = get_topogen()
- if tgen.routers_have_failure():
- pytest.skip(tgen.errors)
- pe1 = tgen.gears["pe1"]
- logger.info("check ipv6 routes installed on pe1")
-
- ref_file = "{}/{}/ipv6_routes_vrf.json".format(CWD, pe1.name)
- expected = json.loads(open(ref_file).read())
- test_func = partial(
- topotest.router_json_cmp, pe1, "show ipv6 route vrf vrf1 json", expected
- )
- _, res = topotest.run_and_expect(test_func, None, count=30, wait=1)
- assertmsg = "{}: Zebra Installation failure on vrf vrf1".format(pe1.name)
- assert res is None, assertmsg
-
-
-def test_bgp_ping6_ok():
- "Check that h1 pings h2"
- tgen = get_topogen()
- if tgen.routers_have_failure():
- pytest.skip(tgen.errors)
-
- check_ping("h1", "fd00:200::6", True, 5, 1)
-
-
-if __name__ == "__main__":
- args = ["-s"] + sys.argv[1:]
- sys.exit(pytest.main(args))
projects / matthieu / frr.git / commitdiff