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