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
  28 import os
  29 import re
  30 import sys
  31 import pytest
  32 import json
  33 import platform
  34 from functools import partial
  35
  36 # Save the Current Working Directory to find configuration files.
  37 CWD = os.path.dirname(os.path.realpath(__file__))
  38 sys.path.append(os.path.join(CWD, "../"))
  39
  40 # pylint: disable=C0413
  41 # Import topogen and topotest helpers
  42 from lib import topotest
  43 from lib.topogen import Topogen, TopoRouter, get_topogen
  44 from lib.topolog import logger
  45 from lib.common_config import shutdown_bringup_interface
  46
  47 # Required to instantiate the topology builder class.
  48 from mininet.topo import Topo
  49
  50 pytestmark = [pytest.mark.sharpd]
  51
  52
  53 #####################################################
  54 ##
  55 ##   Network Topology Definition
  56 ##
  57 #####################################################
  58
  59
  60 class ZebraTopo(Topo):
  61     "Test topology builder"
  62
  63     def build(self, *_args, **_opts):
  64         "Build function"
  65         tgen = get_topogen(self)
  66
  67         tgen.add_router("r1")
  68
  69         # Create a empty network for router 1
  70         switch = tgen.add_switch("s1")
  71         switch.add_link(tgen.gears["r1"])
  72
  73
  74 #####################################################
  75 ##
  76 ##   Tests starting
  77 ##
  78 #####################################################
  79
  80
  81 def setup_module(mod):
  82     "Sets up the pytest environment"
  83     tgen = Topogen(ZebraTopo, mod.__name__)
  84     tgen.start_topology()
  85
  86     router_list = tgen.routers()
  87     for rname, router in router_list.items():
  88         router.load_config(
  89             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
  90         )
  91
  92         router.load_config(
  93             TopoRouter.RD_SHARP, os.path.join(CWD, "{}/sharpd.conf".format(rname))
  94         )
  95
  96     # Initialize all routers.
  97     tgen.start_router()
  98
  99
 100 def teardown_module(_mod):
 101     "Teardown the pytest environment"
 102     tgen = get_topogen()
 103
 104     # This function tears down the whole topology.
 105     tgen.stop_topology()
 106
 107
 108 def test_zebra_netlink_batching():
 109     "Test the situation where dataplane fills netlink send buffer entirely."
 110     logger.info(
 111         "Test the situation where dataplane fills netlink send buffer entirely."
 112     )
 113     tgen = get_topogen()
 114     if tgen.routers_have_failure():
 115         pytest.skip("skipped because of previous test failure")
 116     r1 = tgen.gears["r1"]
 117
 118     # Reduce the size of the buffer to hit the limit.
 119     r1.vtysh_cmd("conf t\nzebra kernel netlink batch-tx-buf 256 256")
 120
 121     r1.vtysh_cmd("sharp install routes 2.1.3.7 nexthop 192.168.1.1 100")
 122     json_file = "{}/r1/v4_route.json".format(CWD)
 123     expected = json.loads(open(json_file).read())
 124     test_func = partial(
 125         topotest.router_json_cmp,
 126         r1,
 127         "show ip route json",
 128         expected,
 129     )
 130     _, result = topotest.run_and_expect(test_func, None, count=2, wait=0.5)
 131     assertmsg = '"r1" JSON output mismatches'
 132     assert result is None, assertmsg
 133
 134     r1.vtysh_cmd("sharp remove routes 2.1.3.7 100")
 135
 136
 137 if __name__ == "__main__":
 138     args = ["-s"] + sys.argv[1:]
 139     sys.exit(pytest.main(args))