]>
Commit | Line | Data |
---|---|---|
4d85efb0 | 1 | #!/usr/bin/env python |
acddc0ed | 2 | # SPDX-License-Identifier: ISC |
4d85efb0 RS |
3 | |
4 | # Copyright (c) 2022, LINE Corporation | |
5 | # Authored by Ryoga Saito <ryoga.saito@linecorp.com> | |
6 | # | |
4d85efb0 RS |
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(name, command, checker): | |
68 | tgen = get_topogen() | |
69 | router = tgen.gears[name] | |
70 | ||
71 | def _check(): | |
72 | try: | |
73 | return checker(router.vtysh_cmd(command)) | |
74 | except: | |
75 | return False | |
76 | ||
77 | logger.info('[+] check {} "{}"'.format(name, command)) | |
78 | _, result = topotest.run_and_expect(_check, None, count=10, wait=0.5) | |
79 | assert result is None, "Failed" | |
80 | ||
81 | ||
82 | def check_vrf10_bgp_rib(output): | |
83 | expected = open_json_file("%s/pe1/results/vrf10_ipv4_unicast.json" % CWD) | |
84 | actual = json.loads(output) | |
85 | return topotest.json_cmp(actual, expected) | |
86 | ||
87 | ||
88 | def check_default_bgp_vpn_rib(output): | |
89 | expected = open_json_file("%s/pe1/results/default_ipv4_vpn.json" % CWD) | |
90 | actual = json.loads(output) | |
91 | return topotest.json_cmp(actual, expected) | |
92 | ||
93 | ||
94 | def check_vrf20_bgp_rib(output): | |
95 | expected = open_json_file("%s/pe1/results/vrf20_ipv4_unicast.json" % CWD) | |
96 | actual = json.loads(output) | |
97 | return topotest.json_cmp(actual, expected) | |
98 | ||
99 | ||
100 | def check_vrf20_rib(output): | |
101 | expected = open_json_file("%s/pe1/results/vrf20_ipv4.json" % CWD) | |
102 | actual = json.loads(output) | |
103 | return topotest.json_cmp(actual, expected) | |
104 | ||
105 | ||
106 | def test_rib(): | |
107 | check("pe1", "show bgp vrf vrf10 ipv4 unicast json", check_vrf10_bgp_rib) | |
108 | check("pe1", "show bgp ipv4 vpn json", check_default_bgp_vpn_rib) | |
109 | check("pe1", "show bgp vrf vrf20 ipv4 unicast json", check_vrf20_bgp_rib) | |
110 | check("pe1", "show ip route vrf vrf20 json", check_vrf20_rib) | |
111 | ||
112 | ||
113 | if __name__ == "__main__": | |
114 | args = ["-s"] + sys.argv[1:] | |
115 | sys.exit(pytest.main(args)) |