2 # SPDX-License-Identifier: ISC
4 # Copyright (c) 2022, LINE Corporation
5 # Authored by Ryoga Saito <ryoga.saito@linecorp.com>
15 CWD
= os
.path
.dirname(os
.path
.realpath(__file__
))
16 sys
.path
.append(os
.path
.join(CWD
, "../"))
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
25 pytestmark
= [pytest
.mark
.bgpd
]
29 tgen
.add_router("pe1")
30 tgen
.add_router("ce1")
32 tgen
.add_link(tgen
.gears
["pe1"], tgen
.gears
["ce1"], "eth0", "eth0")
35 def setup_module(mod
):
36 tgen
= Topogen(build_topo
, mod
.__name
__)
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
)))
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")
54 def teardown_module(mod
):
59 def open_json_file(path
):
61 with
open(path
, "r") as f
:
64 assert False, "Could not read file {}".format(path
)
67 def check(name
, command
, checker
):
69 router
= tgen
.gears
[name
]
73 return checker(router
.vtysh_cmd(command
))
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"
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
)
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
)
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
)
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
)
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
)
113 if __name__
== "__main__":
114 args
= ["-s"] + sys
.argv
[1:]
115 sys
.exit(pytest
.main(args
))