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_memory_leak():
 240     "Run the memory leak test and report results."
 241     tgen = get_topogen()
 242     if not tgen.is_memleak_enabled():
 243         pytest.skip("Memory leak test/report is disabled")
 244
 245     tgen.report_memory_leaks()
 246
 247
 248 if __name__ == "__main__":
 249     args = ["-s"] + sys.argv[1:]
 250     sys.exit(pytest.main(args))
 251
 252
 253 #
 254 # Auxiliary functions
 255 #
 256
 257
 258 def dict_merge(dct, merge_dct):
 259     """
 260     Recursive dict merge. Inspired by :meth:``dict.update()``, instead of
 261     updating only top-level keys, dict_merge recurses down into dicts nested
 262     to an arbitrary depth, updating keys. The ``merge_dct`` is merged into
 263     ``dct``.
 264     :param dct: dict onto which the merge is executed
 265     :param merge_dct: dct merged into dct
 266     :return: None
 267
 268     Source:
 269     https://gist.github.com/angstwad/bf22d1822c38a92ec0a9
 270     """
 271     for k, v in merge_dct.items():
 272         if k in dct and isinstance(dct[k], dict) and topotest.is_mapping(merge_dct[k]):
 273             dict_merge(dct[k], merge_dct[k])
 274         else:
 275             dct[k] = merge_dct[k]
 276
 277
 278 def parse_topology(lines, level):
 279     """
 280     Parse the output of 'show isis topology level-X' into a Python dict.
 281     """
 282     areas = {}
 283     area = None
 284     ipv = None
 285     vertex_type_regex = "|".join(VERTEX_TYPE_LIST)
 286
 287     for line in lines:
 288         area_match = re.match(r"Area (.+):", line)
 289         if area_match:
 290             area = area_match.group(1)
 291             if area not in areas:
 292                 areas[area] = {level: {"ipv4": [], "ipv6": []}}
 293             ipv = None
 294             continue
 295         elif area is None:
 296             continue
 297
 298         if re.match(r"IS\-IS paths to level-. routers that speak IPv6", line):
 299             ipv = "ipv6"
 300             continue
 301         if re.match(r"IS\-IS paths to level-. routers that speak IP", line):
 302             ipv = "ipv4"
 303             continue
 304
 305         item_match = re.match(
 306             r"([^\s]+) ([^\s]+) ([^\s]+) ([^\s]+) ([^\s]+) ([^\s]+)", line
 307         )
 308         if (
 309             item_match is not None
 310             and item_match.group(1) == "Vertex"
 311             and item_match.group(2) == "Type"
 312             and item_match.group(3) == "Metric"
 313             and item_match.group(4) == "Next-Hop"
 314             and item_match.group(5) == "Interface"
 315             and item_match.group(6) == "Parent"
 316         ):
 317             # Skip header
 318             continue
 319
 320         item_match = re.match(
 321             r"([^\s]+) ({}) ([0]|([1-9][0-9]*)) ([^\s]+) ([^\s]+) ([^\s]+)".format(
 322                 vertex_type_regex
 323             ),
 324             line,
 325         )
 326         if item_match is not None:
 327             areas[area][level][ipv].append(
 328                 {
 329                     "vertex": item_match.group(1),
 330                     "type": item_match.group(2),
 331                     "metric": item_match.group(3),
 332                     "next-hop": item_match.group(5),
 333                     "interface": item_match.group(6),
 334                     "parent": item_match.group(7),
 335                 }
 336             )
 337             continue
 338
 339         item_match = re.match(
 340             r"([^\s]+) ({}) ([0]|([1-9][0-9]*)) ([^\s]+)".format(vertex_type_regex),
 341             line,
 342         )
 343
 344         if item_match is not None:
 345             areas[area][level][ipv].append(
 346                 {
 347                     "vertex": item_match.group(1),
 348                     "type": item_match.group(2),
 349                     "metric": item_match.group(3),
 350                     "parent": item_match.group(5),
 351                 }
 352             )
 353             continue
 354
 355         item_match = re.match(r"([^\s]+)", line)
 356         if item_match is not None:
 357             areas[area][level][ipv].append({"vertex": item_match.group(1)})
 358             continue
 359
 360     return areas
 361
 362
 363 def show_isis_topology(router):
 364     """
 365     Get the ISIS topology in a dictionary format.
 366
 367     Sample:
 368     {
 369       'area-name': {
 370         'level-1': [
 371           {
 372             'vertex': 'r1'
 373           }
 374         ],
 375         'level-2': [
 376           {
 377             'vertex': '10.0.0.1/24',
 378             'type': 'IP',
 379             'parent': '0',
 380             'metric': 'internal'
 381           }
 382         ]
 383       },
 384       'area-name-2': {
 385         'level-2': [
 386           {
 387             "interface": "rX-ethY",
 388             "metric": "Z",
 389             "next-hop": "rA",
 390             "parent": "rC(B)",
 391             "type": "TE-IS",
 392             "vertex": "rD"
 393           }
 394         ]
 395       }
 396     }
 397     """
 398     l1out = topotest.normalize_text(
 399         router.vtysh_cmd("show isis topology level-1")
 400     ).splitlines()
 401     l2out = topotest.normalize_text(
 402         router.vtysh_cmd("show isis topology level-2")
 403     ).splitlines()
 404
 405     l1 = parse_topology(l1out, "level-1")
 406     l2 = parse_topology(l2out, "level-2")
 407
 408     dict_merge(l1, l2)
 409     return l1