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 bin_path = cfg.test_dir / "so_txtime" 22 23 tstart = time.time_ns() + (2000_000_000 if slow_machine else 200_000_000) 24 25 cmd_addr = f"-S {cfg.addr_v[ipver]} -D {cfg.remote_addr_v[ipver]}" 26 cmd_base = f"{bin_path} -{ipver} -c {clockid} -t {tstart} {cmd_addr}" 27 cmd_rx = f"{cmd_base} {args_rx} -r" 28 cmd_tx = f"{cmd_base} {args_tx}" 29 30 expect_fail = not expect_success 31 if slow_machine: 32 expect_success = False 33 34 with bkg(cmd_rx, host=cfg.remote, fail=expect_success, 35 expect_fail=expect_fail, exit_wait=True): 36 cmd(cmd_tx) 37 38 39def _qdisc_setup(ifname, qdisc, optargs=""): 40 """Replace root qdisc. Restore the original after the test. 41 42 If the original is mq, children will be of type default_qdisc. 43 """ 44 orig = tc(f"qdisc show dev {ifname} root", json=True)[0].get("kind", None) 45 defer(tc, f"qdisc replace dev {ifname} root {orig}") 46 tc(f"qdisc replace dev {ifname} root {qdisc} {optargs}") 47 48 49def _test_variants_fq(): 50 for ipver in ["4", "6"]: 51 for testcase in [ 52 ["no_delay", "a,-1", "a,-1"], 53 ["zero_delay", "a,0", "a,0"], 54 ["one_pkt", "a,10", "a,10"], 55 ["in_order", "a,10,b,20", "a,10,b,20"], 56 ["reverse_order", "a,20,b,10", "b,10,a,20"], 57 ]: 58 name = f"v{ipver}_{testcase[0]}" 59 yield KsftNamedVariant(name, ipver, testcase[1], testcase[2]) 60 61 62@ksft_variants(_test_variants_fq()) 63def test_so_txtime_fq_mono(cfg, ipver, args_tx, args_rx): 64 """Run all variants of monotonic (fq) tests.""" 65 _qdisc_setup(cfg.ifname, "fq") 66 test_so_txtime(cfg, "mono", ipver, args_tx, args_rx, True) 67 68 69@ksft_variants(_test_variants_fq()) 70def test_so_txtime_fq_tai(cfg, ipver, args_tx, args_rx): 71 """Run all variants of fq tests, but pass CLOCK_TAI to test conversion.""" 72 _qdisc_setup(cfg.ifname, "fq") 73 test_so_txtime(cfg, "tai", ipver, args_tx, args_rx, True) 74 75 76def _test_variants_etf(): 77 for ipver in ["4", "6"]: 78 for testcase in [ 79 ["no_delay", "a,-1", "a,-1", False], 80 ["zero_delay", "a,0", "a,0", False], 81 ["one_pkt", "a,10", "a,10", True], 82 ["in_order", "a,10,b,20", "a,10,b,20", True], 83 ["reverse_order", "a,20,b,10", "b,10,a,20", True], 84 ]: 85 name = f"v{ipver}_{testcase[0]}" 86 yield KsftNamedVariant( 87 name, ipver, testcase[1], testcase[2], testcase[3] 88 ) 89 90 91@ksft_variants(_test_variants_etf()) 92def test_so_txtime_etf(cfg, ipver, args_tx, args_rx, expect_fail): 93 """Run all variants of etf tests.""" 94 try: 95 _qdisc_setup(cfg.ifname, "etf", "clockid CLOCK_TAI delta 400000") 96 except Exception as e: 97 raise KsftSkipEx("tc does not support qdisc etf. skipping") from e 98 99 test_so_txtime(cfg, "tai", ipver, args_tx, args_rx, expect_fail) 100 101 102def main() -> None: 103 """Boilerplate ksft main.""" 104 with NetDrvEpEnv(__file__) as cfg: 105 ksft_run( 106 [test_so_txtime_fq_mono, test_so_txtime_fq_tai, test_so_txtime_etf], 107 args=(cfg,), 108 ) 109 ksft_exit() 110 111 112if __name__ == "__main__": 113 main() 114