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