2 # SPDX-License-Identifier: ISC
5 # test_ospf_multi_vrf_bgp_route_leak.py
7 # Copyright (c) 2022 ATCorp
13 from functools
import partial
16 # pylint: disable=C0413
17 # Import topogen and topotest helpers
18 from lib
import topotest
19 from lib
.topogen
import Topogen
, TopoRouter
, get_topogen
20 from lib
.topolog
import logger
24 test_ospf_multi_vrf_bgp_route_leak.py: Test OSPF with multi vrf setup and route leaking.
28 bgp route leaking (connected/ospf), vrfs: neno <==> default <==> ray
29 routes leaking to vrfs are limited to neno and ray routes.
36 +-----------------+ R1 +
39 |.3 .1|vrf:default | 10.0.4.4/24
44 +---+---+ +---+---+ +---+---+
46 v | R2 +----------------+
47 10.0.3.3/24 | |.2 10.0.40.0/24
55 # Save the Current Working Directory to find configuration files.
56 CWD
= os
.path
.dirname(os
.path
.realpath(__file__
))
57 sys
.path
.append(os
.path
.join(CWD
, "../"))
59 # Required to instantiate the topology builder class.
61 pytestmark
= [pytest
.mark
.ospfd
, pytest
.mark
.bgpd
]
68 for routern
in range(1, 5):
69 tgen
.add_router("r{}".format(routern
))
71 # Create a empty network for router 1
72 switch
= tgen
.add_switch("s1")
73 switch
.add_link(tgen
.gears
["r1"])
75 # Create a empty network for router 2
76 switch
= tgen
.add_switch("s2")
77 switch
.add_link(tgen
.gears
["r2"])
79 # Create a empty network for router 3
80 switch
= tgen
.add_switch("s3")
81 switch
.add_link(tgen
.gears
["r3"])
83 # Create a empty network for router 4
84 switch
= tgen
.add_switch("s4")
85 switch
.add_link(tgen
.gears
["r4"])
87 # Interconect router 1, 2
88 switch
= tgen
.add_switch("s1-2")
89 switch
.add_link(tgen
.gears
["r1"])
90 switch
.add_link(tgen
.gears
["r2"])
92 # Interconect router 1, 3
93 switch
= tgen
.add_switch("s1-3")
94 switch
.add_link(tgen
.gears
["r1"])
95 switch
.add_link(tgen
.gears
["r3"])
97 # Interconect router 2, 4
98 switch
= tgen
.add_switch("s2-4")
99 switch
.add_link(tgen
.gears
["r2"])
100 switch
.add_link(tgen
.gears
["r4"])
103 def setup_module(mod
):
104 logger
.info("OSPF Multi VRF Topology with BGP route leaking:\n {}".format(TOPOLOGY
))
106 tgen
= Topogen(build_topo
, mod
.__name
__)
107 tgen
.start_topology()
109 r1_vrf_setup_cmds
= [
110 "ip link add name neno type vrf table 11",
111 "ip link set dev neno up",
112 "ip link set dev r1-eth2 vrf neno up",
114 r2_vrf_setup_cmds
= [
115 "ip link add name ray type vrf table 11",
116 "ip link set dev ray up",
117 "ip link set dev r2-eth2 vrf ray up",
121 router_list
= tgen
.routers()
123 # Create VRFs on r1/r2 and bind to interfaces
124 for cmd
in r1_vrf_setup_cmds
:
125 tgen
.net
["r1"].cmd(cmd
)
126 for cmd
in r2_vrf_setup_cmds
:
127 tgen
.net
["r2"].cmd(cmd
)
129 logger
.info("Testing OSPF VRF support")
131 for rname
, router
in router_list
.items():
132 logger
.info("Loading router %s" % rname
)
133 router
.load_frr_config(os
.path
.join(CWD
, "{}/frr.conf".format(rname
)))
135 # Initialize all routers.
137 for router
in router_list
.values():
138 if router
.has_version("<", "4.0"):
139 tgen
.set_error("unsupported version")
142 def teardown_module(mod
):
143 "Teardown the pytest environment"
148 # Shared test function to validate expected output.
149 def compare_show_ip_route_vrf(rname
, expected
, vrf_name
):
151 Calls 'show ip route vrf [vrf_name] route' and compare the obtained
152 result with the expected output.
155 current
= topotest
.ip4_route_zebra(tgen
.gears
[rname
], vrf_name
)
156 ret
= topotest
.difflines(
157 current
, expected
, title1
="Current output", title2
="Expected output"
162 def test_ospf_convergence():
163 "Test OSPF daemon convergence"
166 if tgen
.routers_have_failure():
167 pytest
.skip("skipped because of router(s) failure")
169 for rname
, router
in tgen
.routers().items():
170 logger
.info('Waiting for router "%s" convergence', rname
)
172 for vrf
in ["default", "neno", "ray"]:
173 # Load expected results from the command
174 reffile
= os
.path
.join(CWD
, "{}/ospf-vrf-{}.txt".format(rname
, vrf
))
175 if vrf
== "default" or os
.path
.exists(reffile
):
176 expected
= open(reffile
).read()
178 # Run test function until we get an result. Wait at most 80 seconds.
180 topotest
.router_output_cmp
,
182 "show ip ospf vrf {} route".format(vrf
),
185 result
, diff
= topotest
.run_and_expect(
186 test_func
, "", count
=80, wait
=1
188 assertmsg
= "OSPF did not converge on {}:\n{}".format(rname
, diff
)
189 assert result
, assertmsg
192 def test_ospf_kernel_route():
193 "Test OSPF kernel route installation"
196 if tgen
.routers_have_failure():
197 pytest
.skip("skipped because of router(s) failure")
199 rlist
= tgen
.routers().values()
201 logger
.info('Checking OSPF IPv4 kernel routes in "%s"', router
.name
)
202 for vrf
in ["default", "neno", "ray"]:
203 reffile
= os
.path
.join(CWD
, "{}/zebra-vrf-{}.txt".format(router
.name
, vrf
))
204 if vrf
== "default" or os
.path
.exists(reffile
):
205 expected
= open(reffile
).read()
206 # Run test function until we get an result. Wait at most 80 seconds.
208 compare_show_ip_route_vrf
, router
.name
, expected
, vrf
210 result
, diff
= topotest
.run_and_expect(
211 test_func
, "", count
=80, wait
=1
213 assertmsg
= 'OSPF IPv4 route mismatch in router "{}": {}'.format(
216 assert result
, assertmsg
219 def test_memory_leak():
220 "Run the memory leak test and report results."
222 if not tgen
.is_memleak_enabled():
223 pytest
.skip("Memory leak test/report is disabled")
225 tgen
.report_memory_leaks()
228 if __name__
== "__main__":
229 args
= ["-s"] + sys
.argv
[1:]
230 sys
.exit(pytest
.main(args
))