tests/topotests/isis_topo1_vrf/test_isis_topo1_vrf.py

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