tests/topotests/zebra_netlink/test_zebra_netlink.py

   1 #!/usr/bin/env python
   2
   3 #
   4 # test_zebra_netlink.py
   5 #
   6 # Copyright (c) 2020 by
   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_zebra_netlink.py: Test some basic interactions with kernel using Netlink
  25
  26 """
  27 # pylint: disable=C0413
  28 import ipaddress
  29 import json
  30 import sys
  31 from functools import partial
  32
  33 import pytest
  34 from lib import topotest
  35 from lib.topogen import Topogen, TopoRouter
  36 from lib.topolog import logger
  37
  38
  39 pytestmark = [pytest.mark.sharpd]
  40
  41
  42 #####################################################
  43 ##
  44 ##   Tests starting
  45 ##
  46 #####################################################
  47
  48
  49 @pytest.fixture(scope="module")
  50 def tgen(request):
  51     "Sets up the pytest environment"
  52
  53     topodef = {"s1": ("r1")}
  54     tgen = Topogen(topodef, request.module.__name__)
  55     tgen.start_topology()
  56
  57     # Initialize all routers.
  58     router_list = tgen.routers()
  59     for rname, router in router_list.items():
  60         router.load_config(TopoRouter.RD_ZEBRA, "zebra.conf")
  61         router.load_config(TopoRouter.RD_SHARP)
  62
  63     tgen.start_router()
  64     yield tgen
  65     tgen.stop_topology()
  66
  67
  68 @pytest.fixture(autouse=True)
  69 def skip_on_failure(tgen):
  70     if tgen.routers_have_failure():
  71         pytest.skip("skipped because of previous test failure")
  72
  73
  74 def test_zebra_netlink_batching(tgen):
  75     "Test the situation where dataplane fills netlink send buffer entirely."
  76     logger.info(
  77         "Test the situation where dataplane fills netlink send buffer entirely."
  78     )
  79     r1 = tgen.gears["r1"]
  80
  81     # Reduce the size of the buffer to hit the limit.
  82     r1.vtysh_cmd("conf t\nzebra kernel netlink batch-tx-buf 256 256")
  83
  84     count = 100
  85     r1.vtysh_cmd("sharp install routes 2.1.3.7 nexthop 192.168.1.1 " + str(count))
  86
  87     # Generate expected results
  88     entry = {
  89         "protocol": "sharp",
  90         "distance": 150,
  91         "metric": 0,
  92         "installed": True,
  93         "table": 254,
  94         "nexthops": [
  95             {
  96                 "fib": True,
  97                 "ip": "192.168.1.1",
  98                 "afi": "ipv4",
  99                 "interfaceName": "r1-eth0",
 100                 "active": True,
 101                 "weight": 1,
 102             }
 103         ],
 104     }
 105
 106     match = {}
 107     base = int(ipaddress.ip_address(u"2.1.3.7"))
 108     for i in range(base, base + count):
 109         pfx = str(ipaddress.ip_network((i, 32)))
 110         match[pfx] = [dict(entry, prefix=pfx)]
 111
 112     ok = topotest.router_json_cmp_retry(r1, "show ip route json", match, False, 30)
 113     assert ok, '"r1" JSON output mismatches'
 114
 115     r1.vtysh_cmd("sharp remove routes 2.1.3.7 " + str(count))
 116
 117
 118 if __name__ == "__main__":
 119     args = ["-s"] + sys.argv[1:]
 120     sys.exit(pytest.main(args))