4 # test_ldp_isis_topo1.py
5 # Part of NetDEF Topology Tests
7 # Copyright (c) 2020 by Volta Networks
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
25 test_ldp_vpls_topo1.py:
27 +---------+ +---------+
31 +---------+ +---------+
32 ce1-eth0 (172.16.1.1/24)| |ce2-eth0 (172.16.1.2/24)
36 +---------+ 10.0.1.0/24 +---------+
38 | RT1 +----------------+ RT2 |
39 | 1.1.1.1 | rt2-eth1| 2.2.2.2 |
41 +---------+ +---------+
45 10.0.2.0/24| +---------+ |10.0.3.0/24
48 +--------+ 3.3.3.3 +-------+
54 ce3-eth0 (172.16.1.3/24)|
67 from functools
import partial
69 # Save the Current Working Directory to find configuration files.
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
79 # Required to instantiate the topology builder class.
81 pytestmark
= [pytest
.mark
.isisd
, pytest
.mark
.ldpd
]
90 for router
in ["ce1", "ce2", "ce3", "r1", "r2", "r3"]:
91 tgen
.add_router(router
)
96 switch
= tgen
.add_switch("s1")
97 switch
.add_link(tgen
.gears
["ce1"])
98 switch
.add_link(tgen
.gears
["r1"])
100 switch
= tgen
.add_switch("s2")
101 switch
.add_link(tgen
.gears
["ce2"])
102 switch
.add_link(tgen
.gears
["r2"])
104 switch
= tgen
.add_switch("s3")
105 switch
.add_link(tgen
.gears
["ce3"])
106 switch
.add_link(tgen
.gears
["r3"])
108 switch
= tgen
.add_switch("s4")
109 switch
.add_link(tgen
.gears
["r1"])
110 switch
.add_link(tgen
.gears
["r2"])
112 switch
= tgen
.add_switch("s5")
113 switch
.add_link(tgen
.gears
["r1"])
114 switch
.add_link(tgen
.gears
["r3"])
116 switch
= tgen
.add_switch("s6")
117 switch
.add_link(tgen
.gears
["r2"])
118 switch
.add_link(tgen
.gears
["r3"])
121 def setup_module(mod
):
122 "Sets up the pytest environment"
123 tgen
= Topogen(build_topo
, mod
.__name
__)
124 tgen
.start_topology()
126 router_list
= tgen
.routers()
128 # For all registered routers, load the zebra configuration file
129 for rname
, router
in router_list
.items():
131 TopoRouter
.RD_ZEBRA
, os
.path
.join(CWD
, "{}/zebra.conf".format(rname
))
133 # Don't start isisd and ldpd in the CE nodes
134 if router
.name
[0] == "r":
136 TopoRouter
.RD_ISIS
, os
.path
.join(CWD
, "{}/isisd.conf".format(rname
))
139 TopoRouter
.RD_LDP
, os
.path
.join(CWD
, "{}/ldpd.conf".format(rname
))
145 def teardown_module(mod
):
146 "Teardown the pytest environment"
149 # This function tears down the whole topology.
153 def router_compare_json_output(rname
, command
, reference
):
154 "Compare router JSON output"
156 logger
.info('Comparing router "%s" "%s" output', rname
, command
)
159 filename
= "{}/{}/{}".format(CWD
, rname
, reference
)
160 expected
= json
.loads(open(filename
).read())
162 # Run test function until we get an result.
163 test_func
= partial(topotest
.router_json_cmp
, tgen
.gears
[rname
], command
, expected
)
164 _
, diff
= topotest
.run_and_expect(test_func
, None, count
=320, wait
=0.5)
165 assertmsg
= '"{}" JSON output mismatches the expected result'.format(rname
)
166 assert diff
is None, assertmsg
169 def test_isis_convergence():
170 logger
.info("Test: check ISIS adjacencies")
173 # Skip if previous fatal error condition is raised
174 if tgen
.routers_have_failure():
175 pytest
.skip(tgen
.errors
)
177 for rname
in ["r1", "r2", "r3"]:
178 router_compare_json_output(
180 "show yang operational-data /frr-interface:lib isisd",
181 "show_yang_interface_isis_adjacencies.ref",
186 logger
.info("Test: verify RIB")
189 # Skip if previous fatal error condition is raised
190 if tgen
.routers_have_failure():
191 pytest
.skip(tgen
.errors
)
193 for rname
in ["r1", "r2", "r3"]:
194 router_compare_json_output(rname
, "show ip route json", "show_ip_route.ref")
197 def test_ldp_adjacencies():
198 logger
.info("Test: verify LDP adjacencies")
201 # Skip if previous fatal error condition is raised
202 if tgen
.routers_have_failure():
203 pytest
.skip(tgen
.errors
)
205 for rname
in ["r1", "r2", "r3"]:
206 router_compare_json_output(
207 rname
, "show mpls ldp discovery json", "show_ldp_discovery.ref"
211 def test_ldp_neighbors():
212 logger
.info("Test: verify LDP neighbors")
215 # Skip if previous fatal error condition is raised
216 if tgen
.routers_have_failure():
217 pytest
.skip(tgen
.errors
)
219 for rname
in ["r1", "r2", "r3"]:
220 router_compare_json_output(
221 rname
, "show mpls ldp neighbor json", "show_ldp_neighbor.ref"
225 def test_ldp_bindings():
226 logger
.info("Test: verify LDP bindings")
229 # Skip if previous fatal error condition is raised
230 if tgen
.routers_have_failure():
231 pytest
.skip(tgen
.errors
)
233 for rname
in ["r1", "r2", "r3"]:
234 router_compare_json_output(
235 rname
, "show mpls ldp binding json", "show_ldp_binding.ref"
239 def test_ldp_pwid_bindings():
240 logger
.info("Test: verify LDP PW-ID bindings")
243 # Skip if previous fatal error condition is raised
244 if tgen
.routers_have_failure():
245 pytest
.skip(tgen
.errors
)
247 for rname
in ["r1", "r2", "r3"]:
248 router_compare_json_output(
249 rname
, "show l2vpn atom binding json", "show_l2vpn_binding.ref"
253 def test_ldp_pseudowires():
254 logger
.info("Test: verify LDP pseudowires")
257 # Skip if previous fatal error condition is raised
258 if tgen
.routers_have_failure():
259 pytest
.skip(tgen
.errors
)
261 for rname
in ["r1", "r2", "r3"]:
262 router_compare_json_output(
263 rname
, "show l2vpn atom vc json", "show_l2vpn_vc.ref"
267 def test_ldp_igp_sync():
268 logger
.info("Test: verify LDP igp-sync")
271 # Skip if previous fatal error condition is raised
272 if tgen
.routers_have_failure():
273 pytest
.skip(tgen
.errors
)
275 for rname
in ["r1", "r2", "r3"]:
276 router_compare_json_output(
277 rname
, "show mpls ldp igp-sync json", "show_ldp_igp_sync.ref"
281 def test_isis_ldp_sync():
282 logger
.info("Test: verify ISIS igp-sync")
285 # Skip if previous fatal error condition is raised
286 if tgen
.routers_have_failure():
287 pytest
.skip(tgen
.errors
)
289 for rname
in ["r1", "r2", "r3"]:
290 (result
, diff
) = validate_show_isis_ldp_sync(rname
, "show_isis_ldp_sync.ref")
291 assert result
, "ISIS did not converge on {}:\n{}".format(rname
, diff
)
293 for rname
in ["r1", "r2", "r3"]:
294 (result
, diff
) = validate_show_isis_interface_detail(
295 rname
, "show_isis_interface_detail.ref"
297 assert result
, "ISIS interface did not converge on {}:\n{}".format(rname
, diff
)
300 def test_r1_eth1_shutdown():
301 logger
.info("Test: verify behaviour after r1-eth1 is shutdown")
304 # Skip if previous fatal error condition is raised
305 if tgen
.routers_have_failure():
306 pytest
.skip(tgen
.errors
)
308 # Shut down r1-r2 link */
310 tgen
.gears
["r1"].peer_link_enable("r1-eth1", False)
311 topotest
.sleep(5, "Waiting for the network to reconverge")
313 # check if the pseudowire is still up (using an alternate path for nexthop resolution)
314 for rname
in ["r1", "r2", "r3"]:
315 router_compare_json_output(
316 rname
, "show l2vpn atom vc json", "show_l2vpn_vc.ref"
319 for rname
in ["r1", "r2", "r3"]:
320 router_compare_json_output(
322 "show mpls ldp igp-sync json",
323 "show_ldp_igp_sync_r1_eth1_shutdown.ref",
326 for rname
in ["r1", "r2", "r3"]:
327 (result
, diff
) = validate_show_isis_ldp_sync(
328 rname
, "show_isis_ldp_sync_r1_eth1_shutdown.ref"
330 assert result
, "ISIS did not converge on {}:\n{}".format(rname
, diff
)
332 for rname
in ["r1", "r2", "r3"]:
333 (result
, diff
) = validate_show_isis_interface_detail(
334 rname
, "show_isis_interface_detail_r1_eth1_shutdown.ref"
336 assert result
, "ISIS interface did not converge on {}:\n{}".format(rname
, diff
)
339 def test_r1_eth1_no_shutdown():
340 logger
.info("Test: verify behaviour after r1-eth1 is no shutdown")
343 # Skip if previous fatal error condition is raised
344 if tgen
.routers_have_failure():
345 pytest
.skip(tgen
.errors
)
347 # Run no shutdown on r1-eth1 interface */
349 tgen
.gears
["r1"].peer_link_enable("r1-eth1", True)
350 topotest
.sleep(5, "Waiting for the network to reconverge")
352 for rname
in ["r1", "r2", "r3"]:
353 router_compare_json_output(
354 rname
, "show mpls ldp igp-sync json", "show_ldp_igp_sync.ref"
357 for rname
in ["r1", "r2", "r3"]:
358 (result
, diff
) = validate_show_isis_ldp_sync(rname
, "show_isis_ldp_sync.ref")
359 assert result
, "ISIS did not converge on {}:\n{}".format(rname
, diff
)
361 for rname
in ["r1", "r2", "r3"]:
362 (result
, diff
) = validate_show_isis_interface_detail(
363 rname
, "show_isis_interface_detail.ref"
365 assert result
, "ISIS interface did not converge on {}:\n{}".format(rname
, diff
)
368 def test_r2_eth1_shutdown():
369 logger
.info("Test: verify behaviour after r2-eth1 is shutdown")
372 # Skip if previous fatal error condition is raised
373 if tgen
.routers_have_failure():
374 pytest
.skip(tgen
.errors
)
376 # Shut down r1-r2 link */
378 tgen
.gears
["r2"].peer_link_enable("r2-eth1", False)
379 topotest
.sleep(5, "Waiting for the network to reconverge")
381 for rname
in ["r1", "r2", "r3"]:
382 router_compare_json_output(
384 "show mpls ldp igp-sync json",
385 "show_ldp_igp_sync_r1_eth1_shutdown.ref",
388 for rname
in ["r1", "r2", "r3"]:
389 (result
, diff
) = validate_show_isis_ldp_sync(
390 rname
, "show_isis_ldp_sync_r2_eth1_shutdown.ref"
392 assert result
, "ISIS did not converge on {}:\n{}".format(rname
, diff
)
394 for rname
in ["r1", "r2", "r3"]:
395 (result
, diff
) = validate_show_isis_interface_detail(
396 rname
, "show_isis_interface_detail_r2_eth1_shutdown.ref"
398 assert result
, "ISIS interface did not converge on {}:\n{}".format(rname
, diff
)
401 def test_r2_eth1_no_shutdown():
402 logger
.info("Test: verify behaviour after r2-eth1 is no shutdown")
405 # Skip if previous fatal error condition is raised
406 if tgen
.routers_have_failure():
407 pytest
.skip(tgen
.errors
)
409 # Run no shutdown on r2-eth1 interface */
411 tgen
.gears
["r2"].peer_link_enable("r2-eth1", True)
412 topotest
.sleep(5, "Waiting for the network to reconverge")
414 for rname
in ["r1", "r2", "r3"]:
415 router_compare_json_output(
416 rname
, "show mpls ldp igp-sync json", "show_ldp_igp_sync.ref"
419 for rname
in ["r1", "r2", "r3"]:
420 (result
, diff
) = validate_show_isis_ldp_sync(rname
, "show_isis_ldp_sync.ref")
421 assert result
, "ISIS did not converge on {}:\n{}".format(rname
, diff
)
423 for rname
in ["r1", "r2", "r3"]:
424 (result
, diff
) = validate_show_isis_interface_detail(
425 rname
, "show_isis_interface_detail.ref"
427 assert result
, "ISIS interface did not converge on {}:\n{}".format(rname
, diff
)
430 # Memory leak test template
431 def test_memory_leak():
432 "Run the memory leak test and report results."
434 if not tgen
.is_memleak_enabled():
435 pytest
.skip("Memory leak test/report is disabled")
437 tgen
.report_memory_leaks()
440 if __name__
== "__main__":
441 args
= ["-s"] + sys
.argv
[1:]
442 sys
.exit(pytest
.main(args
))
446 # Auxiliary functions
450 def parse_show_isis_ldp_sync(lines
, rname
):
452 Parse the output of 'show isis mpls ldp sync' into a Python dict.
461 interface_name
= None
465 if line
.startswith(rname
+ "-eth"):
466 interface_name
= line
470 if line
.startswith(" LDP-IGP Synchronization enabled: "):
471 interface
["ldpIgpSyncEnabled"] = line
.endswith("yes")
474 if line
.startswith(" holddown timer in seconds: "):
475 interface
["holdDownTimeInSec"] = int(line
.split(": ")[-1])
478 if line
.startswith(" State: "):
479 interface
["ldpIgpSyncState"] = line
.split(": ")[-1]
481 elif line
.startswith(" Interface "):
482 interface
["Interface"] = line
.endswith("down")
484 interfaces
[interface_name
] = interface
486 except StopIteration:
492 def show_isis_ldp_sync(router
, rname
):
494 Get the show isis mpls ldp-sync info in a dictionary format.
497 out
= topotest
.normalize_text(
498 router
.vtysh_cmd("show isis mpls ldp-sync")
501 parsed
= parse_show_isis_ldp_sync(out
, rname
)
506 def validate_show_isis_ldp_sync(rname
, fname
):
509 filename
= "{0}/{1}/{2}".format(CWD
, rname
, fname
)
510 expected
= json
.loads(open(filename
).read())
512 router
= tgen
.gears
[rname
]
514 def compare_isis_ldp_sync(router
, expected
):
515 "Helper function to test show isis mpls ldp-sync"
516 actual
= show_isis_ldp_sync(router
, rname
)
517 return topotest
.json_cmp(actual
, expected
)
519 test_func
= partial(compare_isis_ldp_sync
, router
, expected
)
520 (result
, diff
) = topotest
.run_and_expect(test_func
, None, wait
=0.5, count
=160)
522 return (result
, diff
)
525 def parse_show_isis_interface_detail(lines
, rname
):
527 Parse the output of 'show isis interface detail' into a Python dict.
538 area_match
= re
.match(r
"Area (.+):", line
)
542 area_id
= area_match
.group(1)
547 while line
.startswith(" Interface: "):
548 interface_name
= re
.split(":|,", line
)[1].lstrip()
550 area
[interface_name
] = []
552 # Look for keyword: Level-1 or Level-2
553 while not line
.startswith(" Level-"):
556 while line
.startswith(" Level-"):
560 level_name
= line
.split()[0]
561 level
["level"] = level_name
565 if line
.startswith(" Metric:"):
566 level
["metric"] = re
.split(":|,", line
)[1].lstrip()
568 area
[interface_name
].append(level
)
570 # Look for keyword: Level-1 or Level-2 or Interface:
571 while not line
.startswith(" Level-") and not line
.startswith(
576 if line
.startswith(" Level-"):
579 if line
.startswith(" Interface: "):
582 areas
[area_id
] = area
584 except StopIteration:
586 areas
[area_id
] = area
592 def show_isis_interface_detail(router
, rname
):
594 Get the show isis mpls ldp-sync info in a dictionary format.
597 out
= topotest
.normalize_text(
598 router
.vtysh_cmd("show isis interface detail")
603 parsed
= parse_show_isis_interface_detail(out
, rname
)
605 logger
.warning(parsed
)
610 def validate_show_isis_interface_detail(rname
, fname
):
613 filename
= "{0}/{1}/{2}".format(CWD
, rname
, fname
)
614 expected
= json
.loads(open(filename
).read())
616 router
= tgen
.gears
[rname
]
618 def compare_isis_interface_detail(router
, expected
):
619 "Helper function to test show isis interface detail"
620 actual
= show_isis_interface_detail(router
, rname
)
621 return topotest
.json_cmp(actual
, expected
)
623 test_func
= partial(compare_isis_interface_detail
, router
, expected
)
624 (result
, diff
) = topotest
.run_and_expect(test_func
, None, wait
=0.5, count
=160)
626 return (result
, diff
)