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
  34 CWD = os.path.dirname(os.path.realpath(__file__))
  35 sys.path.append(os.path.join(CWD, "../"))
  36
  37 # pylint: disable=C0413
  38 from lib import topotest
  39 from lib.topogen import Topogen, TopoRouter, get_topogen
  40 from lib.topolog import logger
  41 from lib.topotest import iproute2_is_vrf_capable
  42 from lib.common_config import required_linux_kernel_version
  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     logger.info("Testing with VRF Lite support")
 103
 104     cmds = [
 105         "ip link add {0}-cust1 type vrf table 1001",
 106         "ip link add loop1 type dummy",
 107         "ip link set {0}-eth0 master {0}-cust1",
 108         "ip link set {0}-eth1 master {0}-cust1",
 109     ]
 110
 111     # For all registered routers, load the zebra configuration file
 112     for rname, router in tgen.routers().items():
 113         # create VRF rx-cust1 and link rx-eth0 to rx-cust1
 114         for cmd in cmds:
 115             output = tgen.net[rname].cmd(cmd.format(rname))
 116
 117     for rname, router in tgen.routers().items():
 118         router.load_config(
 119             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
 120         )
 121         router.load_config(
 122             TopoRouter.RD_ISIS, os.path.join(CWD, "{}/isisd.conf".format(rname))
 123         )
 124     # After loading the configurations, this function loads configured daemons.
 125     tgen.start_router()
 126
 127
 128 def teardown_module(mod):
 129     "Teardown the pytest environment"
 130     tgen = get_topogen()
 131     # move back rx-eth0 to default VRF
 132     # delete rx-vrf
 133     tgen.stop_topology()
 134
 135
 136 def test_isis_convergence():
 137     "Wait for the protocol to converge before starting to test"
 138     tgen = get_topogen()
 139     # Don't run this test if we have any failure.
 140     if tgen.routers_have_failure():
 141         pytest.skip(tgen.errors)
 142
 143     logger.info("waiting for ISIS protocol to converge")
 144
 145     for rname, router in tgen.routers().items():
 146         filename = "{0}/{1}/{1}_topology.json".format(CWD, rname)
 147         expected = json.loads(open(filename).read())
 148
 149         def compare_isis_topology(router, expected):
 150             "Helper function to test ISIS vrf topology convergence."
 151             actual = show_isis_topology(router)
 152
 153             return topotest.json_cmp(actual, expected)
 154
 155         test_func = functools.partial(compare_isis_topology, router, expected)
 156         (result, diff) = topotest.run_and_expect(test_func, None, wait=0.5, count=120)
 157         assert result, "ISIS did not converge on {}:\n{}".format(rname, diff)
 158
 159
 160 def test_isis_route_installation():
 161     "Check whether all expected routes are present"
 162     tgen = get_topogen()
 163     # Don't run this test if we have any failure.
 164     if tgen.routers_have_failure():
 165         pytest.skip(tgen.errors)
 166
 167     logger.info("Checking routers for installed ISIS vrf routes")
 168     # Check for routes in 'show ip route vrf {}-cust1 json'
 169     for rname, router in tgen.routers().items():
 170         filename = "{0}/{1}/{1}_route.json".format(CWD, rname)
 171         expected = json.loads(open(filename, "r").read())
 172         actual = router.vtysh_cmd(
 173             "show ip route vrf {0}-cust1 json".format(rname), isjson=True
 174         )
 175         assertmsg = "Router '{}' routes mismatch".format(rname)
 176         assert topotest.json_cmp(actual, expected) is None, assertmsg
 177
 178
 179 def test_isis_linux_route_installation():
 180     "Check whether all expected routes are present and installed in the OS"
 181     tgen = get_topogen()
 182     # Don't run this test if we have any failure.
 183     if tgen.routers_have_failure():
 184         pytest.skip(tgen.errors)
 185
 186     # Required linux kernel version for this suite to run.
 187     result = required_linux_kernel_version("4.15")
 188     if result is not True:
 189         pytest.skip("Kernel requirements are not met")
 190
 191     # iproute2 needs to support VRFs for this suite to run.
 192     if not iproute2_is_vrf_capable():
 193         pytest.skip("Installed iproute2 version does not support VRFs")
 194
 195     logger.info("Checking routers for installed ISIS vrf routes in OS")
 196     # Check for routes in `ip route show vrf {}-cust1`
 197     for rname, router in tgen.routers().items():
 198         filename = "{0}/{1}/{1}_route_linux.json".format(CWD, rname)
 199         expected = json.loads(open(filename, "r").read())
 200         actual = topotest.ip4_vrf_route(router)
 201         assertmsg = "Router '{}' OS routes mismatch".format(rname)
 202         assert topotest.json_cmp(actual, expected) is None, assertmsg
 203
 204
 205 def test_isis_route6_installation():
 206     "Check whether all expected routes are present"
 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 vrf IPv6 routes")
 213     # Check for routes in 'show ipv6 route vrf {}-cust1 json'
 214     for rname, router in tgen.routers().items():
 215         filename = "{0}/{1}/{1}_route6.json".format(CWD, rname)
 216         expected = json.loads(open(filename, "r").read())
 217         actual = router.vtysh_cmd(
 218             "show ipv6 route vrf {}-cust1 json".format(rname), isjson=True
 219         )
 220
 221         assertmsg = "Router '{}' routes mismatch".format(rname)
 222         assert topotest.json_cmp(actual, expected) is None, assertmsg
 223
 224
 225 def test_isis_linux_route6_installation():
 226     "Check whether all expected routes are present and installed in the OS"
 227     tgen = get_topogen()
 228     # Don't run this test if we have any failure.
 229     if tgen.routers_have_failure():
 230         pytest.skip(tgen.errors)
 231
 232     # Required linux kernel version for this suite to run.
 233     result = required_linux_kernel_version("4.15")
 234     if result is not True:
 235         pytest.skip("Kernel requirements are not met")
 236
 237     # iproute2 needs to support VRFs for this suite to run.
 238     if not iproute2_is_vrf_capable():
 239         pytest.skip("Installed iproute2 version does not support VRFs")
 240
 241     logger.info("Checking routers for installed ISIS vrf IPv6 routes in OS")
 242     # Check for routes in `ip -6 route show vrf {}-cust1`
 243     for rname, router in tgen.routers().items():
 244         filename = "{0}/{1}/{1}_route6_linux.json".format(CWD, rname)
 245         expected = json.loads(open(filename, "r").read())
 246         actual = topotest.ip6_vrf_route(router)
 247         assertmsg = "Router '{}' OS routes mismatch".format(rname)
 248         assert topotest.json_cmp(actual, expected) is None, assertmsg
 249
 250
 251 def test_memory_leak():
 252     "Run the memory leak test and report results."
 253     tgen = get_topogen()
 254     if not tgen.is_memleak_enabled():
 255         pytest.skip("Memory leak test/report is disabled")
 256
 257     tgen.report_memory_leaks()
 258
 259
 260 if __name__ == "__main__":
 261     args = ["-s"] + sys.argv[1:]
 262     sys.exit(pytest.main(args))
 263
 264
 265 #
 266 # Auxiliary functions
 267 #
 268
 269
 270 def dict_merge(dct, merge_dct):
 271     """
 272     Recursive dict merge. Inspired by :meth:``dict.update()``, instead of
 273     updating only top-level keys, dict_merge recurses down into dicts nested
 274     to an arbitrary depth, updating keys. The ``merge_dct`` is merged into
 275     ``dct``.
 276     :param dct: dict onto which the merge is executed
 277     :param merge_dct: dct merged into dct
 278     :return: None
 279
 280     Source:
 281     https://gist.github.com/angstwad/bf22d1822c38a92ec0a9
 282     """
 283     for k, v in merge_dct.items():
 284         if k in dct and isinstance(dct[k], dict) and topotest.is_mapping(merge_dct[k]):
 285             dict_merge(dct[k], merge_dct[k])
 286         else:
 287             dct[k] = merge_dct[k]
 288
 289
 290 def parse_topology(lines, level):
 291     """
 292     Parse the output of 'show isis topology level-X' into a Python dict.
 293     """
 294     areas = {}
 295     area = None
 296     ipv = None
 297     vertex_type_regex = "|".join(VERTEX_TYPE_LIST)
 298
 299     for line in lines:
 300         area_match = re.match(r"Area (.+):", line)
 301         if area_match:
 302             area = area_match.group(1)
 303             if area not in areas:
 304                 areas[area] = {level: {"ipv4": [], "ipv6": []}}
 305             ipv = None
 306             continue
 307         elif area is None:
 308             continue
 309
 310         if re.match(r"IS\-IS paths to level-. routers that speak IPv6", line):
 311             ipv = "ipv6"
 312             continue
 313         if re.match(r"IS\-IS paths to level-. routers that speak IP", line):
 314             ipv = "ipv4"
 315             continue
 316
 317         item_match = re.match(
 318             r"([^\s]+) ([^\s]+) ([^\s]+) ([^\s]+) ([^\s]+) ([^\s]+)", line
 319         )
 320         if (
 321             item_match is not None
 322             and item_match.group(1) == "Vertex"
 323             and item_match.group(2) == "Type"
 324             and item_match.group(3) == "Metric"
 325             and item_match.group(4) == "Next-Hop"
 326             and item_match.group(5) == "Interface"
 327             and item_match.group(6) == "Parent"
 328         ):
 329             # Skip header
 330             continue
 331
 332         item_match = re.match(
 333             r"([^\s]+) ({}) ([0]|([1-9][0-9]*)) ([^\s]+) ([^\s]+) ([^\s]+)".format(
 334                 vertex_type_regex
 335             ),
 336             line,
 337         )
 338         if item_match is not None:
 339             areas[area][level][ipv].append(
 340                 {
 341                     "vertex": item_match.group(1),
 342                     "type": item_match.group(2),
 343                     "metric": item_match.group(3),
 344                     "next-hop": item_match.group(5),
 345                     "interface": item_match.group(6),
 346                     "parent": item_match.group(7),
 347                 }
 348             )
 349             continue
 350
 351         item_match = re.match(
 352             r"([^\s]+) ({}) ([0]|([1-9][0-9]*)) ([^\s]+)".format(vertex_type_regex),
 353             line,
 354         )
 355
 356         if item_match is not None:
 357             areas[area][level][ipv].append(
 358                 {
 359                     "vertex": item_match.group(1),
 360                     "type": item_match.group(2),
 361                     "metric": item_match.group(3),
 362                     "parent": item_match.group(5),
 363                 }
 364             )
 365             continue
 366
 367         item_match = re.match(r"([^\s]+)", line)
 368         if item_match is not None:
 369             areas[area][level][ipv].append({"vertex": item_match.group(1)})
 370             continue
 371
 372     return areas
 373
 374
 375 def show_isis_topology(router):
 376     """
 377     Get the ISIS vrf topology in a dictionary format.
 378
 379     Sample:
 380     {
 381       'area-name': {
 382         'level-1': [
 383           {
 384             'vertex': 'r1'
 385           }
 386         ],
 387         'level-2': [
 388           {
 389             'vertex': '10.0.0.1/24',
 390             'type': 'IP',
 391             'parent': '0',
 392             'metric': 'internal'
 393           }
 394         ]
 395       },
 396       'area-name-2': {
 397         'level-2': [
 398           {
 399             "interface": "rX-ethY",
 400             "metric": "Z",
 401             "next-hop": "rA",
 402             "parent": "rC(B)",
 403             "type": "TE-IS",
 404             "vertex": "rD"
 405           }
 406         ]
 407       }
 408     }
 409     """
 410     l1out = topotest.normalize_text(
 411         router.vtysh_cmd("show isis vrf {}-cust1 topology level-1".format(router.name))
 412     ).splitlines()
 413     l2out = topotest.normalize_text(
 414         router.vtysh_cmd("show isis vrf {}-cust1 topology level-2".format(router.name))
 415     ).splitlines()
 416
 417     l1 = parse_topology(l1out, "level-1")
 418     l2 = parse_topology(l2out, "level-2")
 419
 420     dict_merge(l1, l2)
 421     return l1