tests/topotests/ldp_sync_isis_topo1/test_ldp_sync_isis_topo1.py

   1 #!/usr/bin/env python
   2
   3 #
   4 # test_ldp_isis_topo1.py
   5 # Part of NetDEF Topology Tests
   6 #
   7 # Copyright (c) 2020 by Volta Networks
   8 #
   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
  12 # in all copies.
  13 #
  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
  21 # OF THIS SOFTWARE.
  22 #
  23
  24 """
  25 test_ldp_vpls_topo1.py:
  26
  27                    +---------+                +---------+
  28                    |         |                |         |
  29                    |   CE1   |                |   CE2   |
  30                    |         |                |         |
  31                    +---------+                +---------+
  32 ce1-eth0 (172.16.1.1/24)|                          |ce2-eth0 (172.16.1.2/24)
  33                         |                          |
  34                         |                          |
  35                 rt1-eth0|                          |rt2-eth0
  36                    +---------+  10.0.1.0/24   +---------+
  37                    |         |rt1-eth1        |         |
  38                    |   RT1   +----------------+   RT2   |
  39                    | 1.1.1.1 |        rt2-eth1| 2.2.2.2 |
  40                    |         |                |         |
  41                    +---------+                +---------+
  42                 rt1-eth2|                          |rt2-eth2
  43                         |                          |
  44                         |                          |
  45              10.0.2.0/24|        +---------+       |10.0.3.0/24
  46                         |        |         |       |
  47                         |        |   RT3   |       |
  48                         +--------+ 3.3.3.3 +-------+
  49                          rt3-eth2|         |rt3-eth1
  50                                  +---------+
  51                                       |rt3-eth0
  52                                       |
  53                                       |
  54               ce3-eth0 (172.16.1.3/24)|
  55                                  +---------+
  56                                  |         |
  57                                  |   CE3   |
  58                                  |         |
  59                                  +---------+
  60 """
  61
  62 import os
  63 import re
  64 import sys
  65 import pytest
  66 import json
  67 from functools import partial
  68
  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, "../"))
  72
  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
  79 # Required to instantiate the topology builder class.
  80
  81 pytestmark = [pytest.mark.isisd, pytest.mark.ldpd]
  82
  83
  84 def build_topo(tgen):
  85     "Build function"
  86
  87     #
  88     # Define FRR Routers
  89     #
  90     for router in ["ce1", "ce2", "ce3", "r1", "r2", "r3"]:
  91         tgen.add_router(router)
  92
  93     #
  94     # Define connections
  95     #
  96     switch = tgen.add_switch("s1")
  97     switch.add_link(tgen.gears["ce1"])
  98     switch.add_link(tgen.gears["r1"])
  99
 100     switch = tgen.add_switch("s2")
 101     switch.add_link(tgen.gears["ce2"])
 102     switch.add_link(tgen.gears["r2"])
 103
 104     switch = tgen.add_switch("s3")
 105     switch.add_link(tgen.gears["ce3"])
 106     switch.add_link(tgen.gears["r3"])
 107
 108     switch = tgen.add_switch("s4")
 109     switch.add_link(tgen.gears["r1"])
 110     switch.add_link(tgen.gears["r2"])
 111
 112     switch = tgen.add_switch("s5")
 113     switch.add_link(tgen.gears["r1"])
 114     switch.add_link(tgen.gears["r3"])
 115
 116     switch = tgen.add_switch("s6")
 117     switch.add_link(tgen.gears["r2"])
 118     switch.add_link(tgen.gears["r3"])
 119
 120
 121 def setup_module(mod):
 122     "Sets up the pytest environment"
 123     tgen = Topogen(build_topo, mod.__name__)
 124     tgen.start_topology()
 125
 126     router_list = tgen.routers()
 127
 128     # For all registered routers, load the zebra configuration file
 129     for rname, router in router_list.items():
 130         router.load_config(
 131             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
 132         )
 133         # Don't start isisd and ldpd in the CE nodes
 134         if router.name[0] == "r":
 135             router.load_config(
 136                 TopoRouter.RD_ISIS, os.path.join(CWD, "{}/isisd.conf".format(rname))
 137             )
 138             router.load_config(
 139                 TopoRouter.RD_LDP, os.path.join(CWD, "{}/ldpd.conf".format(rname))
 140             )
 141
 142     tgen.start_router()
 143
 144
 145 def teardown_module(mod):
 146     "Teardown the pytest environment"
 147     tgen = get_topogen()
 148
 149     # This function tears down the whole topology.
 150     tgen.stop_topology()
 151
 152
 153 def router_compare_json_output(rname, command, reference):
 154     "Compare router JSON output"
 155
 156     logger.info('Comparing router "%s" "%s" output', rname, command)
 157
 158     tgen = get_topogen()
 159     filename = "{}/{}/{}".format(CWD, rname, reference)
 160     expected = json.loads(open(filename).read())
 161
 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
 167
 168
 169 def test_isis_convergence():
 170     logger.info("Test: check ISIS adjacencies")
 171     tgen = get_topogen()
 172
 173     # Skip if previous fatal error condition is raised
 174     if tgen.routers_have_failure():
 175         pytest.skip(tgen.errors)
 176
 177     for rname in ["r1", "r2", "r3"]:
 178         router_compare_json_output(
 179             rname,
 180             "show yang operational-data /frr-interface:lib isisd",
 181             "show_yang_interface_isis_adjacencies.ref",
 182         )
 183
 184
 185 def test_rib():
 186     logger.info("Test: verify RIB")
 187     tgen = get_topogen()
 188
 189     # Skip if previous fatal error condition is raised
 190     if tgen.routers_have_failure():
 191         pytest.skip(tgen.errors)
 192
 193     for rname in ["r1", "r2", "r3"]:
 194         router_compare_json_output(rname, "show ip route json", "show_ip_route.ref")
 195
 196
 197 def test_ldp_adjacencies():
 198     logger.info("Test: verify LDP adjacencies")
 199     tgen = get_topogen()
 200
 201     # Skip if previous fatal error condition is raised
 202     if tgen.routers_have_failure():
 203         pytest.skip(tgen.errors)
 204
 205     for rname in ["r1", "r2", "r3"]:
 206         router_compare_json_output(
 207             rname, "show mpls ldp discovery json", "show_ldp_discovery.ref"
 208         )
 209
 210
 211 def test_ldp_neighbors():
 212     logger.info("Test: verify LDP neighbors")
 213     tgen = get_topogen()
 214
 215     # Skip if previous fatal error condition is raised
 216     if tgen.routers_have_failure():
 217         pytest.skip(tgen.errors)
 218
 219     for rname in ["r1", "r2", "r3"]:
 220         router_compare_json_output(
 221             rname, "show mpls ldp neighbor json", "show_ldp_neighbor.ref"
 222         )
 223
 224
 225 def test_ldp_bindings():
 226     logger.info("Test: verify LDP bindings")
 227     tgen = get_topogen()
 228
 229     # Skip if previous fatal error condition is raised
 230     if tgen.routers_have_failure():
 231         pytest.skip(tgen.errors)
 232
 233     for rname in ["r1", "r2", "r3"]:
 234         router_compare_json_output(
 235             rname, "show mpls ldp binding json", "show_ldp_binding.ref"
 236         )
 237
 238
 239 def test_ldp_pwid_bindings():
 240     logger.info("Test: verify LDP PW-ID bindings")
 241     tgen = get_topogen()
 242
 243     # Skip if previous fatal error condition is raised
 244     if tgen.routers_have_failure():
 245         pytest.skip(tgen.errors)
 246
 247     for rname in ["r1", "r2", "r3"]:
 248         router_compare_json_output(
 249             rname, "show l2vpn atom binding json", "show_l2vpn_binding.ref"
 250         )
 251
 252
 253 def test_ldp_pseudowires():
 254     logger.info("Test: verify LDP pseudowires")
 255     tgen = get_topogen()
 256
 257     # Skip if previous fatal error condition is raised
 258     if tgen.routers_have_failure():
 259         pytest.skip(tgen.errors)
 260
 261     for rname in ["r1", "r2", "r3"]:
 262         router_compare_json_output(
 263             rname, "show l2vpn atom vc json", "show_l2vpn_vc.ref"
 264         )
 265
 266
 267 def test_ldp_igp_sync():
 268     logger.info("Test: verify LDP igp-sync")
 269     tgen = get_topogen()
 270
 271     # Skip if previous fatal error condition is raised
 272     if tgen.routers_have_failure():
 273         pytest.skip(tgen.errors)
 274
 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"
 278         )
 279
 280
 281 def test_isis_ldp_sync():
 282     logger.info("Test: verify ISIS igp-sync")
 283     tgen = get_topogen()
 284
 285     # Skip if previous fatal error condition is raised
 286     if tgen.routers_have_failure():
 287         pytest.skip(tgen.errors)
 288
 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)
 292
 293     for rname in ["r1", "r2", "r3"]:
 294         (result, diff) = validate_show_isis_interface_detail(
 295             rname, "show_isis_interface_detail.ref"
 296         )
 297         assert result, "ISIS interface did not converge on {}:\n{}".format(rname, diff)
 298
 299
 300 def test_r1_eth1_shutdown():
 301     logger.info("Test: verify behaviour after r1-eth1 is shutdown")
 302     tgen = get_topogen()
 303
 304     # Skip if previous fatal error condition is raised
 305     if tgen.routers_have_failure():
 306         pytest.skip(tgen.errors)
 307
 308     # Shut down r1-r2 link */
 309     tgen = get_topogen()
 310     tgen.gears["r1"].peer_link_enable("r1-eth1", False)
 311     topotest.sleep(5, "Waiting for the network to reconverge")
 312
 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"
 317         )
 318
 319     for rname in ["r1", "r2", "r3"]:
 320         router_compare_json_output(
 321             rname,
 322             "show mpls ldp igp-sync json",
 323             "show_ldp_igp_sync_r1_eth1_shutdown.ref",
 324         )
 325
 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"
 329         )
 330         assert result, "ISIS did not converge on {}:\n{}".format(rname, diff)
 331
 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"
 335         )
 336         assert result, "ISIS interface did not converge on {}:\n{}".format(rname, diff)
 337
 338
 339 def test_r1_eth1_no_shutdown():
 340     logger.info("Test: verify behaviour after r1-eth1 is no shutdown")
 341     tgen = get_topogen()
 342
 343     # Skip if previous fatal error condition is raised
 344     if tgen.routers_have_failure():
 345         pytest.skip(tgen.errors)
 346
 347     # Run no shutdown on r1-eth1 interface */
 348     tgen = get_topogen()
 349     tgen.gears["r1"].peer_link_enable("r1-eth1", True)
 350     topotest.sleep(5, "Waiting for the network to reconverge")
 351
 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"
 355         )
 356
 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)
 360
 361     for rname in ["r1", "r2", "r3"]:
 362         (result, diff) = validate_show_isis_interface_detail(
 363             rname, "show_isis_interface_detail.ref"
 364         )
 365         assert result, "ISIS interface did not converge on {}:\n{}".format(rname, diff)
 366
 367
 368 def test_r2_eth1_shutdown():
 369     logger.info("Test: verify behaviour after r2-eth1 is shutdown")
 370     tgen = get_topogen()
 371
 372     # Skip if previous fatal error condition is raised
 373     if tgen.routers_have_failure():
 374         pytest.skip(tgen.errors)
 375
 376     # Shut down r1-r2 link */
 377     tgen = get_topogen()
 378     tgen.gears["r2"].peer_link_enable("r2-eth1", False)
 379     topotest.sleep(5, "Waiting for the network to reconverge")
 380
 381     for rname in ["r1", "r2", "r3"]:
 382         router_compare_json_output(
 383             rname,
 384             "show mpls ldp igp-sync json",
 385             "show_ldp_igp_sync_r1_eth1_shutdown.ref",
 386         )
 387
 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"
 391         )
 392         assert result, "ISIS did not converge on {}:\n{}".format(rname, diff)
 393
 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"
 397         )
 398         assert result, "ISIS interface did not converge on {}:\n{}".format(rname, diff)
 399
 400
 401 def test_r2_eth1_no_shutdown():
 402     logger.info("Test: verify behaviour after r2-eth1 is no shutdown")
 403     tgen = get_topogen()
 404
 405     # Skip if previous fatal error condition is raised
 406     if tgen.routers_have_failure():
 407         pytest.skip(tgen.errors)
 408
 409     # Run no shutdown on r2-eth1 interface */
 410     tgen = get_topogen()
 411     tgen.gears["r2"].peer_link_enable("r2-eth1", True)
 412     topotest.sleep(5, "Waiting for the network to reconverge")
 413
 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"
 417         )
 418
 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)
 422
 423     for rname in ["r1", "r2", "r3"]:
 424         (result, diff) = validate_show_isis_interface_detail(
 425             rname, "show_isis_interface_detail.ref"
 426         )
 427         assert result, "ISIS interface did not converge on {}:\n{}".format(rname, diff)
 428
 429
 430 # Memory leak test template
 431 def test_memory_leak():
 432     "Run the memory leak test and report results."
 433     tgen = get_topogen()
 434     if not tgen.is_memleak_enabled():
 435         pytest.skip("Memory leak test/report is disabled")
 436
 437     tgen.report_memory_leaks()
 438
 439
 440 if __name__ == "__main__":
 441     args = ["-s"] + sys.argv[1:]
 442     sys.exit(pytest.main(args))
 443
 444
 445 #
 446 # Auxiliary functions
 447 #
 448
 449
 450 def parse_show_isis_ldp_sync(lines, rname):
 451     """
 452     Parse the output of 'show isis mpls ldp sync' into a Python dict.
 453     """
 454     interfaces = {}
 455
 456     it = iter(lines)
 457
 458     while True:
 459         try:
 460             interface = {}
 461             interface_name = None
 462
 463             line = next(it)
 464
 465             if line.startswith(rname + "-eth"):
 466                 interface_name = line
 467
 468             line = next(it)
 469
 470             if line.startswith(" LDP-IGP Synchronization enabled: "):
 471                 interface["ldpIgpSyncEnabled"] = line.endswith("yes")
 472                 line = next(it)
 473
 474                 if line.startswith(" holddown timer in seconds: "):
 475                     interface["holdDownTimeInSec"] = int(line.split(": ")[-1])
 476                     line = next(it)
 477
 478                 if line.startswith(" State: "):
 479                     interface["ldpIgpSyncState"] = line.split(": ")[-1]
 480
 481             elif line.startswith(" Interface "):
 482                 interface["Interface"] = line.endswith("down")
 483
 484             interfaces[interface_name] = interface
 485
 486         except StopIteration:
 487             break
 488
 489     return interfaces
 490
 491
 492 def show_isis_ldp_sync(router, rname):
 493     """
 494     Get the show isis mpls ldp-sync info in a dictionary format.
 495
 496     """
 497     out = topotest.normalize_text(
 498         router.vtysh_cmd("show isis mpls ldp-sync")
 499     ).splitlines()
 500
 501     parsed = parse_show_isis_ldp_sync(out, rname)
 502
 503     return parsed
 504
 505
 506 def validate_show_isis_ldp_sync(rname, fname):
 507     tgen = get_topogen()
 508
 509     filename = "{0}/{1}/{2}".format(CWD, rname, fname)
 510     expected = json.loads(open(filename).read())
 511
 512     router = tgen.gears[rname]
 513
 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)
 518
 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)
 521
 522     return (result, diff)
 523
 524
 525 def parse_show_isis_interface_detail(lines, rname):
 526     """
 527     Parse the output of 'show isis interface detail' into a Python dict.
 528     """
 529     areas = {}
 530     area_id = None
 531
 532     it = iter(lines)
 533
 534     while True:
 535         try:
 536             line = next(it)
 537
 538             area_match = re.match(r"Area (.+):", line)
 539             if not area_match:
 540                 continue
 541
 542             area_id = area_match.group(1)
 543             area = {}
 544
 545             line = next(it)
 546
 547             while line.startswith(" Interface: "):
 548                 interface_name = re.split(":|,", line)[1].lstrip()
 549
 550                 area[interface_name] = []
 551
 552                 # Look for keyword: Level-1 or Level-2
 553                 while not line.startswith(" Level-"):
 554                     line = next(it)
 555
 556                 while line.startswith(" Level-"):
 557
 558                     level = {}
 559
 560                     level_name = line.split()[0]
 561                     level["level"] = level_name
 562
 563                     line = next(it)
 564
 565                     if line.startswith(" Metric:"):
 566                         level["metric"] = re.split(":|,", line)[1].lstrip()
 567
 568                     area[interface_name].append(level)
 569
 570                     # Look for keyword: Level-1 or Level-2 or Interface:
 571                     while not line.startswith(" Level-") and not line.startswith(
 572                         " Interface: "
 573                     ):
 574                         line = next(it)
 575
 576                     if line.startswith(" Level-"):
 577                         continue
 578
 579                     if line.startswith(" Interface: "):
 580                         break
 581
 582             areas[area_id] = area
 583
 584         except StopIteration:
 585
 586             areas[area_id] = area
 587             break
 588
 589     return areas
 590
 591
 592 def show_isis_interface_detail(router, rname):
 593     """
 594     Get the show isis mpls ldp-sync info in a dictionary format.
 595
 596     """
 597     out = topotest.normalize_text(
 598         router.vtysh_cmd("show isis interface detail")
 599     ).splitlines()
 600
 601     logger.warning(out)
 602
 603     parsed = parse_show_isis_interface_detail(out, rname)
 604
 605     logger.warning(parsed)
 606
 607     return parsed
 608
 609
 610 def validate_show_isis_interface_detail(rname, fname):
 611     tgen = get_topogen()
 612
 613     filename = "{0}/{1}/{2}".format(CWD, rname, fname)
 614     expected = json.loads(open(filename).read())
 615
 616     router = tgen.gears[rname]
 617
 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)
 622
 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)
 625
 626     return (result, diff)