2 # SPDX-License-Identifier: ISC
5 # test_ospf6_topo1_vrf.py
6 # Part of NetDEF Topology Tests
8 # Copyright (c) 2021 by Niral Networks, Inc. ("Niral Networks")
9 # Used Copyright (c) 2016 by Network Device Education Foundation,
10 # Inc. ("NetDEF") in this file.
14 test_ospf6_topo1_vrf.py:
17 SW1 - Stub Net 1 SW2 - Stub Net 2 \
18 fc00:1:1:1::/64 fc00:2:2:2::/64 \
19 \___________________/ \___________________/ |
23 +---------+---------+ +---------+---------+ |
25 | FRRouting | | FRRouting | |
26 | Rtr-ID: 10.0.0.1 | | Rtr-ID: 10.0.0.2 | |
27 +---------+---------+ +---------+---------+ |
29 \______ ___________/ OSPFv3
35 ~~ fc00:A:A:A::/64 ~~ |
38 | ::3 | SW3 - Stub Net 3 |
39 +---------+---------+ /-+ fc00:3:3:3::/64 |
41 | FRRouting +--/ \---- /
42 | Rtr-ID: 10.0.0.3 | ::3 ___________/
43 +---------+---------+ \
49 ~~ fc00:B:B:B::/64 ~~ \
50 ~~~~~~~~~~~~~~~~~~ OSPFv3
53 +---------+---------+ /---- |
54 | R4 | | SW4 - Stub Net 4 |
55 | FRRouting +------+ fc00:4:4:4::/64 |
56 | Rtr-ID: 10.0.0.4 | ::4 | /
57 +-------------------+ \---- /
66 from functools
import partial
69 # Save the Current Working Directory to find configuration files later.
70 CWD
= os
.path
.dirname(os
.path
.realpath(__file__
))
71 sys
.path
.append(os
.path
.join(CWD
, "../"))
73 # pylint: disable=C0413
74 # Import topogen and topotest helpers
75 from lib
import topotest
76 from lib
.topogen
import Topogen
, TopoRouter
, get_topogen
77 from lib
.topolog
import logger
78 from lib
.topotest
import iproute2_is_vrf_capable
79 from lib
.common_config
import required_linux_kernel_version
82 pytestmark
= [pytest
.mark
.ospfd
]
89 for routern
in range(1, 5):
90 tgen
.add_router("r{}".format(routern
))
93 # Wire up the switches and routers
94 # Note that we specify the link names so we match the config files
97 # Create a empty network for router 1
98 switch
= tgen
.add_switch("s1")
99 switch
.add_link(tgen
.gears
["r1"], nodeif
="r1-stubnet")
101 # Create a empty network for router 2
102 switch
= tgen
.add_switch("s2")
103 switch
.add_link(tgen
.gears
["r2"], nodeif
="r2-stubnet")
105 # Create a empty network for router 3
106 switch
= tgen
.add_switch("s3")
107 switch
.add_link(tgen
.gears
["r3"], nodeif
="r3-stubnet")
109 # Create a empty network for router 4
110 switch
= tgen
.add_switch("s4")
111 switch
.add_link(tgen
.gears
["r4"], nodeif
="r4-stubnet")
113 # Interconnect routers 1, 2, and 3
114 switch
= tgen
.add_switch("s5")
115 switch
.add_link(tgen
.gears
["r1"], nodeif
="r1-sw5")
116 switch
.add_link(tgen
.gears
["r2"], nodeif
="r2-sw5")
117 switch
.add_link(tgen
.gears
["r3"], nodeif
="r3-sw5")
119 # Interconnect routers 3 and 4
120 switch
= tgen
.add_switch("s6")
121 switch
.add_link(tgen
.gears
["r3"], nodeif
="r3-sw6")
122 switch
.add_link(tgen
.gears
["r4"], nodeif
="r4-sw6")
125 #####################################################
129 #####################################################
132 def setup_module(mod
):
133 "Sets up the pytest environment"
135 # Required linux kernel version for this suite to run.
136 result
= required_linux_kernel_version("5.0")
137 if result
is not True:
138 pytest
.skip("Kernel requirements are not met")
140 tgen
= Topogen(build_topo
, mod
.__name
__)
141 tgen
.start_topology()
143 logger
.info("** %s: Setup Topology" % mod
.__name
__)
144 logger
.info("******************************************")
146 # For debugging after starting net, but before starting FRR,
147 # uncomment the next line
150 logger
.info("Testing with VRF Lite support")
153 "ip link add {0}-cust1 type vrf table 1001",
154 "ip link add loop1 type dummy",
155 "ip link set {0}-stubnet master {0}-cust1",
158 cmds1
= ["ip link set {0}-sw5 master {0}-cust1"]
160 cmds2
= ["ip link set {0}-sw6 master {0}-cust1"]
162 # For all registered routers, load the zebra configuration file
163 for rname
, router
in tgen
.routers().items():
164 # create VRF rx-cust1 and link rx-eth0 to rx-cust1
166 output
= tgen
.net
[rname
].cmd(cmd
.format(rname
))
167 if rname
== "r1" or rname
== "r2" or rname
== "r3":
169 output
= tgen
.net
[rname
].cmd(cmd
.format(rname
))
170 if rname
== "r3" or rname
== "r4":
172 output
= tgen
.net
[rname
].cmd(cmd
.format(rname
))
174 for rname
, router
in tgen
.routers().items():
176 TopoRouter
.RD_ZEBRA
, os
.path
.join(CWD
, "{}/zebra.conf".format(rname
))
179 TopoRouter
.RD_OSPF6
, os
.path
.join(CWD
, "{}/ospf6d.conf".format(rname
))
182 # Initialize all routers.
185 # For debugging after starting FRR daemons, uncomment the next line
189 def teardown_module(mod
):
190 "Teardown the pytest environment"
195 def test_wait_protocol_convergence():
196 "Wait for OSPFv3 to converge"
198 if tgen
.routers_have_failure():
199 pytest
.skip(tgen
.errors
)
201 logger
.info("waiting for protocols to converge")
203 def expect_neighbor_full(router
, neighbor
):
204 "Wait until OSPFv3 convergence."
205 logger
.info("waiting OSPFv3 router '{}'".format(router
))
207 topotest
.router_json_cmp
,
209 "show ipv6 ospf6 vrf {0}-cust1 neighbor json".format(router
),
210 {"neighbors": [{"neighborId": neighbor
, "state": "Full"}]},
212 _
, result
= topotest
.run_and_expect(test_func
, None, count
=130, wait
=1)
213 assertmsg
= '"{}" convergence failure'.format(router
)
214 assert result
is None, assertmsg
216 expect_neighbor_full("r1", "10.0.0.2")
217 expect_neighbor_full("r1", "10.0.0.3")
219 expect_neighbor_full("r2", "10.0.0.1")
220 expect_neighbor_full("r2", "10.0.0.3")
222 expect_neighbor_full("r3", "10.0.0.1")
223 expect_neighbor_full("r3", "10.0.0.2")
224 expect_neighbor_full("r3", "10.0.0.4")
227 def compare_show_ipv6_vrf(rname
, expected
):
229 Calls 'show ipv6 route' for router `rname` and compare the obtained
230 result with the expected output.
234 # Use the vtysh output, with some masking to make comparison easy
235 vrf_name
= "{0}-cust1".format(rname
)
236 current
= topotest
.ip6_route_zebra(tgen
.gears
[rname
], vrf_name
)
238 # Use just the 'O'spf lines of the output
240 for line
in current
.splitlines():
241 if re
.match("^O", line
):
244 current
= "\n".join(linearr
)
246 return topotest
.difflines(
247 topotest
.normalize_text(current
),
248 topotest
.normalize_text(expected
),
249 title1
="Current output",
250 title2
="Expected output",
254 def test_ospfv3_routingTable():
257 if tgen
.routers_have_failure():
258 pytest
.skip("skipped because of router(s) failure")
260 # For debugging, uncomment the next line
262 # Verify OSPFv3 Routing Table
263 for router
, rnode
in tgen
.routers().items():
264 logger
.info('Waiting for router "%s" convergence', router
)
266 # Load expected results from the command
267 reffile
= os
.path
.join(CWD
, "{}/show_ipv6_vrf_route.ref".format(router
))
268 expected
= open(reffile
).read()
270 # Run test function until we get an result. Wait at most 60 seconds.
271 test_func
= partial(compare_show_ipv6_vrf
, router
, expected
)
272 result
, diff
= topotest
.run_and_expect(test_func
, "", count
=120, wait
=0.5)
273 assert result
, "OSPFv3 did not converge on {}:\n{}".format(router
, diff
)
276 def test_linux_ipv6_kernel_routingTable():
278 # Required linux kernel version for this suite to run.
279 result
= required_linux_kernel_version("4.15")
280 if result
is not True:
281 pytest
.skip("Kernel requirements are not met")
283 # iproute2 needs to support VRFs for this suite to run.
284 if not iproute2_is_vrf_capable():
285 pytest
.skip("Installed iproute2 version does not support VRFs")
289 if tgen
.routers_have_failure():
290 pytest
.skip("skipped because of router(s) failure")
292 # Verify Linux Kernel Routing Table
293 logger
.info("Verifying Linux IPv6 Kernel Routing Table")
297 # Get a list of all current link-local addresses first as they change for
298 # each run and we need to translate them
300 for i
in range(1, 5):
301 linklocals
+= tgen
.net
["r{}".format(i
)].get_ipv6_linklocal()
303 # Now compare the routing tables (after substituting link-local addresses)
305 for i
in range(1, 5):
306 # Actual output from router
308 tgen
.gears
["r{}".format(i
)]
309 .run("ip -6 route show vrf r{}-cust1".format(i
))
313 refTableFile
= os
.path
.join(CWD
, "r{}/ip_6_address.nhg.ref".format(i
))
315 refTableFile
= os
.path
.join(CWD
, "r{}/ip_6_address.ref".format(i
))
317 if os
.path
.isfile(refTableFile
):
318 expected
= open(refTableFile
).read().rstrip()
319 # Fix newlines (make them all the same)
320 expected
= ("\n".join(expected
.splitlines())).splitlines(1)
322 # Mask out Link-Local mac addresses
323 for ll
in linklocals
:
324 actual
= actual
.replace(ll
[1], "fe80::__(%s)__" % ll
[0])
325 # Mask out protocol name or number
326 actual
= re
.sub(r
"[ ]+proto [0-9a-z]+ +", " proto XXXX ", actual
)
327 actual
= re
.sub(r
"[ ]+nhid [0-9]+ +", " nhid XXXX ", actual
)
328 # Remove ff00::/8 routes (seen on some kernels - not from FRR)
329 actual
= re
.sub(r
"ff00::/8.*", "", actual
)
332 actual
= actual
.lstrip()
333 actual
= actual
.rstrip()
334 actual
= re
.sub(r
" +", " ", actual
)
337 for line
in sorted(actual
.splitlines()):
338 if line
.startswith("fe80::/64 ") or line
.startswith(
339 "unreachable fe80::/64 "
342 if "anycast" in line
:
344 if "multicast" in line
:
346 filtered_lines
.append(line
)
347 actual
= "\n".join(filtered_lines
).splitlines(1)
350 # logger.info("Router r%s table" % i)
351 # for line in actual:
352 # logger.info(line.rstrip())
355 diff
= topotest
.get_textdiff(
358 title1
="actual OSPFv3 IPv6 routing table",
359 title2
="expected OSPFv3 IPv6 routing table",
362 # Empty string if it matches, otherwise diff contains unified diff
365 "r%s failed Linux IPv6 Kernel Routing Table Check:\n%s\n"
370 logger
.info("r%s ok" % i
)
372 assert failures
== 0, (
373 "Linux Kernel IPv6 Routing Table verification failed for router r%s:\n%s"
378 def test_ospfv3_routingTable_write_multiplier():
381 if tgen
.routers_have_failure():
382 pytest
.skip("skipped because of router(s) failure")
384 # For debugging, uncomment the next line
386 # Modify R1 write muliplier and reset the interfaces
387 r1
= tgen
.gears
["r1"]
389 r1
.vtysh_cmd("conf t\nrouter ospf6 vrf r1-cust1 \n write-multiplier 100")
390 r1
.vtysh_cmd("clear ipv6 ospf interface r1-stubnet")
391 r1
.vtysh_cmd("clear ipv6 ospf interface r1-sw5")
393 # Verify OSPFv3 Routing Table
394 for router
, rnode
in tgen
.routers().items():
395 logger
.info('Waiting for router "%s" convergence', router
)
397 # Load expected results from the command
398 reffile
= os
.path
.join(CWD
, "{}/show_ipv6_vrf_route.ref".format(router
))
399 expected
= open(reffile
).read()
401 # Run test function until we get an result. Wait at most 60 seconds.
402 test_func
= partial(compare_show_ipv6_vrf
, router
, expected
)
403 result
, diff
= topotest
.run_and_expect(test_func
, "", count
=120, wait
=0.5)
404 assert result
, "OSPFv3 did not converge on {}:\n{}".format(router
, diff
)
407 def test_shutdown_check_stderr():
411 if tgen
.routers_have_failure():
412 pytest
.skip("skipped because of router(s) failure")
414 if os
.environ
.get("TOPOTESTS_CHECK_STDERR") is None:
416 "SKIPPED final check on StdErr output: Disabled (TOPOTESTS_CHECK_STDERR undefined)\n"
418 pytest
.skip("Skipping test for Stderr output")
422 logger
.info("\n\n** Verifying unexpected STDERR output from daemons")
423 logger
.info("******************************************")
425 for i
in range(1, 5):
426 net
["r%s" % i
].stopRouter()
427 log
= net
["r%s" % i
].getStdErr("ospf6d")
429 logger
.info("\nRouter r%s OSPF6d StdErr Log:\n%s" % (i
, log
))
430 log
= net
["r%s" % i
].getStdErr("zebra")
432 logger
.info("\nRouter r%s Zebra StdErr Log:\n%s" % (i
, log
))
435 def test_shutdown_check_memleak():
436 "Run the memory leak test and report results."
438 if os
.environ
.get("TOPOTESTS_CHECK_MEMLEAK") is None:
440 "SKIPPED final check on Memory leaks: Disabled (TOPOTESTS_CHECK_MEMLEAK undefined)"
442 pytest
.skip("Skipping test for memory leaks")
448 for i
in range(1, 5):
449 net
["r%s" % i
].stopRouter()
450 net
["r%s" % i
].report_memory_leaks(
451 os
.environ
.get("TOPOTESTS_CHECK_MEMLEAK"), os
.path
.basename(__file__
)
455 if __name__
== "__main__":
457 # To suppress tracebacks, either use the following pytest call or
458 # add "--tb=no" to cli
459 # retval = pytest.main(["-s", "--tb=no"])
461 retval
= pytest
.main(["-s"])