]>
Commit | Line | Data |
---|---|---|
1 | #!/usr/bin/env python | |
2 | # SPDX-License-Identifier: ISC | |
3 | ||
4 | # Copyright (c) 2022, LINE Corporation | |
5 | # Authored by Ryoga Saito <ryoga.saito@linecorp.com> | |
6 | # | |
7 | ||
8 | import os | |
9 | import re | |
10 | import sys | |
11 | import json | |
12 | import functools | |
13 | import pytest | |
14 | ||
15 | CWD = os.path.dirname(os.path.realpath(__file__)) | |
16 | sys.path.append(os.path.join(CWD, "../")) | |
17 | ||
18 | # pylint: disable=C0413 | |
19 | # Import topogen and topotest helpers | |
20 | from lib import topotest | |
21 | from lib.topogen import Topogen, TopoRouter, get_topogen | |
22 | from lib.topolog import logger | |
23 | from lib.common_config import required_linux_kernel_version | |
24 | ||
25 | pytestmark = [pytest.mark.bgpd] | |
26 | ||
27 | ||
28 | def build_topo(tgen): | |
29 | tgen.add_router("pe1") | |
30 | tgen.add_router("ce1") | |
31 | ||
32 | tgen.add_link(tgen.gears["pe1"], tgen.gears["ce1"], "eth0", "eth0") | |
33 | ||
34 | ||
35 | def setup_module(mod): | |
36 | tgen = Topogen(build_topo, mod.__name__) | |
37 | tgen.start_topology() | |
38 | ||
39 | for rname, router in tgen.routers().items(): | |
40 | router.load_config(TopoRouter.RD_ZEBRA, | |
41 | os.path.join(CWD, '{}/zebra.conf'.format(rname))) | |
42 | router.load_config(TopoRouter.RD_BGP, | |
43 | os.path.join(CWD, '{}/bgpd.conf'.format(rname))) | |
44 | ||
45 | tgen.gears["pe1"].run("ip link add vrf10 type vrf table 10") | |
46 | tgen.gears["pe1"].run("ip link set vrf10 up") | |
47 | tgen.gears["pe1"].run("ip link add vrf20 type vrf table 20") | |
48 | tgen.gears["pe1"].run("ip link set vrf20 up") | |
49 | tgen.gears["pe1"].run("ip link set eth0 master vrf10") | |
50 | ||
51 | tgen.start_router() | |
52 | ||
53 | ||
54 | def teardown_module(mod): | |
55 | tgen = get_topogen() | |
56 | tgen.stop_topology() | |
57 | ||
58 | ||
59 | def open_json_file(path): | |
60 | try: | |
61 | with open(path, "r") as f: | |
62 | return json.load(f) | |
63 | except IOError: | |
64 | assert False, "Could not read file {}".format(path) | |
65 | ||
66 | ||
67 | def check_vrf10_rib(output): | |
68 | expected = open_json_file("%s/pe1/results/vrf10_ipv4_unicast.json" % CWD) | |
69 | actual = json.loads(output) | |
70 | return topotest.json_cmp(actual, expected) | |
71 | ||
72 | ||
73 | def check_default_vpn_rib(output): | |
74 | expected = open_json_file("%s/pe1/results/default_ipv4_vpn.json" % CWD) | |
75 | actual = json.loads(output) | |
76 | return topotest.json_cmp(actual, expected) | |
77 | ||
78 | ||
79 | def check_vrf20_rib(output): | |
80 | expected = open_json_file("%s/pe1/results/vrf20_ipv4_unicast.json" % CWD) | |
81 | actual = json.loads(output) | |
82 | return topotest.json_cmp(actual, expected) | |
83 | ||
84 | ||
85 | def check(name, command, checker): | |
86 | tgen = get_topogen() | |
87 | router = tgen.gears[name] | |
88 | ||
89 | def _check(): | |
90 | try: | |
91 | return checker(router.vtysh_cmd(command)) | |
92 | except: | |
93 | return False | |
94 | ||
95 | logger.info('[+] check {} "{}"'.format(name, command)) | |
96 | _, result = topotest.run_and_expect(_check, None, count=10, wait=0.5) | |
97 | assert result is None, "Failed" | |
98 | ||
99 | ||
100 | def test_rib(): | |
101 | check("pe1", "show bgp vrf vrf10 ipv4 unicast json", check_vrf10_rib) | |
102 | check("pe1", "show bgp ipv4 vpn json", check_default_vpn_rib) | |
103 | check("pe1", "show bgp vrf vrf20 ipv4 unicast json", check_vrf20_rib) | |
104 | ||
105 | ||
106 | if __name__ == "__main__": | |
107 | args = ["-s"] + sys.argv[1:] | |
108 | sys.exit(pytest.main(args)) |