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     ]
 109
 110     eth1_cmds = ["ip link set {0}-eth1 master {0}-cust1"]
 111
 112     # For all registered routers, load the zebra configuration file
 113     for rname, router in tgen.routers().items():
 114         # create VRF rx-cust1 and link rx-eth0 to rx-cust1
 115         for cmd in cmds:
 116             output = tgen.net[rname].cmd(cmd.format(rname))
 117
 118         # If router has an rX-eth1, link that to vrf also
 119         if "{}-eth1".format(rname) in router.links.keys():
 120             for cmd in eth1_cmds:
 121                 output = output + tgen.net[rname].cmd(cmd.format(rname))
 122
 123     for rname, router in tgen.routers().items():
 124         router.load_config(
 125             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
 126         )
 127         router.load_config(
 128             TopoRouter.RD_ISIS, os.path.join(CWD, "{}/isisd.conf".format(rname))
 129         )
 130     # After loading the configurations, this function loads configured daemons.
 131     tgen.start_router()
 132
 133
 134 def teardown_module(mod):
 135     "Teardown the pytest environment"
 136     tgen = get_topogen()
 137     # move back rx-eth0 to default VRF
 138     # delete rx-vrf
 139     tgen.stop_topology()
 140
 141
 142 def test_isis_convergence():
 143     "Wait for the protocol to converge before starting to test"
 144     tgen = get_topogen()
 145     # Don't run this test if we have any failure.
 146     if tgen.routers_have_failure():
 147         pytest.skip(tgen.errors)
 148
 149     logger.info("waiting for ISIS protocol to converge")
 150
 151     for rname, router in tgen.routers().items():
 152         filename = "{0}/{1}/{1}_topology.json".format(CWD, rname)
 153         expected = json.loads(open(filename).read())
 154
 155         def compare_isis_topology(router, expected):
 156             "Helper function to test ISIS vrf topology convergence."
 157             actual = show_isis_topology(router)
 158
 159             return topotest.json_cmp(actual, expected)
 160
 161         test_func = functools.partial(compare_isis_topology, router, expected)
 162         (result, diff) = topotest.run_and_expect(test_func, None, wait=0.5, count=120)
 163         assert result, "ISIS did not converge on {}:\n{}".format(rname, diff)
 164
 165
 166 def test_isis_route_installation():
 167     "Check whether all expected routes are present"
 168     tgen = get_topogen()
 169     # Don't run this test if we have any failure.
 170     if tgen.routers_have_failure():
 171         pytest.skip(tgen.errors)
 172
 173     logger.info("Checking routers for installed ISIS vrf routes")
 174     # Check for routes in 'show ip route vrf {}-cust1 json'
 175     for rname, router in tgen.routers().items():
 176         filename = "{0}/{1}/{1}_route.json".format(CWD, rname)
 177         expected = json.loads(open(filename, "r").read())
 178         actual = router.vtysh_cmd(
 179             "show ip route vrf {0}-cust1 json".format(rname), isjson=True
 180         )
 181         assertmsg = "Router '{}' routes mismatch".format(rname)
 182         assert topotest.json_cmp(actual, expected) is None, assertmsg
 183
 184
 185 def test_isis_linux_route_installation():
 186     "Check whether all expected routes are present and installed in the OS"
 187     tgen = get_topogen()
 188     # Don't run this test if we have any failure.
 189     if tgen.routers_have_failure():
 190         pytest.skip(tgen.errors)
 191
 192     # Required linux kernel version for this suite to run.
 193     result = required_linux_kernel_version("4.15")
 194     if result is not True:
 195         pytest.skip("Kernel requirements are not met")
 196
 197     # iproute2 needs to support VRFs for this suite to run.
 198     if not iproute2_is_vrf_capable():
 199         pytest.skip("Installed iproute2 version does not support VRFs")
 200
 201     logger.info("Checking routers for installed ISIS vrf routes in OS")
 202     # Check for routes in `ip route show vrf {}-cust1`
 203     for rname, router in tgen.routers().items():
 204         filename = "{0}/{1}/{1}_route_linux.json".format(CWD, rname)
 205         expected = json.loads(open(filename, "r").read())
 206         actual = topotest.ip4_vrf_route(router)
 207         assertmsg = "Router '{}' OS routes mismatch".format(rname)
 208         assert topotest.json_cmp(actual, expected) is None, assertmsg
 209
 210
 211 def test_isis_route6_installation():
 212     "Check whether all expected routes are present"
 213     tgen = get_topogen()
 214     # Don't run this test if we have any failure.
 215     if tgen.routers_have_failure():
 216         pytest.skip(tgen.errors)
 217
 218     logger.info("Checking routers for installed ISIS vrf IPv6 routes")
 219     # Check for routes in 'show ipv6 route vrf {}-cust1 json'
 220     for rname, router in tgen.routers().items():
 221         filename = "{0}/{1}/{1}_route6.json".format(CWD, rname)
 222         expected = json.loads(open(filename, "r").read())
 223         actual = router.vtysh_cmd(
 224             "show ipv6 route vrf {}-cust1 json".format(rname), isjson=True
 225         )
 226
 227         assertmsg = "Router '{}' routes mismatch".format(rname)
 228         assert topotest.json_cmp(actual, expected) is None, assertmsg
 229
 230
 231 def test_isis_linux_route6_installation():
 232     "Check whether all expected routes are present and installed in the OS"
 233     tgen = get_topogen()
 234     # Don't run this test if we have any failure.
 235     if tgen.routers_have_failure():
 236         pytest.skip(tgen.errors)
 237
 238     # Required linux kernel version for this suite to run.
 239     result = required_linux_kernel_version("4.15")
 240     if result is not True:
 241         pytest.skip("Kernel requirements are not met")
 242
 243     # iproute2 needs to support VRFs for this suite to run.
 244     if not iproute2_is_vrf_capable():
 245         pytest.skip("Installed iproute2 version does not support VRFs")
 246
 247     logger.info("Checking routers for installed ISIS vrf IPv6 routes in OS")
 248     # Check for routes in `ip -6 route show vrf {}-cust1`
 249     for rname, router in tgen.routers().items():
 250         filename = "{0}/{1}/{1}_route6_linux.json".format(CWD, rname)
 251         expected = json.loads(open(filename, "r").read())
 252         actual = topotest.ip6_vrf_route(router)
 253         assertmsg = "Router '{}' OS routes mismatch".format(rname)
 254         assert topotest.json_cmp(actual, expected) is None, assertmsg
 255
 256
 257 def test_memory_leak():
 258     "Run the memory leak test and report results."
 259     tgen = get_topogen()
 260     if not tgen.is_memleak_enabled():
 261         pytest.skip("Memory leak test/report is disabled")
 262
 263     tgen.report_memory_leaks()
 264
 265
 266 if __name__ == "__main__":
 267     args = ["-s"] + sys.argv[1:]
 268     sys.exit(pytest.main(args))
 269
 270
 271 #
 272 # Auxiliary functions
 273 #
 274
 275
 276 def dict_merge(dct, merge_dct):
 277     """
 278     Recursive dict merge. Inspired by :meth:``dict.update()``, instead of
 279     updating only top-level keys, dict_merge recurses down into dicts nested
 280     to an arbitrary depth, updating keys. The ``merge_dct`` is merged into
 281     ``dct``.
 282     :param dct: dict onto which the merge is executed
 283     :param merge_dct: dct merged into dct
 284     :return: None
 285
 286     Source:
 287     https://gist.github.com/angstwad/bf22d1822c38a92ec0a9
 288     """
 289     for k, v in merge_dct.items():
 290         if k in dct and isinstance(dct[k], dict) and topotest.is_mapping(merge_dct[k]):
 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