tests/topotests/isis_topo1/test_isis_topo1.py

   1 #!/usr/bin/env python
   2
   3 #
   4 # test_isis_topo1.py
   5 # Part of NetDEF Topology Tests
   6 #
   7 # Copyright (c) 2017 by
   8 # Network Device Education Foundation, Inc. ("NetDEF")
   9 #
  10 # Permission to use, copy, modify, and/or distribute this software
  11 # for any purpose with or without fee is hereby granted, provided
  12 # that the above copyright notice and this permission notice appear
  13 # in all copies.
  14 #
  15 # THE SOFTWARE IS PROVIDED "AS IS" AND NETDEF DISCLAIMS ALL WARRANTIES
  16 # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
  17 # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NETDEF BE LIABLE FOR
  18 # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
  19 # DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
  20 # WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
  21 # ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
  22 # OF THIS SOFTWARE.
  23 #
  24
  25 """
  26 test_isis_topo1.py: Test ISIS topology.
  27 """
  28
  29 import functools
  30 import json
  31 import os
  32 import re
  33 import sys
  34 import pytest
  35
  36 CWD = os.path.dirname(os.path.realpath(__file__))
  37 sys.path.append(os.path.join(CWD, "../"))
  38
  39 # pylint: disable=C0413
  40 from lib import topotest
  41 from lib.topogen import Topogen, TopoRouter, get_topogen
  42 from lib.topolog import logger
  43
  44
  45 pytestmark = [pytest.mark.isisd]
  46
  47 VERTEX_TYPE_LIST = [
  48     "pseudo_IS",
  49     "pseudo_TE-IS",
  50     "IS",
  51     "TE-IS",
  52     "ES",
  53     "IP internal",
  54     "IP external",
  55     "IP TE",
  56     "IP6 internal",
  57     "IP6 external",
  58     "UNKNOWN",
  59 ]
  60
  61
  62 def build_topo(tgen):
  63     "Build function"
  64
  65     # Add ISIS routers:
  66     # r1      r2
  67     #  | sw1  | sw2
  68     # r3     r4
  69     #  |      |
  70     # sw3    sw4
  71     #   \    /
  72     #     r5
  73     for routern in range(1, 6):
  74         tgen.add_router("r{}".format(routern))
  75
  76     # r1 <- sw1 -> r3
  77     sw = tgen.add_switch("sw1")
  78     sw.add_link(tgen.gears["r1"])
  79     sw.add_link(tgen.gears["r3"])
  80
  81     # r2 <- sw2 -> r4
  82     sw = tgen.add_switch("sw2")
  83     sw.add_link(tgen.gears["r2"])
  84     sw.add_link(tgen.gears["r4"])
  85
  86     # r3 <- sw3 -> r5
  87     sw = tgen.add_switch("sw3")
  88     sw.add_link(tgen.gears["r3"])
  89     sw.add_link(tgen.gears["r5"])
  90
  91     # r4 <- sw4 -> r5
  92     sw = tgen.add_switch("sw4")
  93     sw.add_link(tgen.gears["r4"])
  94     sw.add_link(tgen.gears["r5"])
  95
  96
  97 def setup_module(mod):
  98     "Sets up the pytest environment"
  99     tgen = Topogen(build_topo, mod.__name__)
 100     tgen.start_topology()
 101
 102     # For all registered routers, load the zebra configuration file
 103     for rname, router in tgen.routers().items():
 104         router.load_config(
 105             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
 106         )
 107         router.load_config(
 108             TopoRouter.RD_ISIS, os.path.join(CWD, "{}/isisd.conf".format(rname))
 109         )
 110
 111     # After loading the configurations, this function loads configured daemons.
 112     tgen.start_router()
 113
 114
 115 def teardown_module(mod):
 116     "Teardown the pytest environment"
 117     tgen = get_topogen()
 118
 119     # This function tears down the whole topology.
 120     tgen.stop_topology()
 121
 122
 123 def test_isis_convergence():
 124     "Wait for the protocol to converge before starting to test"
 125     tgen = get_topogen()
 126     # Don't run this test if we have any failure.
 127     if tgen.routers_have_failure():
 128         pytest.skip(tgen.errors)
 129
 130     logger.info("waiting for ISIS protocol to converge")
 131     # Code to generate the json files.
 132     # for rname, router in tgen.routers().items():
 133     #     open('/tmp/{}_topology.json'.format(rname), 'w').write(
 134     #         json.dumps(show_isis_topology(router), indent=2, sort_keys=True)
 135     #     )
 136
 137     for rname, router in tgen.routers().items():
 138         filename = "{0}/{1}/{1}_topology.json".format(CWD, rname)
 139         expected = json.loads(open(filename).read())
 140
 141         def compare_isis_topology(router, expected):
 142             "Helper function to test ISIS topology convergence."
 143             actual = show_isis_topology(router)
 144             return topotest.json_cmp(actual, expected)
 145
 146         test_func = functools.partial(compare_isis_topology, router, expected)
 147         (result, diff) = topotest.run_and_expect(test_func, None, wait=0.5, count=120)
 148         assert result, "ISIS did not converge on {}:\n{}".format(rname, diff)
 149
 150
 151 def test_isis_route_installation():
 152     "Check whether all expected routes are present"
 153     tgen = get_topogen()
 154     # Don't run this test if we have any failure.
 155     if tgen.routers_have_failure():
 156         pytest.skip(tgen.errors)
 157
 158     logger.info("Checking routers for installed ISIS routes")
 159
 160     # Check for routes in 'show ip route json'
 161     for rname, router in tgen.routers().items():
 162         filename = "{0}/{1}/{1}_route.json".format(CWD, rname)
 163         expected = json.loads(open(filename, "r").read())
 164
 165         def compare_isis_installed_routes(router, expected):
 166             "Helper function to test ISIS routes installed in rib."
 167             actual = router.vtysh_cmd("show ip route json", isjson=True)
 168             return topotest.json_cmp(actual, expected)
 169
 170         test_func = functools.partial(compare_isis_installed_routes, router, expected)
 171         (result, diff) = topotest.run_and_expect(test_func, None, wait=1, count=10)
 172         assertmsg = "Router '{}' routes mismatch".format(rname)
 173         assert result, assertmsg
 174
 175
 176 def test_isis_linux_route_installation():
 177     "Check whether all expected routes are present and installed in the OS"
 178     tgen = get_topogen()
 179     # Don't run this test if we have any failure.
 180     if tgen.routers_have_failure():
 181         pytest.skip(tgen.errors)
 182
 183     logger.info("Checking routers for installed ISIS routes in OS")
 184
 185     # Check for routes in `ip route`
 186     for rname, router in tgen.routers().items():
 187         filename = "{0}/{1}/{1}_route_linux.json".format(CWD, rname)
 188         expected = json.loads(open(filename, "r").read())
 189         actual = topotest.ip4_route(router)
 190         assertmsg = "Router '{}' OS routes mismatch".format(rname)
 191         assert topotest.json_cmp(actual, expected) is None, assertmsg
 192
 193
 194 def test_isis_route6_installation():
 195     "Check whether all expected routes are present"
 196     tgen = get_topogen()
 197     # Don't run this test if we have any failure.
 198     if tgen.routers_have_failure():
 199         pytest.skip(tgen.errors)
 200
 201     logger.info("Checking routers for installed ISIS IPv6 routes")
 202
 203     # Check for routes in 'show ip route json'
 204     for rname, router in tgen.routers().items():
 205         filename = "{0}/{1}/{1}_route6.json".format(CWD, rname)
 206         expected = json.loads(open(filename, "r").read())
 207
 208         def compare_isis_v6_installed_routes(router, expected):
 209             "Helper function to test ISIS v6 routes installed in rib."
 210             actual = router.vtysh_cmd("show ipv6 route json", isjson=True)
 211             return topotest.json_cmp(actual, expected)
 212
 213         test_func = functools.partial(
 214             compare_isis_v6_installed_routes, router, expected
 215         )
 216         (result, diff) = topotest.run_and_expect(test_func, None, wait=1, count=10)
 217         assertmsg = "Router '{}' routes mismatch".format(rname)
 218         assert result, assertmsg
 219
 220
 221 def test_isis_linux_route6_installation():
 222     "Check whether all expected routes are present and installed in the OS"
 223     tgen = get_topogen()
 224     # Don't run this test if we have any failure.
 225     if tgen.routers_have_failure():
 226         pytest.skip(tgen.errors)
 227
 228     logger.info("Checking routers for installed ISIS IPv6 routes in OS")
 229
 230     # Check for routes in `ip route`
 231     for rname, router in tgen.routers().items():
 232         filename = "{0}/{1}/{1}_route6_linux.json".format(CWD, rname)
 233         expected = json.loads(open(filename, "r").read())
 234         actual = topotest.ip6_route(router)
 235         assertmsg = "Router '{}' OS routes mismatch".format(rname)
 236         assert topotest.json_cmp(actual, expected) is None, assertmsg
 237
 238
 239 def test_isis_summary_json():
 240     "Check json struct in show isis summary json"
 241
 242     tgen = get_topogen()
 243     # Don't run this test if we have any failure.
 244     if tgen.routers_have_failure():
 245         pytest.skip(tgen.errors)
 246
 247     logger.info("Checking 'show isis summary json'")
 248     for rname, router in tgen.routers().items():
 249         logger.info("Checking router %s", rname)
 250         json_output = tgen.gears[rname].vtysh_cmd("show isis summary json", isjson=True)
 251         assertmsg = "Test isis summary json failed in '{}' data '{}'".format(rname, json_output)
 252         assert json_output['vrf'] == "default", assertmsg
 253         assert json_output['areas'][0]['area'] == "1", assertmsg
 254         assert json_output['areas'][0]['levels'][0]['id'] != '3', assertmsg
 255
 256
 257 def test_isis_interface_json():
 258     "Check json struct in show isis interface json"
 259
 260     tgen = get_topogen()
 261     # Don't run this test if we have any failure.
 262     if tgen.routers_have_failure():
 263         pytest.skip(tgen.errors)
 264
 265     logger.info("Checking 'show isis interface json'")
 266     for rname, router in tgen.routers().items():
 267         logger.info("Checking router %s", rname)
 268         json_output = tgen.gears[rname].vtysh_cmd("show isis interface json", isjson=True)
 269         assertmsg = "Test isis interface json failed in '{}' data '{}'".format(rname, json_output)
 270         assert json_output['areas'][0]['circuits'][0]['interface']['name'] == rname+"-eth0", assertmsg
 271
 272     for rname, router in tgen.routers().items():
 273         logger.info("Checking router %s", rname)
 274         json_output = tgen.gears[rname].vtysh_cmd("show isis interface detail json", isjson=True)
 275         assertmsg = "Test isis interface json failed in '{}' data '{}'".format(rname, json_output)
 276         assert json_output['areas'][0]['circuits'][0]['interface']['name'] == rname+"-eth0", assertmsg
 277
 278
 279 def test_isis_neighbor_json():
 280     "Check json struct in show isis neighbor json"
 281
 282     tgen = get_topogen()
 283     # Don't run this test if we have any failure.
 284     if tgen.routers_have_failure():
 285         pytest.skip(tgen.errors)
 286
 287     #tgen.mininet_cli()
 288     logger.info("Checking 'show isis neighbor json'")
 289     for rname, router in tgen.routers().items():
 290         logger.info("Checking router %s", rname)
 291         json_output = tgen.gears[rname].vtysh_cmd("show isis neighbor json", isjson=True)
 292         assertmsg = "Test isis neighbor json failed in '{}' data '{}'".format(rname, json_output)
 293         assert json_output['areas'][0]['circuits'][0]['interface'] == rname+"-eth0", assertmsg
 294
 295     for rname, router in tgen.routers().items():
 296         logger.info("Checking router %s", rname)
 297         json_output = tgen.gears[rname].vtysh_cmd("show isis neighbor detail json", isjson=True)
 298         assertmsg = "Test isis neighbor json failed in '{}' data '{}'".format(rname, json_output)
 299         assert json_output['areas'][0]['circuits'][0]['interface']['name'] == rname+"-eth0", assertmsg
 300
 301
 302 def test_isis_database_json():
 303     "Check json struct in show isis database json"
 304
 305     tgen = get_topogen()
 306     # Don't run this test if we have any failure.
 307     if tgen.routers_have_failure():
 308         pytest.skip(tgen.errors)
 309
 310     #tgen.mininet_cli()
 311     logger.info("Checking 'show isis database json'")
 312     for rname, router in tgen.routers().items():
 313         logger.info("Checking router %s", rname)
 314         json_output = tgen.gears[rname].vtysh_cmd("show isis database json", isjson=True)
 315         assertmsg = "Test isis database json failed in '{}' data '{}'".format(rname, json_output)
 316         assert json_output['areas'][0]['area']['name'] == "1", assertmsg
 317         assert json_output['areas'][0]['levels'][0]['id'] != '3', assertmsg
 318
 319     for rname, router in tgen.routers().items():
 320         logger.info("Checking router %s", rname)
 321         json_output = tgen.gears[rname].vtysh_cmd("show isis database detail json", isjson=True)
 322         assertmsg = "Test isis database json failed in '{}' data '{}'".format(rname, json_output)
 323         assert json_output['areas'][0]['area']['name'] == "1", assertmsg
 324         assert json_output['areas'][0]['levels'][0]['id'] != '3', assertmsg
 325
 326
 327 def test_memory_leak():
 328     "Run the memory leak test and report results."
 329     tgen = get_topogen()
 330     if not tgen.is_memleak_enabled():
 331         pytest.skip("Memory leak test/report is disabled")
 332
 333     tgen.report_memory_leaks()
 334
 335
 336 if __name__ == "__main__":
 337     args = ["-s"] + sys.argv[1:]
 338     sys.exit(pytest.main(args))
 339
 340
 341 #
 342 # Auxiliary functions
 343 #
 344
 345
 346 def dict_merge(dct, merge_dct):
 347     """
 348     Recursive dict merge. Inspired by :meth:``dict.update()``, instead of
 349     updating only top-level keys, dict_merge recurses down into dicts nested
 350     to an arbitrary depth, updating keys. The ``merge_dct`` is merged into
 351     ``dct``.
 352     :param dct: dict onto which the merge is executed
 353     :param merge_dct: dct merged into dct
 354     :return: None
 355
 356     Source:
 357     https://gist.github.com/angstwad/bf22d1822c38a92ec0a9
 358     """
 359     for k, v in merge_dct.items():
 360         if k in dct and isinstance(dct[k], dict) and topotest.is_mapping(merge_dct[k]):
 361             dict_merge(dct[k], merge_dct[k])
 362         else:
 363             dct[k] = merge_dct[k]
 364
 365
 366 def parse_topology(lines, level):
 367     """
 368     Parse the output of 'show isis topology level-X' into a Python dict.
 369     """
 370     areas = {}
 371     area = None
 372     ipv = None
 373     vertex_type_regex = "|".join(VERTEX_TYPE_LIST)
 374
 375     for line in lines:
 376         area_match = re.match(r"Area (.+):", line)
 377         if area_match:
 378             area = area_match.group(1)
 379             if area not in areas:
 380                 areas[area] = {level: {"ipv4": [], "ipv6": []}}
 381             ipv = None
 382             continue
 383         elif area is None:
 384             continue
 385
 386         if re.match(r"IS\-IS paths to level-. routers that speak IPv6", line):
 387             ipv = "ipv6"
 388             continue
 389         if re.match(r"IS\-IS paths to level-. routers that speak IP", line):
 390             ipv = "ipv4"
 391             continue
 392
 393         item_match = re.match(
 394             r"([^\s]+) ([^\s]+) ([^\s]+) ([^\s]+) ([^\s]+) ([^\s]+)", line
 395         )
 396         if (
 397             item_match is not None
 398             and item_match.group(1) == "Vertex"
 399             and item_match.group(2) == "Type"
 400             and item_match.group(3) == "Metric"
 401             and item_match.group(4) == "Next-Hop"
 402             and item_match.group(5) == "Interface"
 403             and item_match.group(6) == "Parent"
 404         ):
 405             # Skip header
 406             continue
 407
 408         item_match = re.match(
 409             r"([^\s]+) ({}) ([0]|([1-9][0-9]*)) ([^\s]+) ([^\s]+) ([^\s]+)".format(
 410                 vertex_type_regex
 411             ),
 412             line,
 413         )
 414         if item_match is not None:
 415             areas[area][level][ipv].append(
 416                 {
 417                     "vertex": item_match.group(1),
 418                     "type": item_match.group(2),
 419                     "metric": item_match.group(3),
 420                     "next-hop": item_match.group(5),
 421                     "interface": item_match.group(6),
 422                     "parent": item_match.group(7),
 423                 }
 424             )
 425             continue
 426
 427         item_match = re.match(
 428             r"([^\s]+) ({}) ([0]|([1-9][0-9]*)) ([^\s]+)".format(vertex_type_regex),
 429             line,
 430         )
 431
 432         if item_match is not None:
 433             areas[area][level][ipv].append(
 434                 {
 435                     "vertex": item_match.group(1),
 436                     "type": item_match.group(2),
 437                     "metric": item_match.group(3),
 438                     "parent": item_match.group(5),
 439                 }
 440             )
 441             continue
 442
 443         item_match = re.match(r"([^\s]+)", line)
 444         if item_match is not None:
 445             areas[area][level][ipv].append({"vertex": item_match.group(1)})
 446             continue
 447
 448     return areas
 449
 450
 451 def show_isis_topology(router):
 452     """
 453     Get the ISIS topology in a dictionary format.
 454
 455     Sample:
 456     {
 457       'area-name': {
 458         'level-1': [
 459           {
 460             'vertex': 'r1'
 461           }
 462         ],
 463         'level-2': [
 464           {
 465             'vertex': '10.0.0.1/24',
 466             'type': 'IP',
 467             'parent': '0',
 468             'metric': 'internal'
 469           }
 470         ]
 471       },
 472       'area-name-2': {
 473         'level-2': [
 474           {
 475             "interface": "rX-ethY",
 476             "metric": "Z",
 477             "next-hop": "rA",
 478             "parent": "rC(B)",
 479             "type": "TE-IS",
 480             "vertex": "rD"
 481           }
 482         ]
 483       }
 484     }
 485     """
 486     l1out = topotest.normalize_text(
 487         router.vtysh_cmd("show isis topology level-1")
 488     ).splitlines()
 489     l2out = topotest.normalize_text(
 490         router.vtysh_cmd("show isis topology level-2")
 491     ).splitlines()
 492
 493     l1 = parse_topology(l1out, "level-1")
 494     l2 = parse_topology(l2out, "level-2")
 495
 496     dict_merge(l1, l2)
 497     return l1