1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
|
#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-2.0
"""Run the tools/testing/selftests/net/csum testsuite."""
from os import path
from lib.py import ksft_run, ksft_exit, KsftSkipEx
from lib.py import EthtoolFamily, NetDrvEpEnv
from lib.py import bkg, cmd, wait_port_listen
def test_receive(cfg, ipv4=False, extra_args=None):
"""Test local nic checksum receive. Remote host sends crafted packets."""
if not cfg.have_rx_csum:
raise KsftSkipEx(f"Test requires rx checksum offload on {cfg.ifname}")
if ipv4:
ip_args = f"-4 -S {cfg.remote_v4} -D {cfg.v4}"
else:
ip_args = f"-6 -S {cfg.remote_v6} -D {cfg.v6}"
rx_cmd = f"{cfg.bin_local} -i {cfg.ifname} -n 100 {ip_args} -r 1 -R {extra_args}"
tx_cmd = f"{cfg.bin_remote} -i {cfg.ifname} -n 100 {ip_args} -r 1 -T {extra_args}"
with bkg(rx_cmd, exit_wait=True):
wait_port_listen(34000, proto="udp")
cmd(tx_cmd, host=cfg.remote)
def test_transmit(cfg, ipv4=False, extra_args=None):
"""Test local nic checksum transmit. Remote host verifies packets."""
if (not cfg.have_tx_csum_generic and
not (cfg.have_tx_csum_ipv4 and ipv4) and
not (cfg.have_tx_csum_ipv6 and not ipv4)):
raise KsftSkipEx(f"Test requires tx checksum offload on {cfg.ifname}")
if ipv4:
ip_args = f"-4 -S {cfg.v4} -D {cfg.remote_v4}"
else:
ip_args = f"-6 -S {cfg.v6} -D {cfg.remote_v6}"
# Cannot randomize input when calculating zero checksum
if extra_args != "-U -Z":
extra_args += " -r 1"
rx_cmd = f"{cfg.bin_remote} -i {cfg.ifname} -L 1 -n 100 {ip_args} -R {extra_args}"
tx_cmd = f"{cfg.bin_local} -i {cfg.ifname} -L 1 -n 100 {ip_args} -T {extra_args}"
with bkg(rx_cmd, host=cfg.remote, exit_wait=True):
wait_port_listen(34000, proto="udp", host=cfg.remote)
cmd(tx_cmd)
def test_builder(name, cfg, ipv4=False, tx=False, extra_args=""):
"""Construct specific tests from the common template.
Most tests follow the same basic pattern, differing only in
Direction of the test and optional flags passed to csum."""
def f(cfg):
if ipv4:
cfg.require_v4()
else:
cfg.require_v6()
if tx:
test_transmit(cfg, ipv4, extra_args)
else:
test_receive(cfg, ipv4, extra_args)
if ipv4:
f.__name__ = "ipv4_" + name
else:
f.__name__ = "ipv6_" + name
return f
def check_nic_features(cfg) -> None:
"""Test whether Tx and Rx checksum offload are enabled.
If the device under test has either off, then skip the relevant tests."""
cfg.have_tx_csum_generic = False
cfg.have_tx_csum_ipv4 = False
cfg.have_tx_csum_ipv6 = False
cfg.have_rx_csum = False
ethnl = EthtoolFamily()
features = ethnl.features_get({"header": {"dev-index": cfg.ifindex}})
for f in features["active"]["bits"]["bit"]:
if f["name"] == "tx-checksum-ip-generic":
cfg.have_tx_csum_generic = True
elif f["name"] == "tx-checksum-ipv4":
cfg.have_tx_csum_ipv4 = True
elif f["name"] == "tx-checksum-ipv6":
cfg.have_tx_csum_ipv6 = True
elif f["name"] == "rx-checksum":
cfg.have_rx_csum = True
def main() -> None:
with NetDrvEpEnv(__file__, nsim_test=False) as cfg:
check_nic_features(cfg)
cfg.bin_local = path.abspath(path.dirname(__file__) + "/../../../net/lib/csum")
cfg.bin_remote = cfg.remote.deploy(cfg.bin_local)
cases = []
for ipv4 in [True, False]:
cases.append(test_builder("rx_tcp", cfg, ipv4, False, "-t"))
cases.append(test_builder("rx_tcp_invalid", cfg, ipv4, False, "-t -E"))
cases.append(test_builder("rx_udp", cfg, ipv4, False, ""))
cases.append(test_builder("rx_udp_invalid", cfg, ipv4, False, "-E"))
cases.append(test_builder("tx_udp_csum_offload", cfg, ipv4, True, "-U"))
cases.append(test_builder("tx_udp_zero_checksum", cfg, ipv4, True, "-U -Z"))
ksft_run(cases=cases, args=(cfg, ))
ksft_exit()
if __name__ == "__main__":
main()
|
