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         actual = router.vtysh_cmd("show ip route json", isjson=True)
 165         assertmsg = "Router '{}' routes mismatch".format(rname)
 166         assert topotest.json_cmp(actual, expected) is None, assertmsg
 167
 168
 169 def test_isis_linux_route_installation():
 170     "Check whether all expected routes are present and installed in the OS"
 171     tgen = get_topogen()
 172     # Don't run this test if we have any failure.
 173     if tgen.routers_have_failure():
 174         pytest.skip(tgen.errors)
 175
 176     logger.info("Checking routers for installed ISIS routes in OS")
 177
 178     # Check for routes in `ip route`
 179     for rname, router in tgen.routers().items():
 180         filename = "{0}/{1}/{1}_route_linux.json".format(CWD, rname)
 181         expected = json.loads(open(filename, "r").read())
 182         actual = topotest.ip4_route(router)
 183         assertmsg = "Router '{}' OS routes mismatch".format(rname)
 184         assert topotest.json_cmp(actual, expected) is None, assertmsg
 185
 186
 187 def test_isis_route6_installation():
 188     "Check whether all expected routes are present"
 189     tgen = get_topogen()
 190     # Don't run this test if we have any failure.
 191     if tgen.routers_have_failure():
 192         pytest.skip(tgen.errors)
 193
 194     logger.info("Checking routers for installed ISIS IPv6 routes")
 195
 196     # Check for routes in 'show ip route json'
 197     for rname, router in tgen.routers().items():
 198         filename = "{0}/{1}/{1}_route6.json".format(CWD, rname)
 199         expected = json.loads(open(filename, "r").read())
 200         actual = router.vtysh_cmd("show ipv6 route json", isjson=True)
 201         assertmsg = "Router '{}' routes mismatch".format(rname)
 202         assert topotest.json_cmp(actual, expected) is None, assertmsg
 203
 204
 205 def test_isis_linux_route6_installation():
 206     "Check whether all expected routes are present and installed in the OS"
 207     tgen = get_topogen()
 208     # Don't run this test if we have any failure.
 209     if tgen.routers_have_failure():
 210         pytest.skip(tgen.errors)
 211
 212     logger.info("Checking routers for installed ISIS IPv6 routes in OS")
 213
 214     # Check for routes in `ip route`
 215     for rname, router in tgen.routers().items():
 216         filename = "{0}/{1}/{1}_route6_linux.json".format(CWD, rname)
 217         expected = json.loads(open(filename, "r").read())
 218         actual = topotest.ip6_route(router)
 219         assertmsg = "Router '{}' OS routes mismatch".format(rname)
 220         assert topotest.json_cmp(actual, expected) is None, assertmsg
 221
 222
 223 def test_memory_leak():
 224     "Run the memory leak test and report results."
 225     tgen = get_topogen()
 226     if not tgen.is_memleak_enabled():
 227         pytest.skip("Memory leak test/report is disabled")
 228
 229     tgen.report_memory_leaks()
 230
 231
 232 if __name__ == "__main__":
 233     args = ["-s"] + sys.argv[1:]
 234     sys.exit(pytest.main(args))
 235
 236
 237 #
 238 # Auxiliary functions
 239 #
 240
 241
 242 def dict_merge(dct, merge_dct):
 243     """
 244     Recursive dict merge. Inspired by :meth:``dict.update()``, instead of
 245     updating only top-level keys, dict_merge recurses down into dicts nested
 246     to an arbitrary depth, updating keys. The ``merge_dct`` is merged into
 247     ``dct``.
 248     :param dct: dict onto which the merge is executed
 249     :param merge_dct: dct merged into dct
 250     :return: None
 251
 252     Source:
 253     https://gist.github.com/angstwad/bf22d1822c38a92ec0a9
 254     """
 255     for k, v in merge_dct.items():
 256         if k in dct and isinstance(dct[k], dict) and topotest.is_mapping(merge_dct[k]):
 257             dict_merge(dct[k], merge_dct[k])
 258         else:
 259             dct[k] = merge_dct[k]
 260
 261
 262 def parse_topology(lines, level):
 263     """
 264     Parse the output of 'show isis topology level-X' into a Python dict.
 265     """
 266     areas = {}
 267     area = None
 268     ipv = None
 269     vertex_type_regex = "|".join(VERTEX_TYPE_LIST)
 270
 271     for line in lines:
 272         area_match = re.match(r"Area (.+):", line)
 273         if area_match:
 274             area = area_match.group(1)
 275             if area not in areas:
 276                 areas[area] = {level: {"ipv4": [], "ipv6": []}}
 277             ipv = None
 278             continue
 279         elif area is None:
 280             continue
 281
 282         if re.match(r"IS\-IS paths to level-. routers that speak IPv6", line):
 283             ipv = "ipv6"
 284             continue
 285         if re.match(r"IS\-IS paths to level-. routers that speak IP", line):
 286             ipv = "ipv4"
 287             continue
 288
 289         item_match = re.match(
 290             r"([^\s]+) ([^\s]+) ([^\s]+) ([^\s]+) ([^\s]+) ([^\s]+)", line
 291         )
 292         if (
 293             item_match is not None
 294             and item_match.group(1) == "Vertex"
 295             and item_match.group(2) == "Type"
 296             and item_match.group(3) == "Metric"
 297             and item_match.group(4) == "Next-Hop"
 298             and item_match.group(5) == "Interface"
 299             and item_match.group(6) == "Parent"
 300         ):
 301             # Skip header
 302             continue
 303
 304         item_match = re.match(
 305             r"([^\s]+) ({}) ([0]|([1-9][0-9]*)) ([^\s]+) ([^\s]+) ([^\s]+)".format(
 306                 vertex_type_regex
 307             ),
 308             line,
 309         )
 310         if item_match is not None:
 311             areas[area][level][ipv].append(
 312                 {
 313                     "vertex": item_match.group(1),
 314                     "type": item_match.group(2),
 315                     "metric": item_match.group(3),
 316                     "next-hop": item_match.group(5),
 317                     "interface": item_match.group(6),
 318                     "parent": item_match.group(7),
 319                 }
 320             )
 321             continue
 322
 323         item_match = re.match(
 324             r"([^\s]+) ({}) ([0]|([1-9][0-9]*)) ([^\s]+)".format(vertex_type_regex),
 325             line,
 326         )
 327
 328         if item_match is not None:
 329             areas[area][level][ipv].append(
 330                 {
 331                     "vertex": item_match.group(1),
 332                     "type": item_match.group(2),
 333                     "metric": item_match.group(3),
 334                     "parent": item_match.group(5),
 335                 }
 336             )
 337             continue
 338
 339         item_match = re.match(r"([^\s]+)", line)
 340         if item_match is not None:
 341             areas[area][level][ipv].append({"vertex": item_match.group(1)})
 342             continue
 343
 344     return areas
 345
 346
 347 def show_isis_topology(router):
 348     """
 349     Get the ISIS topology in a dictionary format.
 350
 351     Sample:
 352     {
 353       'area-name': {
 354         'level-1': [
 355           {
 356             'vertex': 'r1'
 357           }
 358         ],
 359         'level-2': [
 360           {
 361             'vertex': '10.0.0.1/24',
 362             'type': 'IP',
 363             'parent': '0',
 364             'metric': 'internal'
 365           }
 366         ]
 367       },
 368       'area-name-2': {
 369         'level-2': [
 370           {
 371             "interface": "rX-ethY",
 372             "metric": "Z",
 373             "next-hop": "rA",
 374             "parent": "rC(B)",
 375             "type": "TE-IS",
 376             "vertex": "rD"
 377           }
 378         ]
 379       }
 380     }
 381     """
 382     l1out = topotest.normalize_text(
 383         router.vtysh_cmd("show isis topology level-1")
 384     ).splitlines()
 385     l2out = topotest.normalize_text(
 386         router.vtysh_cmd("show isis topology level-2")
 387     ).splitlines()
 388
 389     l1 = parse_topology(l1out, "level-1")
 390     l2 = parse_topology(l2out, "level-2")
 391
 392     dict_merge(l1, l2)
 393     return l1