1#!/usr/bin/env python3 2# SPDX-License-Identifier: GPL-2.0 3 4"""Regression tests for the SO_TXTIME interface. 5 6Test delivery time in FQ and ETF qdiscs. 7""" 8 9import os 10import time 11 12from lib.py import ksft_exit, ksft_run, ksft_variants 13from lib.py import KsftNamedVariant, KsftSkipEx 14from lib.py import NetDrvEpEnv, bkg, cmd, defer, tc 15 16 17def test_so_txtime(cfg, clockid, ipver, args_tx, args_rx, expect_success): 18 """Main function. Run so_txtime as sender and receiver.""" 19 slow_machine = os.environ.get('KSFT_MACHINE_SLOW') 20 21 if not hasattr(cfg, "bin_remote"): 22 cfg.bin_local = cfg.test_dir / "so_txtime" 23 cfg.bin_remote = cfg.remote.deploy(cfg.bin_local) 24 25 tstart = time.time_ns() + (2000_000_000 if slow_machine else 200_000_000) 26 27 cmd_addr = f"-S {cfg.addr_v[ipver]} -D {cfg.remote_addr_v[ipver]}" 28 cmd_args = f"-{ipver} -c {clockid} -t {tstart} {cmd_addr}" 29 cmd_rx = f"{cfg.bin_remote} {cmd_args} {args_rx} -r" 30 cmd_tx = f"{cfg.bin_local} {cmd_args} {args_tx}" 31 32 expect_fail = not expect_success 33 if slow_machine: 34 expect_success = False 35 36 with bkg(cmd_rx, host=cfg.remote, fail=expect_success, 37 expect_fail=expect_fail, exit_wait=True): 38 cmd(cmd_tx) 39 40 41def _qdisc_setup(ifname, qdisc, optargs=""): 42 """Replace root qdisc. Restore the original after the test. 43 44 If the original is mq, children will be of type default_qdisc. 45 """ 46 orig = tc(f"qdisc show dev {ifname} root", json=True)[0].get("kind", None) 47 defer(tc, f"qdisc replace dev {ifname} root {orig}") 48 tc(f"qdisc replace dev {ifname} root {qdisc} {optargs}") 49 50 51def _test_variants_fq(): 52 for ipver in ["4", "6"]: 53 for testcase in [ 54 ["no_delay", "a,-1", "a,-1"], 55 ["zero_delay", "a,0", "a,0"], 56 ["one_pkt", "a,10", "a,10"], 57 ["in_order", "a,10,b,20", "a,10,b,20"], 58 ["reverse_order", "a,20,b,10", "b,10,a,20"], 59 ]: 60 name = f"v{ipver}_{testcase[0]}" 61 yield KsftNamedVariant(name, ipver, testcase[1], testcase[2]) 62 63 64@ksft_variants(_test_variants_fq()) 65def test_so_txtime_fq_mono(cfg, ipver, args_tx, args_rx): 66 """Run all variants of monotonic (fq) tests.""" 67 cfg.require_ipver(ipver) 68 _qdisc_setup(cfg.ifname, "fq") 69 test_so_txtime(cfg, "mono", ipver, args_tx, args_rx, True) 70 71 72@ksft_variants(_test_variants_fq()) 73def test_so_txtime_fq_tai(cfg, ipver, args_tx, args_rx): 74 """Run all variants of fq tests, but pass CLOCK_TAI to test conversion.""" 75 cfg.require_ipver(ipver) 76 _qdisc_setup(cfg.ifname, "fq") 77 test_so_txtime(cfg, "tai", ipver, args_tx, args_rx, True) 78 79 80def _test_variants_etf(): 81 for ipver in ["4", "6"]: 82 for testcase in [ 83 ["no_delay", "a,-1", "a,-1", False], 84 ["zero_delay", "a,0", "a,0", False], 85 ["one_pkt", "a,10", "a,10", True], 86 ["in_order", "a,10,b,20", "a,10,b,20", True], 87 ["reverse_order", "a,20,b,10", "b,10,a,20", True], 88 ]: 89 name = f"v{ipver}_{testcase[0]}" 90 yield KsftNamedVariant( 91 name, ipver, testcase[1], testcase[2], testcase[3] 92 ) 93 94 95@ksft_variants(_test_variants_etf()) 96def test_so_txtime_etf(cfg, ipver, args_tx, args_rx, expect_fail): 97 """Run all variants of etf tests.""" 98 cfg.require_ipver(ipver) 99 try: 100 _qdisc_setup(cfg.ifname, "etf", "clockid CLOCK_TAI delta 400000") 101 except Exception as e: 102 raise KsftSkipEx("tc does not support qdisc etf. skipping") from e 103 104 test_so_txtime(cfg, "tai", ipver, args_tx, args_rx, expect_fail) 105 106 107def main() -> None: 108 """Boilerplate ksft main.""" 109 with NetDrvEpEnv(__file__) as cfg: 110 ksft_run( 111 [test_so_txtime_fq_mono, test_so_txtime_fq_tai, test_so_txtime_etf], 112 args=(cfg,), 113 ) 114 ksft_exit() 115 116 117if __name__ == "__main__": 118 main() 119