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 time
import sleep
68 from functools
import partial
70 # Save the Current Working Directory to find configuration files.
71 CWD
= os
.path
.dirname(os
.path
.realpath(__file__
))
72 sys
.path
.append(os
.path
.join(CWD
, "../"))
74 # pylint: disable=C0413
75 # Import topogen and topotest helpers
76 from lib
import topotest
77 from lib
.topogen
import Topogen
, TopoRouter
, get_topogen
78 from lib
.topolog
import logger
80 # Required to instantiate the topology builder class.
81 from mininet
.topo
import Topo
84 class TemplateTopo(Topo
):
85 "Test topology builder"
87 def build(self
, *_args
, **_opts
):
89 tgen
= get_topogen(self
)
94 for router
in ["ce1", "ce2", "ce3", "r1", "r2", "r3"]:
95 tgen
.add_router(router
)
100 switch
= tgen
.add_switch("s1")
101 switch
.add_link(tgen
.gears
["ce1"])
102 switch
.add_link(tgen
.gears
["r1"])
104 switch
= tgen
.add_switch("s2")
105 switch
.add_link(tgen
.gears
["ce2"])
106 switch
.add_link(tgen
.gears
["r2"])
108 switch
= tgen
.add_switch("s3")
109 switch
.add_link(tgen
.gears
["ce3"])
110 switch
.add_link(tgen
.gears
["r3"])
112 switch
= tgen
.add_switch("s4")
113 switch
.add_link(tgen
.gears
["r1"])
114 switch
.add_link(tgen
.gears
["r2"])
116 switch
= tgen
.add_switch("s5")
117 switch
.add_link(tgen
.gears
["r1"])
118 switch
.add_link(tgen
.gears
["r3"])
120 switch
= tgen
.add_switch("s6")
121 switch
.add_link(tgen
.gears
["r2"])
122 switch
.add_link(tgen
.gears
["r3"])
125 def setup_module(mod
):
126 "Sets up the pytest environment"
127 tgen
= Topogen(TemplateTopo
, mod
.__name
__)
128 tgen
.start_topology()
130 router_list
= tgen
.routers()
132 # For all registered routers, load the zebra configuration file
133 for rname
, router
in router_list
.items():
135 TopoRouter
.RD_ZEBRA
, os
.path
.join(CWD
, "{}/zebra.conf".format(rname
))
137 # Don't start isisd and ldpd in the CE nodes
138 if router
.name
[0] == "r":
140 TopoRouter
.RD_ISIS
, os
.path
.join(CWD
, "{}/isisd.conf".format(rname
))
143 TopoRouter
.RD_LDP
, os
.path
.join(CWD
, "{}/ldpd.conf".format(rname
))
149 def teardown_module(mod
):
150 "Teardown the pytest environment"
153 # This function tears down the whole topology.
157 def router_compare_json_output(rname
, command
, reference
):
158 "Compare router JSON output"
160 logger
.info('Comparing router "%s" "%s" output', rname
, command
)
163 filename
= "{}/{}/{}".format(CWD
, rname
, reference
)
164 expected
= json
.loads(open(filename
).read())
166 # Run test function until we get an result.
167 test_func
= partial(topotest
.router_json_cmp
, tgen
.gears
[rname
], command
, expected
)
168 _
, diff
= topotest
.run_and_expect(test_func
, None, count
=320, wait
=0.5)
169 assertmsg
= '"{}" JSON output mismatches the expected result'.format(rname
)
170 assert diff
is None, assertmsg
173 def test_isis_convergence():
174 logger
.info("Test: check ISIS adjacencies")
177 # Skip if previous fatal error condition is raised
178 if tgen
.routers_have_failure():
179 pytest
.skip(tgen
.errors
)
181 for rname
in ["r1", "r2", "r3"]:
182 router_compare_json_output(
184 "show yang operational-data /frr-interface:lib isisd",
185 "show_yang_interface_isis_adjacencies.ref",
190 logger
.info("Test: verify RIB")
193 # Skip if previous fatal error condition is raised
194 if tgen
.routers_have_failure():
195 pytest
.skip(tgen
.errors
)
197 for rname
in ["r1", "r2", "r3"]:
198 router_compare_json_output(rname
, "show ip route json", "show_ip_route.ref")
201 def test_ldp_adjacencies():
202 logger
.info("Test: verify LDP adjacencies")
205 # Skip if previous fatal error condition is raised
206 if tgen
.routers_have_failure():
207 pytest
.skip(tgen
.errors
)
209 for rname
in ["r1", "r2", "r3"]:
210 router_compare_json_output(
211 rname
, "show mpls ldp discovery json", "show_ldp_discovery.ref"
215 def test_ldp_neighbors():
216 logger
.info("Test: verify LDP neighbors")
219 # Skip if previous fatal error condition is raised
220 if tgen
.routers_have_failure():
221 pytest
.skip(tgen
.errors
)
223 for rname
in ["r1", "r2", "r3"]:
224 router_compare_json_output(
225 rname
, "show mpls ldp neighbor json", "show_ldp_neighbor.ref"
229 def test_ldp_bindings():
230 logger
.info("Test: verify LDP bindings")
233 # Skip if previous fatal error condition is raised
234 if tgen
.routers_have_failure():
235 pytest
.skip(tgen
.errors
)
237 for rname
in ["r1", "r2", "r3"]:
238 router_compare_json_output(
239 rname
, "show mpls ldp binding json", "show_ldp_binding.ref"
243 def test_ldp_pwid_bindings():
244 logger
.info("Test: verify LDP PW-ID bindings")
247 # Skip if previous fatal error condition is raised
248 if tgen
.routers_have_failure():
249 pytest
.skip(tgen
.errors
)
251 for rname
in ["r1", "r2", "r3"]:
252 router_compare_json_output(
253 rname
, "show l2vpn atom binding json", "show_l2vpn_binding.ref"
257 def test_ldp_pseudowires():
258 logger
.info("Test: verify LDP pseudowires")
261 # Skip if previous fatal error condition is raised
262 if tgen
.routers_have_failure():
263 pytest
.skip(tgen
.errors
)
265 for rname
in ["r1", "r2", "r3"]:
266 router_compare_json_output(
267 rname
, "show l2vpn atom vc json", "show_l2vpn_vc.ref"
271 def test_ldp_igp_sync():
272 logger
.info("Test: verify LDP igp-sync")
275 # Skip if previous fatal error condition is raised
276 if tgen
.routers_have_failure():
277 pytest
.skip(tgen
.errors
)
279 for rname
in ["r1", "r2", "r3"]:
280 router_compare_json_output(
281 rname
, "show mpls ldp igp-sync json", "show_ldp_igp_sync.ref"
285 def test_isis_ldp_sync():
286 logger
.info("Test: verify ISIS igp-sync")
289 # Skip if previous fatal error condition is raised
290 if tgen
.routers_have_failure():
291 pytest
.skip(tgen
.errors
)
293 for rname
in ["r1", "r2", "r3"]:
294 (result
, diff
) = validate_show_isis_ldp_sync(rname
, "show_isis_ldp_sync.ref")
295 assert result
, "ISIS did not converge on {}:\n{}".format(rname
, diff
)
297 for rname
in ["r1", "r2", "r3"]:
298 (result
, diff
) = validate_show_isis_interface_detail(
299 rname
, "show_isis_interface_detail.ref"
301 assert result
, "ISIS interface did not converge on {}:\n{}".format(rname
, diff
)
304 def test_r1_eth1_shutdown():
305 logger
.info("Test: verify behaviour after r1-eth1 is shutdown")
308 # Skip if previous fatal error condition is raised
309 if tgen
.routers_have_failure():
310 pytest
.skip(tgen
.errors
)
312 # Shut down r1-r2 link */
314 tgen
.gears
["r1"].peer_link_enable("r1-eth1", False)
315 topotest
.sleep(5, "Waiting for the network to reconverge")
317 # check if the pseudowire is still up (using an alternate path for nexthop resolution)
318 for rname
in ["r1", "r2", "r3"]:
319 router_compare_json_output(
320 rname
, "show l2vpn atom vc json", "show_l2vpn_vc.ref"
323 for rname
in ["r1", "r2", "r3"]:
324 router_compare_json_output(
326 "show mpls ldp igp-sync json",
327 "show_ldp_igp_sync_r1_eth1_shutdown.ref",
330 for rname
in ["r1", "r2", "r3"]:
331 (result
, diff
) = validate_show_isis_ldp_sync(
332 rname
, "show_isis_ldp_sync_r1_eth1_shutdown.ref"
334 assert result
, "ISIS did not converge on {}:\n{}".format(rname
, diff
)
336 for rname
in ["r1", "r2", "r3"]:
337 (result
, diff
) = validate_show_isis_interface_detail(
338 rname
, "show_isis_interface_detail_r1_eth1_shutdown.ref"
340 assert result
, "ISIS interface did not converge on {}:\n{}".format(rname
, diff
)
343 def test_r1_eth1_no_shutdown():
344 logger
.info("Test: verify behaviour after r1-eth1 is no shutdown")
347 # Skip if previous fatal error condition is raised
348 if tgen
.routers_have_failure():
349 pytest
.skip(tgen
.errors
)
351 # Run no shutdown on r1-eth1 interface */
353 tgen
.gears
["r1"].peer_link_enable("r1-eth1", True)
354 topotest
.sleep(5, "Waiting for the network to reconverge")
356 for rname
in ["r1", "r2", "r3"]:
357 router_compare_json_output(
358 rname
, "show mpls ldp igp-sync json", "show_ldp_igp_sync.ref"
361 for rname
in ["r1", "r2", "r3"]:
362 (result
, diff
) = validate_show_isis_ldp_sync(rname
, "show_isis_ldp_sync.ref")
363 assert result
, "ISIS did not converge on {}:\n{}".format(rname
, diff
)
365 for rname
in ["r1", "r2", "r3"]:
366 (result
, diff
) = validate_show_isis_interface_detail(
367 rname
, "show_isis_interface_detail.ref"
369 assert result
, "ISIS interface did not converge on {}:\n{}".format(rname
, diff
)
372 def test_r2_eth1_shutdown():
373 logger
.info("Test: verify behaviour after r2-eth1 is shutdown")
376 # Skip if previous fatal error condition is raised
377 if tgen
.routers_have_failure():
378 pytest
.skip(tgen
.errors
)
380 # Shut down r1-r2 link */
382 tgen
.gears
["r2"].peer_link_enable("r2-eth1", False)
383 topotest
.sleep(5, "Waiting for the network to reconverge")
385 for rname
in ["r1", "r2", "r3"]:
386 router_compare_json_output(
388 "show mpls ldp igp-sync json",
389 "show_ldp_igp_sync_r1_eth1_shutdown.ref",
392 for rname
in ["r1", "r2", "r3"]:
393 (result
, diff
) = validate_show_isis_ldp_sync(
394 rname
, "show_isis_ldp_sync_r2_eth1_shutdown.ref"
396 assert result
, "ISIS did not converge on {}:\n{}".format(rname
, diff
)
398 for rname
in ["r1", "r2", "r3"]:
399 (result
, diff
) = validate_show_isis_interface_detail(
400 rname
, "show_isis_interface_detail_r2_eth1_shutdown.ref"
402 assert result
, "ISIS interface did not converge on {}:\n{}".format(rname
, diff
)
405 def test_r2_eth1_no_shutdown():
406 logger
.info("Test: verify behaviour after r2-eth1 is no shutdown")
409 # Skip if previous fatal error condition is raised
410 if tgen
.routers_have_failure():
411 pytest
.skip(tgen
.errors
)
413 # Run no shutdown on r2-eth1 interface */
415 tgen
.gears
["r2"].peer_link_enable("r2-eth1", True)
416 topotest
.sleep(5, "Waiting for the network to reconverge")
418 for rname
in ["r1", "r2", "r3"]:
419 router_compare_json_output(
420 rname
, "show mpls ldp igp-sync json", "show_ldp_igp_sync.ref"
423 for rname
in ["r1", "r2", "r3"]:
424 (result
, diff
) = validate_show_isis_ldp_sync(rname
, "show_isis_ldp_sync.ref")
425 assert result
, "ISIS did not converge on {}:\n{}".format(rname
, diff
)
427 for rname
in ["r1", "r2", "r3"]:
428 (result
, diff
) = validate_show_isis_interface_detail(
429 rname
, "show_isis_interface_detail.ref"
431 assert result
, "ISIS interface did not converge on {}:\n{}".format(rname
, diff
)
434 # Memory leak test template
435 def test_memory_leak():
436 "Run the memory leak test and report results."
438 if not tgen
.is_memleak_enabled():
439 pytest
.skip("Memory leak test/report is disabled")
441 tgen
.report_memory_leaks()
444 if __name__
== "__main__":
445 args
= ["-s"] + sys
.argv
[1:]
446 sys
.exit(pytest
.main(args
))
450 # Auxiliary functions
454 def parse_show_isis_ldp_sync(lines
, rname
):
456 Parse the output of 'show isis mpls ldp sync' into a Python dict.
465 interface_name
= None
469 if line
.startswith(rname
+ "-eth"):
470 interface_name
= line
474 if line
.startswith(" LDP-IGP Synchronization enabled: "):
475 interface
["ldpIgpSyncEnabled"] = line
.endswith("yes")
478 if line
.startswith(" holddown timer in seconds: "):
479 interface
["holdDownTimeInSec"] = int(line
.split(": ")[-1])
482 if line
.startswith(" State: "):
483 interface
["ldpIgpSyncState"] = line
.split(": ")[-1]
485 elif line
.startswith(" Interface "):
486 interface
["Interface"] = line
.endswith("down")
488 interfaces
[interface_name
] = interface
490 except StopIteration:
496 def show_isis_ldp_sync(router
, rname
):
498 Get the show isis mpls ldp-sync info in a dictionary format.
501 out
= topotest
.normalize_text(
502 router
.vtysh_cmd("show isis mpls ldp-sync")
505 parsed
= parse_show_isis_ldp_sync(out
, rname
)
510 def validate_show_isis_ldp_sync(rname
, fname
):
513 filename
= "{0}/{1}/{2}".format(CWD
, rname
, fname
)
514 expected
= json
.loads(open(filename
).read())
516 router
= tgen
.gears
[rname
]
518 def compare_isis_ldp_sync(router
, expected
):
519 "Helper function to test show isis mpls ldp-sync"
520 actual
= show_isis_ldp_sync(router
, rname
)
521 return topotest
.json_cmp(actual
, expected
)
523 test_func
= partial(compare_isis_ldp_sync
, router
, expected
)
524 (result
, diff
) = topotest
.run_and_expect(test_func
, None, wait
=0.5, count
=160)
526 return (result
, diff
)
529 def parse_show_isis_interface_detail(lines
, rname
):
531 Parse the output of 'show isis interface detail' into a Python dict.
542 area_match
= re
.match(r
"Area (.+):", line
)
546 area_id
= area_match
.group(1)
551 while line
.startswith(" Interface: "):
552 interface_name
= re
.split(":|,", line
)[1].lstrip()
554 area
[interface_name
] = []
556 # Look for keyword: Level-1 or Level-2
557 while not line
.startswith(" Level-"):
560 while line
.startswith(" Level-"):
564 level_name
= line
.split()[0]
565 level
["level"] = level_name
569 if line
.startswith(" Metric:"):
570 level
["metric"] = re
.split(":|,", line
)[1].lstrip()
572 area
[interface_name
].append(level
)
574 # Look for keyword: Level-1 or Level-2 or Interface:
575 while not line
.startswith(" Level-") and not line
.startswith(
580 if line
.startswith(" Level-"):
583 if line
.startswith(" Interface: "):
586 areas
[area_id
] = area
588 except StopIteration:
590 areas
[area_id
] = area
596 def show_isis_interface_detail(router
, rname
):
598 Get the show isis mpls ldp-sync info in a dictionary format.
601 out
= topotest
.normalize_text(
602 router
.vtysh_cmd("show isis interface detail")
607 parsed
= parse_show_isis_interface_detail(out
, rname
)
609 logger
.warning(parsed
)
614 def validate_show_isis_interface_detail(rname
, fname
):
617 filename
= "{0}/{1}/{2}".format(CWD
, rname
, fname
)
618 expected
= json
.loads(open(filename
).read())
620 router
= tgen
.gears
[rname
]
622 def compare_isis_interface_detail(router
, expected
):
623 "Helper function to test show isis interface detail"
624 actual
= show_isis_interface_detail(router
, rname
)
625 return topotest
.json_cmp(actual
, expected
)
627 test_func
= partial(compare_isis_interface_detail
, router
, expected
)
628 (result
, diff
) = topotest
.run_and_expect(test_func
, None, wait
=0.5, count
=160)
630 return (result
, diff
)