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] 223