4 # test_ospf_multi_vrf_bgp_route_leak.py
6 # Copyright (c) 2022 ATCorp
9 # Permission to use, copy, modify, and/or distribute this software
10 # for any purpose with or without fee is hereby granted, provided
11 # that the above copyright notice and this permission notice appear
14 # THE SOFTWARE IS PROVIDED "AS IS" AND NETDEF DISCLAIMS ALL WARRANTIES
15 # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
16 # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NETDEF BE LIABLE FOR
17 # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
18 # DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
19 # WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
20 # ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
26 from functools
import partial
29 # pylint: disable=C0413
30 # Import topogen and topotest helpers
31 from lib
import topotest
32 from lib
.topogen
import Topogen
, TopoRouter
, get_topogen
33 from lib
.topolog
import logger
37 test_ospf_multi_vrf_bgp_route_leak.py: Test OSPF with multi vrf setup and route leaking.
41 bgp route leaking (connected/ospf), vrfs: neno <==> default <==> ray
42 routes leaking to vrfs are limited to neno and ray routes.
49 +-----------------+ R1 +
52 |.3 .1|vrf:default | 10.0.4.4/24
57 +---+---+ +---+---+ +---+---+
59 v | R2 +----------------+
60 10.0.3.3/24 | |.2 10.0.40.0/24
68 # Save the Current Working Directory to find configuration files.
69 CWD
= os
.path
.dirname(os
.path
.realpath(__file__
))
70 sys
.path
.append(os
.path
.join(CWD
, "../"))
72 # Required to instantiate the topology builder class.
74 pytestmark
= [pytest
.mark
.ospfd
, pytest
.mark
.bgpd
]
81 for routern
in range(1, 5):
82 tgen
.add_router("r{}".format(routern
))
84 # Create a empty network for router 1
85 switch
= tgen
.add_switch("s1")
86 switch
.add_link(tgen
.gears
["r1"])
88 # Create a empty network for router 2
89 switch
= tgen
.add_switch("s2")
90 switch
.add_link(tgen
.gears
["r2"])
92 # Create a empty network for router 3
93 switch
= tgen
.add_switch("s3")
94 switch
.add_link(tgen
.gears
["r3"])
96 # Create a empty network for router 4
97 switch
= tgen
.add_switch("s4")
98 switch
.add_link(tgen
.gears
["r4"])
100 # Interconect router 1, 2
101 switch
= tgen
.add_switch("s1-2")
102 switch
.add_link(tgen
.gears
["r1"])
103 switch
.add_link(tgen
.gears
["r2"])
105 # Interconect router 1, 3
106 switch
= tgen
.add_switch("s1-3")
107 switch
.add_link(tgen
.gears
["r1"])
108 switch
.add_link(tgen
.gears
["r3"])
110 # Interconect router 2, 4
111 switch
= tgen
.add_switch("s2-4")
112 switch
.add_link(tgen
.gears
["r2"])
113 switch
.add_link(tgen
.gears
["r4"])
116 def setup_module(mod
):
117 logger
.info("OSPF Multi VRF Topology with BGP route leaking:\n {}".format(TOPOLOGY
))
119 tgen
= Topogen(build_topo
, mod
.__name
__)
120 tgen
.start_topology()
122 r1_vrf_setup_cmds
= [
123 "ip link add name neno type vrf table 11",
124 "ip link set dev neno up",
125 "ip link set dev r1-eth2 vrf neno up",
127 r2_vrf_setup_cmds
= [
128 "ip link add name ray type vrf table 11",
129 "ip link set dev ray up",
130 "ip link set dev r2-eth2 vrf ray up",
134 router_list
= tgen
.routers()
136 # Create VRFs on r1/r2 and bind to interfaces
137 for cmd
in r1_vrf_setup_cmds
:
138 tgen
.net
["r1"].cmd(cmd
)
139 for cmd
in r2_vrf_setup_cmds
:
140 tgen
.net
["r2"].cmd(cmd
)
142 logger
.info("Testing OSPF VRF support")
144 for rname
, router
in router_list
.items():
145 logger
.info("Loading router %s" % rname
)
146 router
.load_frr_config(os
.path
.join(CWD
, "{}/frr.conf".format(rname
)))
148 # Initialize all routers.
150 for router
in router_list
.values():
151 if router
.has_version("<", "4.0"):
152 tgen
.set_error("unsupported version")
155 def teardown_module(mod
):
156 "Teardown the pytest environment"
161 # Shared test function to validate expected output.
162 def compare_show_ip_route_vrf(rname
, expected
, vrf_name
):
164 Calls 'show ip route vrf [vrf_name] route' and compare the obtained
165 result with the expected output.
168 current
= topotest
.ip4_route_zebra(tgen
.gears
[rname
], vrf_name
)
169 ret
= topotest
.difflines(
170 current
, expected
, title1
="Current output", title2
="Expected output"
175 def test_ospf_convergence():
176 "Test OSPF daemon convergence"
179 if tgen
.routers_have_failure():
180 pytest
.skip("skipped because of router(s) failure")
182 for rname
, router
in tgen
.routers().items():
183 logger
.info('Waiting for router "%s" convergence', rname
)
185 for vrf
in ["default", "neno", "ray"]:
186 # Load expected results from the command
187 reffile
= os
.path
.join(CWD
, "{}/ospf-vrf-{}.txt".format(rname
, vrf
))
188 if vrf
== "default" or os
.path
.exists(reffile
):
189 expected
= open(reffile
).read()
191 # Run test function until we get an result. Wait at most 80 seconds.
193 topotest
.router_output_cmp
,
195 "show ip ospf vrf {} route".format(vrf
),
198 result
, diff
= topotest
.run_and_expect(
199 test_func
, "", count
=80, wait
=1
201 assertmsg
= "OSPF did not converge on {}:\n{}".format(rname
, diff
)
202 assert result
, assertmsg
205 def test_ospf_kernel_route():
206 "Test OSPF kernel route installation"
209 if tgen
.routers_have_failure():
210 pytest
.skip("skipped because of router(s) failure")
212 rlist
= tgen
.routers().values()
214 logger
.info('Checking OSPF IPv4 kernel routes in "%s"', router
.name
)
215 for vrf
in ["default", "neno", "ray"]:
216 reffile
= os
.path
.join(CWD
, "{}/zebra-vrf-{}.txt".format(router
.name
, vrf
))
217 if vrf
== "default" or os
.path
.exists(reffile
):
218 expected
= open(reffile
).read()
219 # Run test function until we get an result. Wait at most 80 seconds.
221 compare_show_ip_route_vrf
, router
.name
, expected
, vrf
223 result
, diff
= topotest
.run_and_expect(
224 test_func
, "", count
=80, wait
=1
226 assertmsg
= 'OSPF IPv4 route mismatch in router "{}": {}'.format(
229 assert result
, assertmsg
232 def test_memory_leak():
233 "Run the memory leak test and report results."
235 if not tgen
.is_memleak_enabled():
236 pytest
.skip("Memory leak test/report is disabled")
238 tgen
.report_memory_leaks()
241 if __name__
== "__main__":
242 args
= ["-s"] + sys
.argv
[1:]
243 sys
.exit(pytest
.main(args
))