3 # Copyright (c) 2022, LINE Corporation
4 # Authored by Ryoga Saito <ryoga.saito@linecorp.com>
6 # Permission to use, copy, modify, and/or distribute this software
7 # for any purpose with or without fee is hereby granted, provided
8 # that the above copyright notice and this permission notice appear
11 # THE SOFTWARE IS PROVIDED "AS IS" AND NETDEF DISCLAIMS ALL WARRANTIES
12 # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
13 # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NETDEF BE LIABLE FOR
14 # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
15 # DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
16 # WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
17 # ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
28 CWD
= os
.path
.dirname(os
.path
.realpath(__file__
))
29 sys
.path
.append(os
.path
.join(CWD
, "../"))
31 # pylint: disable=C0413
32 # Import topogen and topotest helpers
33 from lib
import topotest
34 from lib
.topogen
import Topogen
, TopoRouter
, get_topogen
35 from lib
.topolog
import logger
36 from lib
.common_config
import required_linux_kernel_version
38 pytestmark
= [pytest
.mark
.bgpd
]
42 tgen
.add_router("pe1")
43 tgen
.add_router("ce1")
45 tgen
.add_link(tgen
.gears
["pe1"], tgen
.gears
["ce1"], "eth0", "eth0")
48 def setup_module(mod
):
49 tgen
= Topogen(build_topo
, mod
.__name
__)
52 for rname
, router
in tgen
.routers().items():
53 router
.load_config(TopoRouter
.RD_ZEBRA
,
54 os
.path
.join(CWD
, '{}/zebra.conf'.format(rname
)))
55 router
.load_config(TopoRouter
.RD_BGP
,
56 os
.path
.join(CWD
, '{}/bgpd.conf'.format(rname
)))
58 tgen
.gears
["pe1"].run("ip link add vrf10 type vrf table 10")
59 tgen
.gears
["pe1"].run("ip link set vrf10 up")
60 tgen
.gears
["pe1"].run("ip link add vrf20 type vrf table 20")
61 tgen
.gears
["pe1"].run("ip link set vrf20 up")
62 tgen
.gears
["pe1"].run("ip link set eth0 master vrf10")
67 def teardown_module(mod
):
72 def open_json_file(path
):
74 with
open(path
, "r") as f
:
77 assert False, "Could not read file {}".format(path
)
80 def check(name
, command
, checker
):
82 router
= tgen
.gears
[name
]
86 return checker(router
.vtysh_cmd(command
))
90 logger
.info('[+] check {} "{}"'.format(name
, command
))
91 _
, result
= topotest
.run_and_expect(_check
, None, count
=10, wait
=0.5)
92 assert result
is None, "Failed"
95 def check_vrf10_bgp_rib(output
):
96 expected
= open_json_file("%s/pe1/results/vrf10_ipv4_unicast.json" % CWD
)
97 actual
= json
.loads(output
)
98 return topotest
.json_cmp(actual
, expected
)
101 def check_default_bgp_vpn_rib(output
):
102 expected
= open_json_file("%s/pe1/results/default_ipv4_vpn.json" % CWD
)
103 actual
= json
.loads(output
)
104 return topotest
.json_cmp(actual
, expected
)
107 def check_vrf20_bgp_rib(output
):
108 expected
= open_json_file("%s/pe1/results/vrf20_ipv4_unicast.json" % CWD
)
109 actual
= json
.loads(output
)
110 return topotest
.json_cmp(actual
, expected
)
113 def check_vrf20_rib(output
):
114 expected
= open_json_file("%s/pe1/results/vrf20_ipv4.json" % CWD
)
115 actual
= json
.loads(output
)
116 return topotest
.json_cmp(actual
, expected
)
120 check("pe1", "show bgp vrf vrf10 ipv4 unicast json", check_vrf10_bgp_rib
)
121 check("pe1", "show bgp ipv4 vpn json", check_default_bgp_vpn_rib
)
122 check("pe1", "show bgp vrf vrf20 ipv4 unicast json", check_vrf20_bgp_rib
)
123 check("pe1", "show ip route vrf vrf20 json", check_vrf20_rib
)
126 if __name__
== "__main__":
127 args
= ["-s"] + sys
.argv
[1:]
128 sys
.exit(pytest
.main(args
))