2 # SPDX-License-Identifier: GPL-2.0
12 def test_receive(cfg, ipv4=False, extra_args=None):  argument
14     if not cfg.have_rx_csum:
15         raise KsftSkipEx(f"Test requires rx checksum offload on {cfg.ifname}")
18         ip_args = f"-4 -S {cfg.remote_v4} -D {cfg.v4}"
20         ip_args = f"-6 -S {cfg.remote_v6} -D {cfg.v6}"
22     rx_cmd = f"{cfg.bin_local} -i {cfg.ifname} -n 100 {ip_args} -r 1 -R {extra_args}"
23     tx_cmd = f"{cfg.bin_remote} -i {cfg.ifname} -n 100 {ip_args} -r 1 -T {extra_args}"
27         cmd(tx_cmd, host=cfg.remote)
30 def test_transmit(cfg, ipv4=False, extra_args=None):  argument
32     if (not cfg.have_tx_csum_generic and
33         not (cfg.have_tx_csum_ipv4 and ipv4) and
34         not (cfg.have_tx_csum_ipv6 and not ipv4)):
35         raise KsftSkipEx(f"Test requires tx checksum offload on {cfg.ifname}")
38         ip_args = f"-4 -S {cfg.v4} -D {cfg.remote_v4}"
40         ip_args = f"-6 -S {cfg.v6} -D {cfg.remote_v6}"
43     if extra_args != "-U -Z":
44         extra_args += " -r 1"
46     rx_cmd = f"{cfg.bin_remote} -i {cfg.ifname} -L 1 -n 100 {ip_args} -R {extra_args}"
47     tx_cmd = f"{cfg.bin_local} -i {cfg.ifname} -L 1 -n 100 {ip_args} -T {extra_args}"
49     with bkg(rx_cmd, host=cfg.remote, exit_wait=True):
50         wait_port_listen(34000, proto="udp", host=cfg.remote)
54 def test_builder(name, cfg, ipv4=False, tx=False, extra_args=""):  argument
59     def f(cfg):  argument
61             cfg.require_v4()
63             cfg.require_v6()
66             test_transmit(cfg, ipv4, extra_args)
68             test_receive(cfg, ipv4, extra_args)
77 def check_nic_features(cfg) -> None:  argument
81     cfg.have_tx_csum_generic = False
82     cfg.have_tx_csum_ipv4 = False
83     cfg.have_tx_csum_ipv6 = False
84     cfg.have_rx_csum = False
87     features = ethnl.features_get({"header": {"dev-index": cfg.ifindex}})
89         if f["name"] == "tx-checksum-ip-generic":
90             cfg.have_tx_csum_generic = True
91         elif f["name"] == "tx-checksum-ipv4":
92             cfg.have_tx_csum_ipv4 = True
93         elif f["name"] == "tx-checksum-ipv6":
94             cfg.have_tx_csum_ipv6 = True
95         elif f["name"] == "rx-checksum":
96             cfg.have_rx_csum = True
99 def main() -> None:
100     with NetDrvEpEnv(__file__, nsim_test=False) as cfg:
101         check_nic_features(cfg)
103         cfg.bin_local = path.abspath(path.dirname(__file__) + "/../../../net/lib/csum")
104         cfg.bin_remote = cfg.remote.deploy(cfg.bin_local)
108             cases.append(test_builder("rx_tcp", cfg, ipv4, False, "-t"))
109             cases.append(test_builder("rx_tcp_invalid", cfg, ipv4, False, "-t -E"))
111             cases.append(test_builder("rx_udp", cfg, ipv4, False, ""))
112             cases.append(test_builder("rx_udp_invalid", cfg, ipv4, False, "-E"))
114             cases.append(test_builder("tx_udp_csum_offload", cfg, ipv4, True, "-U"))
115             cases.append(test_builder("tx_udp_zero_checksum", cfg, ipv4, True, "-U -Z"))
117         ksft_run(cases=cases, args=(cfg, ))