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