tests/topotests/tc_basic/test_tc_basic.py

   1 #!/usr/bin/python
   2
   3 #
   4 # test_tc_basic.py
   5 #
   6 # Copyright (c) 2022 by Shichu Yang
   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 test_tc_basic.py: Test basic TC filters, classes and qdiscs.
  24 """
  25 import sys
  26 import os
  27 import pytest
  28 import time
  29
  30 # Save the Current Working Directory to find configuration files.
  31 CWD = os.path.dirname(os.path.realpath(__file__))
  32 sys.path.append(os.path.join(CWD, "../"))
  33 sys.path.append(os.path.join(CWD, "../lib/"))
  34
  35 from lib.topogen import Topogen, TopoRouter
  36 from lib.topolog import logger
  37
  38 pytestmark = [
  39     pytest.mark.sharpd
  40 ]
  41
  42 def build_topo(tgen):
  43     "Build function"
  44
  45     r1 = tgen.add_router("r1")
  46     r2 = tgen.add_router("r2")
  47
  48     # Create a p2p connection between r1 and r2
  49     tgen.add_link(r1, r2)
  50
  51     # Create switches with each router connected to it to simulate a empty network.
  52     switch = tgen.add_switch("s1")
  53     switch.add_link(r1)
  54
  55     switch = tgen.add_switch("s2")
  56     switch.add_link(r2)
  57
  58 # New form of setup/teardown using pytest fixture
  59 @pytest.fixture(scope="module")
  60 def tgen(request):
  61     "Setup/Teardown the environment and provide tgen argument to tests"
  62
  63     # This function initiates the topology build with Topogen...
  64     tgen = Topogen(build_topo, request.module.__name__)
  65
  66     # ... and here it calls initialization functions.
  67     tgen.start_topology()
  68
  69     # This is a sample of configuration loading.
  70     router_list = tgen.routers()
  71
  72     # For all routers arrange for:
  73     # - starting zebra using config file from <rtrname>/zebra.conf
  74     # - starting ospfd using an empty config file.
  75     for _, router in router_list.items():
  76         router.load_config(TopoRouter.RD_ZEBRA)
  77         router.load_config(TopoRouter.RD_SHARP)
  78
  79     # Start and configure the router daemons
  80     tgen.start_router()
  81
  82     # Provide tgen as argument to each test function
  83     yield tgen
  84
  85     # Teardown after last test runs
  86     tgen.stop_topology()
  87
  88
  89 # Fixture that executes before each test
  90 @pytest.fixture(autouse=True)
  91 def skip_on_failure(tgen):
  92     if tgen.routers_have_failure():
  93         pytest.skip("skipped because of previous test failure")
  94
  95 def fetch_iproute2_tc_info(r, interface):
  96     qdisc = r.cmd("tc qdisc show dev %s" % interface)
  97     tclass = r.cmd("tc class show dev %s" % interface)
  98     tfilter = r.cmd("tc filter show dev %s" % interface)
  99     return qdisc, tclass, tfilter
 100
 101 # ===================
 102 # The tests functions
 103 # ===================
 104
 105 def test_tc_basic(tgen):
 106     "Test installing one pair of filter & class by sharpd"
 107
 108     r1 = tgen.gears["r1"]
 109     intf = "r1-eth0"
 110     r1.vtysh_cmd("sharp tc dev %s source 192.168.100.0/24 destination 192.168.101.0/24 ip-protocol tcp src-port 8000 dst-port 8001 rate 20mbit" % intf)
 111
 112     time.sleep(3)
 113
 114     qdisc, tclass, tfilter = fetch_iproute2_tc_info(r1, intf)
 115
 116     logger.info("tc qdisc on %s: %s", intf, qdisc)
 117     logger.info("tc class on %s: %s", intf, tclass)
 118     logger.info("tc filter on %s: %s", intf, tfilter)
 119
 120     assert "htb" in qdisc
 121     assert "beef:" in qdisc
 122
 123     assert "20Mbit" in tclass
 124
 125     assert "tcp" in tfilter
 126     assert "dst_ip 192.168.101.0/24" in tfilter
 127     assert "src_ip 192.168.100.0/24" in tfilter
 128     assert "dst_port 8001" in tfilter
 129     assert "src_port 8000" in tfilter
 130
 131 if __name__ == "__main__":
 132     args = ["-s"] + sys.argv[1:]
 133     sys.exit(pytest.main(args))