tests/topotests/zebra_rib/test_zebra_rib.py

   1 #!/usr/bin/env python
   2 # SPDX-License-Identifier: ISC
   3 #
   4 # test_zebra_rib.py
   5 #
   6 # Copyright (c) 2019 by
   7 # Cumulus Networks, Inc
   8 # Donald Sharp
   9 #
  10
  11 """
  12 test_zebra_rib.py: Test some basic zebra <-> kernel interactions
  13 """
  14
  15 import os
  16 import re
  17 import sys
  18 from functools import partial
  19 import pytest
  20 import json
  21 import platform
  22
  23 # Save the Current Working Directory to find configuration files.
  24 CWD = os.path.dirname(os.path.realpath(__file__))
  25 sys.path.append(os.path.join(CWD, "../"))
  26
  27 # pylint: disable=C0413
  28 # Import topogen and topotest helpers
  29 from lib import topotest
  30 from lib.topogen import Topogen, TopoRouter, get_topogen
  31 from lib.topolog import logger
  32 from time import sleep
  33
  34
  35 pytestmark = [pytest.mark.sharpd]
  36 krel = platform.release()
  37
  38
  39 def config_macvlan(tgen, r_str, device, macvlan):
  40     "Creates specified macvlan interace on physical device"
  41
  42     if topotest.version_cmp(krel, "5.1") < 0:
  43         return
  44
  45     router = tgen.gears[r_str]
  46     router.run(
  47         "ip link add {} link {} type macvlan mode bridge".format(macvlan, device)
  48     )
  49     router.run("ip link set {} up".format(macvlan))
  50
  51
  52 def setup_module(mod):
  53     "Sets up the pytest environment"
  54     topodef = {"s1": ("r1", "r1", "r1", "r1", "r1", "r1", "r1", "r1")}
  55     tgen = Topogen(topodef, mod.__name__)
  56     tgen.start_topology()
  57
  58     router_list = tgen.routers()
  59     for rname, router in router_list.items():
  60         router.load_config(
  61             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
  62         )
  63         router.load_config(
  64             TopoRouter.RD_SHARP, os.path.join(CWD, "{}/sharpd.conf".format(rname))
  65         )
  66
  67     # Macvlan interface for protodown func test */
  68     config_macvlan(tgen, "r1", "r1-eth0", "r1-eth0-macvlan")
  69     # Initialize all routers.
  70     tgen.start_router()
  71
  72
  73 def teardown_module(mod):
  74     "Teardown the pytest environment"
  75     tgen = get_topogen()
  76     tgen.stop_topology()
  77
  78
  79 def test_zebra_kernel_admin_distance():
  80     "Test some basic kernel routes added that should be accepted"
  81     logger.info("Test some basic kernel routes that should be accepted")
  82     tgen = get_topogen()
  83     if tgen.routers_have_failure():
  84         pytest.skip("skipped because of router(s) failure")
  85
  86     r1 = tgen.gears["r1"]
  87
  88     # Route with 255/8192 metric
  89
  90     distance = 255
  91     metric = 8192
  92
  93     def makekmetric(dist, metric):
  94         return (dist << 24) + metric
  95
  96     r1.run(
  97         "ip route add 4.5.1.0/24 via 192.168.210.2 dev r1-eth0 metric "
  98         + str(makekmetric(255, 8192))
  99     )
 100     # Route with 1/1 metric
 101     r1.run(
 102         "ip route add 4.5.2.0/24 via 192.168.211.2 dev r1-eth1 metric "
 103         + str(makekmetric(1, 1))
 104     )
 105     # Route with 10/1 metric
 106     r1.run(
 107         "ip route add 4.5.3.0/24 via 192.168.212.2 dev r1-eth2 metric "
 108         + str(makekmetric(10, 1))
 109     )
 110     # Same route with a 160/1 metric
 111     r1.run(
 112         "ip route add 4.5.3.0/24 via 192.168.213.2 dev r1-eth3 metric "
 113         + str(makekmetric(160, 1))
 114     )
 115
 116     # Currently I believe we have a bug here with the same route and different
 117     # metric.  That needs to be properly resolved.  Making a note for
 118     # coming back around later and fixing this.
 119     # tgen.mininet_cli()
 120     for i in range(1, 2):
 121         json_file = "{}/r1/v4_route_{}.json".format(CWD, i)
 122         expected = json.loads(open(json_file).read())
 123
 124         test_func = partial(
 125             topotest.router_json_cmp,
 126             r1,
 127             "show ip route 4.5.{}.0 json".format(i),
 128             expected,
 129         )
 130         _, result = topotest.run_and_expect(test_func, None, count=20, wait=0.5)
 131         assertmsg = '"r1" JSON output mismatches'
 132         assert result is None, assertmsg
 133     # tgen.mininet_cli()
 134
 135
 136 def test_zebra_kernel_override():
 137     "Test that a FRR route with a lower admin distance takes over"
 138     logger.info("Test kernel override with a better admin distance")
 139     tgen = get_topogen()
 140     if tgen.routers_have_failure():
 141         pytest.skip("skipped because of previous test failure")
 142
 143     r1 = tgen.gears["r1"]
 144     r1.vtysh_cmd("conf\nip route 4.5.1.0/24 192.168.216.3")
 145     json_file = "{}/r1/v4_route_1_static_override.json".format(CWD)
 146     expected = json.loads(open(json_file).read())
 147
 148     test_func = partial(
 149         topotest.router_json_cmp, r1, "show ip route 4.5.1.0 json", expected
 150     )
 151     _, result = topotest.run_and_expect(test_func, None, count=20, wait=0.5)
 152     assert result is None, '"r1" JSON output mismatches'
 153
 154     logger.info(
 155         "Test that the removal of the static route allows the kernel to take back over"
 156     )
 157     r1.vtysh_cmd("conf\nno ip route 4.5.1.0/24 192.168.216.3")
 158     json_file = "{}/r1/v4_route_1.json".format(CWD)
 159     expected = json.loads(open(json_file).read())
 160
 161     test_func = partial(
 162         topotest.router_json_cmp, r1, "show ip route 4.5.1.0 json", expected
 163     )
 164     _, result = topotest.run_and_expect(test_func, None, count=20, wait=0.5)
 165     assert result is None, '"r1" JSON output mismatches'
 166
 167
 168 def test_route_map_usage():
 169     "Test that FRR only reruns over routes associated with the routemap"
 170     logger.info("Test that FRR runs on selected re's on route-map changes")
 171     tgen = get_topogen()
 172     if tgen.routers_have_failure():
 173         pytest.skip("Skipped because of previous test failure")
 174
 175     thisDir = os.path.dirname(os.path.realpath(__file__))
 176
 177     r1 = tgen.gears["r1"]
 178     # set the delay timer to 1 to improve test coverage (HA)
 179     r1.vtysh_cmd("conf\nzebra route-map delay-timer 1")
 180     r1.vtysh_cmd("conf\nroute-map static permit 10\nset src 192.168.215.1")
 181     r1.vtysh_cmd("conf\naccess-list 5 seq 5 permit 10.0.0.44/32")
 182     r1.vtysh_cmd("conf\naccess-list 10 seq 5 permit 10.0.1.0/24")
 183     r1.vtysh_cmd(
 184         "conf\nroute-map sharp permit 10\nmatch ip address 10\nset src 192.168.214.1"
 185     )
 186     r1.vtysh_cmd("conf\nroute-map sharp permit 20\nset src 192.168.213.1")
 187     r1.vtysh_cmd("conf\nip protocol static route-map static")
 188     r1.vtysh_cmd("conf\nip protocol sharp route-map sharp")
 189     sleep(4)
 190     r1.vtysh_cmd("conf\nip route 10.100.100.100/32 192.168.216.3")
 191     r1.vtysh_cmd("conf\nip route 10.100.100.101/32 10.0.0.44")
 192     r1.vtysh_cmd("sharp install route 10.0.0.0 nexthop 192.168.216.3 500")
 193
 194     def check_initial_routes_installed(router):
 195         output = json.loads(router.vtysh_cmd("show ip route summ json"))
 196         expected = {
 197             "routes": [{"type": "static", "rib": 2}, {"type": "sharp", "rib": 500}]
 198         }
 199         return topotest.json_cmp(output, expected)
 200
 201     test_func = partial(check_initial_routes_installed, r1)
 202     success, result = topotest.run_and_expect(test_func, None, count=40, wait=1)
 203
 204     static_rmapfile = "%s/r1/static_rmap.ref" % (thisDir)
 205     expected = open(static_rmapfile).read().rstrip()
 206     expected = ("\n".join(expected.splitlines()) + "\n").rstrip()
 207     logger.info(
 208         "Does the show route-map static command run the correct number of times"
 209     )
 210
 211     def check_static_map_correct_runs():
 212         actual = r1.vtysh_cmd("show route-map static")
 213         actual = ("\n".join(actual.splitlines()) + "\n").rstrip()
 214         return topotest.get_textdiff(
 215             actual,
 216             expected,
 217             title1="Actual Route-map output",
 218             title2="Expected Route-map output",
 219         )
 220
 221     ok, result = topotest.run_and_expect(
 222         check_static_map_correct_runs, "", count=5, wait=1
 223     )
 224     assert ok, result
 225
 226     sharp_rmapfile = "%s/r1/sharp_rmap.ref" % (thisDir)
 227     expected = open(sharp_rmapfile).read().rstrip()
 228     expected = ("\n".join(expected.splitlines()) + "\n").rstrip()
 229     logger.info("Does the show route-map sharp command run the correct number of times")
 230
 231     def check_sharp_map_correct_runs():
 232         actual = r1.vtysh_cmd("show route-map sharp")
 233         actual = ("\n".join(actual.splitlines()) + "\n").rstrip()
 234         return topotest.get_textdiff(
 235             actual,
 236             expected,
 237             title1="Actual Route-map output",
 238             title2="Expected Route-map output",
 239         )
 240
 241     ok, result = topotest.run_and_expect(
 242         check_sharp_map_correct_runs, "", count=5, wait=1
 243     )
 244     assert ok, result
 245
 246     logger.info(
 247         "Add a extension to the static route-map to see the static route go away"
 248         " and test that the routes installed are correct"
 249     )
 250
 251     r1.vtysh_cmd("conf\nroute-map sharp deny 5\nmatch ip address 5")
 252     # we are only checking the kernel here as that this will give us the implied
 253     # testing of both the route-map and staticd withdrawing the route
 254     # let's spot check that the routes were installed correctly
 255     # in the kernel
 256     sharp_ipfile = "%s/r1/iproute.ref" % (thisDir)
 257     expected = open(sharp_ipfile).read().rstrip()
 258     expected = ("\n".join(expected.splitlines()) + "\n").rstrip()
 259
 260     def check_routes_installed():
 261         actual = r1.run("ip route show")
 262         actual = ("\n".join(actual.splitlines()) + "\n").rstrip()
 263         actual = re.sub(r" nhid [0-9][0-9]", "", actual)
 264         actual = re.sub(r" proto sharp", " proto XXXX", actual)
 265         actual = re.sub(r" proto static", " proto XXXX", actual)
 266         actual = re.sub(r" proto 194", " proto XXXX", actual)
 267         actual = re.sub(r" proto 196", " proto XXXX", actual)
 268         actual = re.sub(r" proto kernel", " proto XXXX", actual)
 269         actual = re.sub(r" proto 2", " proto XXXX", actual)
 270         # Some platforms have double spaces?  Why??????
 271         actual = re.sub(r"  proto XXXX  ", " proto XXXX ", actual)
 272         actual = re.sub(r"  metric", " metric", actual)
 273         actual = re.sub(r" link  ", " link ", actual)
 274         return topotest.get_textdiff(
 275             actual,
 276             expected,
 277             title1="Actual ip route show",
 278             title2="Expected ip route show",
 279         )
 280
 281     ok, result = topotest.run_and_expect(check_routes_installed, "", count=5, wait=1)
 282     assert ok, result
 283
 284
 285 def test_protodown():
 286     "Run protodown basic functionality test and report results."
 287     pdown = False
 288     count = 0
 289     tgen = get_topogen()
 290     if topotest.version_cmp(krel, "5.1") < 0:
 291         tgen.errors = "kernel 5.1 needed for protodown tests"
 292         pytest.skip(tgen.errors)
 293
 294     r1 = tgen.gears["r1"]
 295
 296     # Set interface protodown on
 297     r1.vtysh_cmd("sharp interface r1-eth0-macvlan protodown")
 298
 299     # Timeout to wait for dplane to handle it
 300     while count < 10:
 301         count += 1
 302         output = r1.vtysh_cmd("show interface r1-eth0-macvlan")
 303         if re.search(r"protodown reasons:.*sharp", output):
 304             pdown = True
 305             break
 306         sleep(1)
 307
 308     assert pdown is True, "Interface r1-eth0-macvlan not set protodown"
 309
 310     # Set interface protodown off
 311     r1.vtysh_cmd("no sharp interface r1-eth0-macvlan protodown")
 312
 313     # Timeout to wait for dplane to handle it
 314     while count < 10:
 315         count += 1
 316         output = r1.vtysh_cmd("show interface r1-eth0-macvlan")
 317         if not re.search(r"protodown reasons:.*sharp", output):
 318             pdown = False
 319             break
 320         sleep(1)
 321
 322     assert pdown is False, "Interface r1-eth0-macvlan not set protodown off"
 323
 324
 325 def test_memory_leak():
 326     "Run the memory leak test and report results."
 327     tgen = get_topogen()
 328     if not tgen.is_memleak_enabled():
 329         pytest.skip("Memory leak test/report is disabled")
 330
 331     tgen.report_memory_leaks()
 332
 333
 334 if __name__ == "__main__":
 335     args = ["-s"] + sys.argv[1:]
 336     sys.exit(pytest.main(args))