tjh.dev
Linux kernel mirror (for testing)
git.kernel.org/pub/scm/linux/kernel/git/torvalds/linux.git
kernel
os
linux
1# SPDX-License-Identifier: GPL-2.0
2
3from lib.py import cmd, ethtool, ip
4from lib.py import ksft_pr, ksft_eq, KsftSkipEx
5from typing import Optional
6import re
7import time
8import json
9
10#The LinkConfig class is implemented to handle the link layer configurations.
11#Required minimum ethtool version is 6.10
12
13class LinkConfig:
14 """Class for handling the link layer configurations"""
15 def __init__(self, cfg: object) -> None:
16 self.cfg = cfg
17 self.partner_netif = self.get_partner_netif_name()
18
19 """Get the initial link configuration of local interface"""
20 self.common_link_modes = self.get_common_link_modes()
21
22 def get_partner_netif_name(self) -> Optional[str]:
23 partner_netif = None
24 try:
25 if not self.verify_link_up():
26 return None
27 """Get partner interface name"""
28 partner_json_output = ip("addr show", json=True, host=self.cfg.remote)
29 for interface in partner_json_output:
30 for addr in interface.get('addr_info', []):
31 if addr.get('local') == self.cfg.remote_addr:
32 partner_netif = interface['ifname']
33 ksft_pr(f"Partner Interface name: {partner_netif}")
34 if partner_netif is None:
35 ksft_pr("Unable to get the partner interface name")
36 except Exception as e:
37 print(f"Unexpected error occurred while getting partner interface name: {e}")
38 self.partner_netif = partner_netif
39 return partner_netif
40
41 def verify_link_up(self) -> bool:
42 """Verify whether the local interface link is up"""
43 with open(f"/sys/class/net/{self.cfg.ifname}/operstate", "r") as fp:
44 link_state = fp.read().strip()
45
46 if link_state == "down":
47 ksft_pr(f"Link state of interface {self.cfg.ifname} is DOWN")
48 return False
49 else:
50 return True
51
52 def reset_interface(self, local: bool = True, remote: bool = True) -> bool:
53 ksft_pr("Resetting interfaces in local and remote")
54 if remote:
55 if self.verify_link_up():
56 if self.partner_netif is not None:
57 ifname = self.partner_netif
58 link_up_cmd = f"ip link set up {ifname}"
59 link_down_cmd = f"ip link set down {ifname}"
60 reset_cmd = f"{link_down_cmd} && sleep 5 && {link_up_cmd}"
61 try:
62 cmd(reset_cmd, host=self.cfg.remote)
63 except Exception as e:
64 ksft_pr(f"Unexpected error occurred while resetting remote: {e}")
65 else:
66 ksft_pr("Partner interface not available")
67 if local:
68 ifname = self.cfg.ifname
69 link_up_cmd = f"ip link set up {ifname}"
70 link_down_cmd = f"ip link set down {ifname}"
71 reset_cmd = f"{link_down_cmd} && sleep 5 && {link_up_cmd}"
72 try:
73 cmd(reset_cmd)
74 except Exception as e:
75 ksft_pr(f"Unexpected error occurred while resetting local: {e}")
76 time.sleep(10)
77 if self.verify_link_up() and self.get_ethtool_field("link-detected"):
78 ksft_pr("Local and remote interfaces reset to original state")
79 return True
80 else:
81 ksft_pr("Error occurred after resetting interfaces. Link is DOWN.")
82 return False
83
84 def set_speed_and_duplex(self, speed: str, duplex: str, autoneg: bool = True) -> bool:
85 """Set the speed and duplex state for the interface"""
86 autoneg_state = "on" if autoneg is True else "off"
87 process = None
88 try:
89 process = ethtool(f"--change {self.cfg.ifname} speed {speed} duplex {duplex} autoneg {autoneg_state}")
90 except Exception as e:
91 ksft_pr(f"Unexpected error occurred while setting speed/duplex: {e}")
92 if process is None or process.ret != 0:
93 return False
94 else:
95 ksft_pr(f"Speed: {speed} Mbps, Duplex: {duplex} set for Interface: {self.cfg.ifname}")
96 return True
97
98 def verify_speed_and_duplex(self, expected_speed: str, expected_duplex: str) -> bool:
99 if not self.verify_link_up():
100 return False
101 """Verifying the speed and duplex state for the interface"""
102 with open(f"/sys/class/net/{self.cfg.ifname}/speed", "r") as fp:
103 actual_speed = fp.read().strip()
104 with open(f"/sys/class/net/{self.cfg.ifname}/duplex", "r") as fp:
105 actual_duplex = fp.read().strip()
106
107 ksft_eq(actual_speed, expected_speed)
108 ksft_eq(actual_duplex, expected_duplex)
109 return True
110
111 def set_autonegotiation_state(self, state: str, remote: bool = False) -> bool:
112 common_link_modes = self.common_link_modes
113 speeds, duplex_modes = self.get_speed_duplex_values(self.common_link_modes)
114 speed = speeds[0]
115 duplex = duplex_modes[0]
116 if not speed or not duplex:
117 ksft_pr("No speed or duplex modes found")
118 return False
119
120 speed_duplex_cmd = f"speed {speed} duplex {duplex}" if state == "off" else ""
121 if remote:
122 if not self.verify_link_up():
123 return False
124 """Set the autonegotiation state for the partner"""
125 command = f"-s {self.partner_netif} {speed_duplex_cmd} autoneg {state}"
126 partner_autoneg_change = None
127 """Set autonegotiation state for interface in remote pc"""
128 try:
129 partner_autoneg_change = ethtool(command, host=self.cfg.remote)
130 except Exception as e:
131 ksft_pr(f"Unexpected error occurred while changing auto-neg in remote: {e}")
132 if partner_autoneg_change is None or partner_autoneg_change.ret != 0:
133 ksft_pr(f"Not able to set autoneg parameter for interface {self.partner_netif}.")
134 return False
135 ksft_pr(f"Autoneg set as {state} for {self.partner_netif}")
136 else:
137 """Set the autonegotiation state for the interface"""
138 try:
139 process = ethtool(f"-s {self.cfg.ifname} {speed_duplex_cmd} autoneg {state}")
140 if process.ret != 0:
141 ksft_pr(f"Not able to set autoneg parameter for interface {self.cfg.ifname}")
142 return False
143 except Exception as e:
144 ksft_pr(f"Unexpected error occurred while changing auto-neg in local: {e}")
145 return False
146 ksft_pr(f"Autoneg set as {state} for {self.cfg.ifname}")
147 return True
148
149 def check_autoneg_supported(self, remote: bool = False) -> bool:
150 if not remote:
151 local_autoneg = self.get_ethtool_field("supports-auto-negotiation")
152 if local_autoneg is None:
153 ksft_pr(f"Unable to fetch auto-negotiation status for interface {self.cfg.ifname}")
154 """Return autoneg status of the local interface"""
155 return local_autoneg
156 else:
157 if not self.verify_link_up():
158 raise KsftSkipEx("Link is DOWN")
159 """Check remote auto-negotiation support status"""
160 partner_autoneg = False
161 if self.partner_netif is not None:
162 partner_autoneg = self.get_ethtool_field("supports-auto-negotiation", remote=True)
163 if partner_autoneg is None:
164 ksft_pr(f"Unable to fetch auto-negotiation status for interface {self.partner_netif}")
165 return partner_autoneg
166
167 def get_common_link_modes(self) -> set[str]:
168 common_link_modes = []
169 """Populate common link modes"""
170 link_modes = self.get_ethtool_field("supported-link-modes")
171 partner_link_modes = self.get_ethtool_field("link-partner-advertised-link-modes")
172 if link_modes is None:
173 raise KsftSkipEx(f"Link modes not available for {self.cfg.ifname}")
174 if partner_link_modes is None:
175 raise KsftSkipEx(f"Partner link modes not available for {self.cfg.ifname}")
176 common_link_modes = set(link_modes) and set(partner_link_modes)
177 return common_link_modes
178
179 def get_speed_duplex_values(self, link_modes: list[str]) -> tuple[list[str], list[str]]:
180 speed = []
181 duplex = []
182 """Check the link modes"""
183 for data in link_modes:
184 parts = data.split('/')
185 speed_value = re.match(r'\d+', parts[0])
186 if speed_value:
187 speed.append(speed_value.group())
188 else:
189 ksft_pr(f"No speed value found for interface {self.ifname}")
190 return None, None
191 duplex.append(parts[1].lower())
192 return speed, duplex
193
194 def get_ethtool_field(self, field: str, remote: bool = False) -> Optional[str]:
195 process = None
196 if not remote:
197 """Get the ethtool field value for the local interface"""
198 try:
199 process = ethtool(self.cfg.ifname, json=True)
200 except Exception as e:
201 ksft_pr("Required minimum ethtool version is 6.10")
202 ksft_pr(f"Unexpected error occurred while getting ethtool field in local: {e}")
203 return None
204 else:
205 if not self.verify_link_up():
206 return None
207 """Get the ethtool field value for the remote interface"""
208 self.cfg.require_cmd("ethtool", remote=True)
209 if self.partner_netif is None:
210 ksft_pr(f"Partner interface name is unavailable.")
211 return None
212 try:
213 process = ethtool(self.partner_netif, json=True, host=self.cfg.remote)
214 except Exception as e:
215 ksft_pr("Required minimum ethtool version is 6.10")
216 ksft_pr(f"Unexpected error occurred while getting ethtool field in remote: {e}")
217 return None
218 json_data = process[0]
219 """Check if the field exist in the json data"""
220 if field not in json_data:
221 raise KsftSkipEx(f'Field {field} does not exist in the output of interface {json_data["ifname"]}')
222 return json_data[field]