tests/topotests/tc_basic/test_tc_basic.py

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