4 # test_ospf6_topo1_vrf.py
5 # Part of NetDEF Topology Tests
7 # Copyright (c) 2021 by Niral Networks, Inc. ("Niral Networks")
8 # Used Copyright (c) 2016 by Network Device Education Foundation,
9 # Inc. ("NetDEF") in this file.
11 # Permission to use, copy, modify, and/or distribute this software
12 # for any purpose with or without fee is hereby granted, provided
13 # that the above copyright notice and this permission notice appear
16 # THE SOFTWARE IS PROVIDED "AS IS" AND NETDEF DISCLAIMS ALL WARRANTIES
17 # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
18 # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NETDEF BE LIABLE FOR
19 # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
20 # DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
21 # WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
22 # ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
27 test_ospf6_topo1_vrf.py:
30 SW1 - Stub Net 1 SW2 - Stub Net 2 \
31 fc00:1:1:1::/64 fc00:2:2:2::/64 \
32 \___________________/ \___________________/ |
36 +---------+---------+ +---------+---------+ |
38 | FRRouting | | FRRouting | |
39 | Rtr-ID: 10.0.0.1 | | Rtr-ID: 10.0.0.2 | |
40 +---------+---------+ +---------+---------+ |
42 \______ ___________/ OSPFv3
48 ~~ fc00:A:A:A::/64 ~~ |
51 | ::3 | SW3 - Stub Net 3 |
52 +---------+---------+ /-+ fc00:3:3:3::/64 |
54 | FRRouting +--/ \---- /
55 | Rtr-ID: 10.0.0.3 | ::3 ___________/
56 +---------+---------+ \
62 ~~ fc00:B:B:B::/64 ~~ \
63 ~~~~~~~~~~~~~~~~~~ OSPFv3
66 +---------+---------+ /---- |
67 | R4 | | SW4 - Stub Net 4 |
68 | FRRouting +------+ fc00:4:4:4::/64 |
69 | Rtr-ID: 10.0.0.4 | ::4 | /
70 +-------------------+ \---- /
79 from time
import sleep
81 from functools
import partial
83 from mininet
.topo
import Topo
85 # Save the Current Working Directory to find configuration files later.
86 CWD
= os
.path
.dirname(os
.path
.realpath(__file__
))
87 sys
.path
.append(os
.path
.join(CWD
, "../"))
89 # pylint: disable=C0413
90 # Import topogen and topotest helpers
91 from lib
import topotest
92 from lib
.topogen
import Topogen
, TopoRouter
, get_topogen
93 from lib
.topolog
import logger
94 from lib
.topotest
import iproute2_is_vrf_capable
95 from lib
.common_config
import required_linux_kernel_version
97 pytestmark
= [pytest
.mark
.ospfd
]
100 #####################################################
102 ## Network Topology Definition
104 #####################################################
107 class NetworkTopo(Topo
):
108 "OSPFv3 (IPv6) Test Topology 1"
110 def build(self
, **_opts
):
113 tgen
= get_topogen(self
)
116 for routern
in range(1, 5):
117 tgen
.add_router("r{}".format(routern
))
120 # Wire up the switches and routers
121 # Note that we specify the link names so we match the config files
124 # Create a empty network for router 1
125 switch
= tgen
.add_switch("s1")
126 switch
.add_link(tgen
.gears
["r1"], nodeif
="r1-stubnet")
128 # Create a empty network for router 2
129 switch
= tgen
.add_switch("s2")
130 switch
.add_link(tgen
.gears
["r2"], nodeif
="r2-stubnet")
132 # Create a empty network for router 3
133 switch
= tgen
.add_switch("s3")
134 switch
.add_link(tgen
.gears
["r3"], nodeif
="r3-stubnet")
136 # Create a empty network for router 4
137 switch
= tgen
.add_switch("s4")
138 switch
.add_link(tgen
.gears
["r4"], nodeif
="r4-stubnet")
140 # Interconnect routers 1, 2, and 3
141 switch
= tgen
.add_switch("s5")
142 switch
.add_link(tgen
.gears
["r1"], nodeif
="r1-sw5")
143 switch
.add_link(tgen
.gears
["r2"], nodeif
="r2-sw5")
144 switch
.add_link(tgen
.gears
["r3"], nodeif
="r3-sw5")
146 # Interconnect routers 3 and 4
147 switch
= tgen
.add_switch("s6")
148 switch
.add_link(tgen
.gears
["r3"], nodeif
="r3-sw6")
149 switch
.add_link(tgen
.gears
["r4"], nodeif
="r4-sw6")
152 #####################################################
156 #####################################################
159 def setup_module(mod
):
160 "Sets up the pytest environment"
162 # Required linux kernel version for this suite to run.
163 result
= required_linux_kernel_version("5.0")
164 if result
is not True:
165 pytest
.skip("Kernel requirements are not met")
167 tgen
= Topogen(NetworkTopo
, mod
.__name
__)
168 tgen
.start_topology()
170 logger
.info("** %s: Setup Topology" % mod
.__name
__)
171 logger
.info("******************************************")
173 # For debugging after starting net, but before starting FRR,
174 # uncomment the next line
177 logger
.info("Testing with VRF Lite support")
180 "ip link add {0}-cust1 type vrf table 1001",
181 "ip link add loop1 type dummy",
182 "ip link set {0}-stubnet master {0}-cust1",
185 cmds1
= ["ip link set {0}-sw5 master {0}-cust1"]
187 cmds2
= ["ip link set {0}-sw6 master {0}-cust1"]
189 # For all registered routers, load the zebra configuration file
190 for rname
, router
in tgen
.routers().items():
191 # create VRF rx-cust1 and link rx-eth0 to rx-cust1
193 output
= tgen
.net
[rname
].cmd(cmd
.format(rname
))
194 if rname
== "r1" or rname
== "r2" or rname
== "r3":
196 output
= tgen
.net
[rname
].cmd(cmd
.format(rname
))
197 if rname
== "r3" or rname
== "r4":
199 output
= tgen
.net
[rname
].cmd(cmd
.format(rname
))
201 for rname
, router
in tgen
.routers().items():
203 TopoRouter
.RD_ZEBRA
, os
.path
.join(CWD
, "{}/zebra.conf".format(rname
))
206 TopoRouter
.RD_OSPF6
, os
.path
.join(CWD
, "{}/ospf6d.conf".format(rname
))
209 # Initialize all routers.
212 # For debugging after starting FRR daemons, uncomment the next line
216 def teardown_module(mod
):
217 "Teardown the pytest environment"
222 def test_wait_protocol_convergence():
223 "Wait for OSPFv3 to converge"
225 if tgen
.routers_have_failure():
226 pytest
.skip(tgen
.errors
)
228 logger
.info("waiting for protocols to converge")
230 def expect_neighbor_full(router
, neighbor
):
231 "Wait until OSPFv3 convergence."
232 logger
.info("waiting OSPFv3 router '{}'".format(router
))
234 topotest
.router_json_cmp
,
236 "show ipv6 ospf6 vrf {0}-cust1 neighbor json".format(router
),
237 {"neighbors": [{"neighborId": neighbor
, "state": "Full"}]},
239 _
, result
= topotest
.run_and_expect(test_func
, None, count
=130, wait
=1)
240 assertmsg
= '"{}" convergence failure'.format(router
)
241 assert result
is None, assertmsg
243 expect_neighbor_full("r1", "10.0.0.2")
244 expect_neighbor_full("r1", "10.0.0.3")
246 expect_neighbor_full("r2", "10.0.0.1")
247 expect_neighbor_full("r2", "10.0.0.3")
249 expect_neighbor_full("r3", "10.0.0.1")
250 expect_neighbor_full("r3", "10.0.0.2")
251 expect_neighbor_full("r3", "10.0.0.4")
254 def compare_show_ipv6_vrf(rname
, expected
):
256 Calls 'show ipv6 route' for router `rname` and compare the obtained
257 result with the expected output.
261 # Use the vtysh output, with some masking to make comparison easy
262 vrf_name
= "{0}-cust1".format(rname
)
263 current
= topotest
.ip6_route_zebra(tgen
.gears
[rname
], vrf_name
)
265 # Use just the 'O'spf lines of the output
267 for line
in current
.splitlines():
268 if re
.match("^O", line
):
271 current
= "\n".join(linearr
)
273 return topotest
.difflines(
274 topotest
.normalize_text(current
),
275 topotest
.normalize_text(expected
),
276 title1
="Current output",
277 title2
="Expected output",
281 def test_ospfv3_routingTable():
284 if tgen
.routers_have_failure():
285 pytest
.skip("skipped because of router(s) failure")
287 # For debugging, uncomment the next line
289 # Verify OSPFv3 Routing Table
290 for router
, rnode
in tgen
.routers().iteritems():
291 logger
.info('Waiting for router "%s" convergence', router
)
293 # Load expected results from the command
294 reffile
= os
.path
.join(CWD
, "{}/show_ipv6_vrf_route.ref".format(router
))
295 expected
= open(reffile
).read()
297 # Run test function until we get an result. Wait at most 60 seconds.
298 test_func
= partial(compare_show_ipv6_vrf
, router
, expected
)
299 result
, diff
= topotest
.run_and_expect(test_func
, "", count
=120, wait
=0.5)
300 assert result
, "OSPFv3 did not converge on {}:\n{}".format(router
, diff
)
303 def test_linux_ipv6_kernel_routingTable():
305 # Required linux kernel version for this suite to run.
306 result
= required_linux_kernel_version("4.15")
307 if result
is not True:
308 pytest
.skip("Kernel requirements are not met")
310 # iproute2 needs to support VRFs for this suite to run.
311 if not iproute2_is_vrf_capable():
312 pytest
.skip("Installed iproute2 version does not support VRFs")
316 if tgen
.routers_have_failure():
317 pytest
.skip("skipped because of router(s) failure")
319 # Verify Linux Kernel Routing Table
320 logger
.info("Verifying Linux IPv6 Kernel Routing Table")
324 # Get a list of all current link-local addresses first as they change for
325 # each run and we need to translate them
327 for i
in range(1, 5):
328 linklocals
+= tgen
.net
["r{}".format(i
)].get_ipv6_linklocal()
330 # Now compare the routing tables (after substituting link-local addresses)
332 for i
in range(1, 5):
333 # Actual output from router
335 tgen
.gears
["r{}".format(i
)]
336 .run("ip -6 route show vrf r{}-cust1".format(i
))
340 refTableFile
= os
.path
.join(CWD
, "r{}/ip_6_address.nhg.ref".format(i
))
342 refTableFile
= os
.path
.join(CWD
, "r{}/ip_6_address.ref".format(i
))
344 if os
.path
.isfile(refTableFile
):
345 expected
= open(refTableFile
).read().rstrip()
346 # Fix newlines (make them all the same)
347 expected
= ("\n".join(expected
.splitlines())).splitlines(1)
349 # Mask out Link-Local mac addresses
350 for ll
in linklocals
:
351 actual
= actual
.replace(ll
[1], "fe80::__(%s)__" % ll
[0])
352 # Mask out protocol name or number
353 actual
= re
.sub(r
"[ ]+proto [0-9a-z]+ +", " proto XXXX ", actual
)
354 actual
= re
.sub(r
"[ ]+nhid [0-9]+ +", " nhid XXXX ", actual
)
355 # Remove ff00::/8 routes (seen on some kernels - not from FRR)
356 actual
= re
.sub(r
"ff00::/8.*", "", actual
)
359 actual
= actual
.lstrip()
360 actual
= actual
.rstrip()
361 actual
= re
.sub(r
" +", " ", actual
)
364 for line
in sorted(actual
.splitlines()):
365 if line
.startswith("fe80::/64 ") or line
.startswith(
366 "unreachable fe80::/64 "
369 if "anycast" in line
:
371 if "multicast" in line
:
373 filtered_lines
.append(line
)
374 actual
= "\n".join(filtered_lines
).splitlines(1)
377 # logger.info("Router r%s table" % i)
378 # for line in actual:
379 # logger.info(line.rstrip())
382 diff
= topotest
.get_textdiff(
385 title1
="actual OSPFv3 IPv6 routing table",
386 title2
="expected OSPFv3 IPv6 routing table",
389 # Empty string if it matches, otherwise diff contains unified diff
392 "r%s failed Linux IPv6 Kernel Routing Table Check:\n%s\n"
397 logger
.info("r%s ok" % i
)
399 assert failures
== 0, (
400 "Linux Kernel IPv6 Routing Table verification failed for router r%s:\n%s"
405 def test_ospfv3_routingTable_write_multiplier():
408 if tgen
.routers_have_failure():
409 pytest
.skip("skipped because of router(s) failure")
411 # For debugging, uncomment the next line
413 # Modify R1 write muliplier and reset the interfaces
414 r1
= tgen
.gears
["r1"]
416 r1
.vtysh_cmd("conf t\nrouter ospf6 vrf r1-cust1 \n write-multiplier 100")
417 r1
.vtysh_cmd("clear ipv6 ospf interface r1-stubnet")
418 r1
.vtysh_cmd("clear ipv6 ospf interface r1-sw5")
420 # Verify OSPFv3 Routing Table
421 for router
, rnode
in tgen
.routers().iteritems():
422 logger
.info('Waiting for router "%s" convergence', router
)
424 # Load expected results from the command
425 reffile
= os
.path
.join(CWD
, "{}/show_ipv6_vrf_route.ref".format(router
))
426 expected
= open(reffile
).read()
428 # Run test function until we get an result. Wait at most 60 seconds.
429 test_func
= partial(compare_show_ipv6_vrf
, router
, expected
)
430 result
, diff
= topotest
.run_and_expect(test_func
, "", count
=120, wait
=0.5)
431 assert result
, "OSPFv3 did not converge on {}:\n{}".format(router
, diff
)
434 def test_shutdown_check_stderr():
438 if tgen
.routers_have_failure():
439 pytest
.skip("skipped because of router(s) failure")
441 if os
.environ
.get("TOPOTESTS_CHECK_STDERR") is None:
443 "SKIPPED final check on StdErr output: Disabled (TOPOTESTS_CHECK_STDERR undefined)\n"
445 pytest
.skip("Skipping test for Stderr output")
449 logger
.info("\n\n** Verifying unexpected STDERR output from daemons")
450 logger
.info("******************************************")
452 for i
in range(1, 5):
453 net
["r%s" % i
].stopRouter()
454 log
= net
["r%s" % i
].getStdErr("ospf6d")
456 logger
.info("\nRouter r%s OSPF6d StdErr Log:\n%s" % (i
, log
))
457 log
= net
["r%s" % i
].getStdErr("zebra")
459 logger
.info("\nRouter r%s Zebra StdErr Log:\n%s" % (i
, log
))
462 def test_shutdown_check_memleak():
463 "Run the memory leak test and report results."
465 if os
.environ
.get("TOPOTESTS_CHECK_MEMLEAK") is None:
467 "SKIPPED final check on Memory leaks: Disabled (TOPOTESTS_CHECK_MEMLEAK undefined)"
469 pytest
.skip("Skipping test for memory leaks")
475 for i
in range(1, 5):
476 net
["r%s" % i
].stopRouter()
477 net
["r%s" % i
].report_memory_leaks(
478 os
.environ
.get("TOPOTESTS_CHECK_MEMLEAK"), os
.path
.basename(__file__
)
482 if __name__
== "__main__":
484 # To suppress tracebacks, either use the following pytest call or
485 # add "--tb=no" to cli
486 # retval = pytest.main(["-s", "--tb=no"])
488 retval
= pytest
.main(["-s"])