mirror of https://github.com/torvalds/linux.git
223 lines
10 KiB
Python
223 lines
10 KiB
Python
# SPDX-License-Identifier: GPL-2.0
|
|
|
|
from lib.py import cmd, ethtool, ip
|
|
from lib.py import ksft_pr, ksft_eq, KsftSkipEx
|
|
from typing import Optional
|
|
import re
|
|
import time
|
|
import json
|
|
|
|
#The LinkConfig class is implemented to handle the link layer configurations.
|
|
#Required minimum ethtool version is 6.10
|
|
|
|
class LinkConfig:
|
|
"""Class for handling the link layer configurations"""
|
|
def __init__(self, cfg: object) -> None:
|
|
self.cfg = cfg
|
|
self.partner_netif = self.get_partner_netif_name()
|
|
|
|
"""Get the initial link configuration of local interface"""
|
|
self.common_link_modes = self.get_common_link_modes()
|
|
|
|
def get_partner_netif_name(self) -> Optional[str]:
|
|
partner_netif = None
|
|
try:
|
|
if not self.verify_link_up():
|
|
return None
|
|
"""Get partner interface name"""
|
|
partner_json_output = ip("addr show", json=True, host=self.cfg.remote)
|
|
for interface in partner_json_output:
|
|
for addr in interface.get('addr_info', []):
|
|
if addr.get('local') == self.cfg.remote_addr:
|
|
partner_netif = interface['ifname']
|
|
ksft_pr(f"Partner Interface name: {partner_netif}")
|
|
if partner_netif is None:
|
|
ksft_pr("Unable to get the partner interface name")
|
|
except Exception as e:
|
|
print(f"Unexpected error occurred while getting partner interface name: {e}")
|
|
self.partner_netif = partner_netif
|
|
return partner_netif
|
|
|
|
def verify_link_up(self) -> bool:
|
|
"""Verify whether the local interface link is up"""
|
|
with open(f"/sys/class/net/{self.cfg.ifname}/operstate", "r") as fp:
|
|
link_state = fp.read().strip()
|
|
|
|
if link_state == "down":
|
|
ksft_pr(f"Link state of interface {self.cfg.ifname} is DOWN")
|
|
return False
|
|
else:
|
|
return True
|
|
|
|
def reset_interface(self, local: bool = True, remote: bool = True) -> bool:
|
|
ksft_pr("Resetting interfaces in local and remote")
|
|
if remote:
|
|
if self.verify_link_up():
|
|
if self.partner_netif is not None:
|
|
ifname = self.partner_netif
|
|
link_up_cmd = f"ip link set up {ifname}"
|
|
link_down_cmd = f"ip link set down {ifname}"
|
|
reset_cmd = f"{link_down_cmd} && sleep 5 && {link_up_cmd}"
|
|
try:
|
|
cmd(reset_cmd, host=self.cfg.remote)
|
|
except Exception as e:
|
|
ksft_pr(f"Unexpected error occurred while resetting remote: {e}")
|
|
else:
|
|
ksft_pr("Partner interface not available")
|
|
if local:
|
|
ifname = self.cfg.ifname
|
|
link_up_cmd = f"ip link set up {ifname}"
|
|
link_down_cmd = f"ip link set down {ifname}"
|
|
reset_cmd = f"{link_down_cmd} && sleep 5 && {link_up_cmd}"
|
|
try:
|
|
cmd(reset_cmd)
|
|
except Exception as e:
|
|
ksft_pr(f"Unexpected error occurred while resetting local: {e}")
|
|
time.sleep(10)
|
|
if self.verify_link_up() and self.get_ethtool_field("link-detected"):
|
|
ksft_pr("Local and remote interfaces reset to original state")
|
|
return True
|
|
else:
|
|
ksft_pr("Error occurred after resetting interfaces. Link is DOWN.")
|
|
return False
|
|
|
|
def set_speed_and_duplex(self, speed: str, duplex: str, autoneg: bool = True) -> bool:
|
|
"""Set the speed and duplex state for the interface"""
|
|
autoneg_state = "on" if autoneg is True else "off"
|
|
process = None
|
|
try:
|
|
process = ethtool(f"--change {self.cfg.ifname} speed {speed} duplex {duplex} autoneg {autoneg_state}")
|
|
except Exception as e:
|
|
ksft_pr(f"Unexpected error occurred while setting speed/duplex: {e}")
|
|
if process is None or process.ret != 0:
|
|
return False
|
|
else:
|
|
ksft_pr(f"Speed: {speed} Mbps, Duplex: {duplex} set for Interface: {self.cfg.ifname}")
|
|
return True
|
|
|
|
def verify_speed_and_duplex(self, expected_speed: str, expected_duplex: str) -> bool:
|
|
if not self.verify_link_up():
|
|
return False
|
|
"""Verifying the speed and duplex state for the interface"""
|
|
with open(f"/sys/class/net/{self.cfg.ifname}/speed", "r") as fp:
|
|
actual_speed = fp.read().strip()
|
|
with open(f"/sys/class/net/{self.cfg.ifname}/duplex", "r") as fp:
|
|
actual_duplex = fp.read().strip()
|
|
|
|
ksft_eq(actual_speed, expected_speed)
|
|
ksft_eq(actual_duplex, expected_duplex)
|
|
return True
|
|
|
|
def set_autonegotiation_state(self, state: str, remote: bool = False) -> bool:
|
|
common_link_modes = self.common_link_modes
|
|
speeds, duplex_modes = self.get_speed_duplex_values(self.common_link_modes)
|
|
speed = speeds[0]
|
|
duplex = duplex_modes[0]
|
|
if not speed or not duplex:
|
|
ksft_pr("No speed or duplex modes found")
|
|
return False
|
|
|
|
speed_duplex_cmd = f"speed {speed} duplex {duplex}" if state == "off" else ""
|
|
if remote:
|
|
if not self.verify_link_up():
|
|
return False
|
|
"""Set the autonegotiation state for the partner"""
|
|
command = f"-s {self.partner_netif} {speed_duplex_cmd} autoneg {state}"
|
|
partner_autoneg_change = None
|
|
"""Set autonegotiation state for interface in remote pc"""
|
|
try:
|
|
partner_autoneg_change = ethtool(command, host=self.cfg.remote)
|
|
except Exception as e:
|
|
ksft_pr(f"Unexpected error occurred while changing auto-neg in remote: {e}")
|
|
if partner_autoneg_change is None or partner_autoneg_change.ret != 0:
|
|
ksft_pr(f"Not able to set autoneg parameter for interface {self.partner_netif}.")
|
|
return False
|
|
ksft_pr(f"Autoneg set as {state} for {self.partner_netif}")
|
|
else:
|
|
"""Set the autonegotiation state for the interface"""
|
|
try:
|
|
process = ethtool(f"-s {self.cfg.ifname} {speed_duplex_cmd} autoneg {state}")
|
|
if process.ret != 0:
|
|
ksft_pr(f"Not able to set autoneg parameter for interface {self.cfg.ifname}")
|
|
return False
|
|
except Exception as e:
|
|
ksft_pr(f"Unexpected error occurred while changing auto-neg in local: {e}")
|
|
return False
|
|
ksft_pr(f"Autoneg set as {state} for {self.cfg.ifname}")
|
|
return True
|
|
|
|
def check_autoneg_supported(self, remote: bool = False) -> bool:
|
|
if not remote:
|
|
local_autoneg = self.get_ethtool_field("supports-auto-negotiation")
|
|
if local_autoneg is None:
|
|
ksft_pr(f"Unable to fetch auto-negotiation status for interface {self.cfg.ifname}")
|
|
"""Return autoneg status of the local interface"""
|
|
return local_autoneg
|
|
else:
|
|
if not self.verify_link_up():
|
|
raise KsftSkipEx("Link is DOWN")
|
|
"""Check remote auto-negotiation support status"""
|
|
partner_autoneg = False
|
|
if self.partner_netif is not None:
|
|
partner_autoneg = self.get_ethtool_field("supports-auto-negotiation", remote=True)
|
|
if partner_autoneg is None:
|
|
ksft_pr(f"Unable to fetch auto-negotiation status for interface {self.partner_netif}")
|
|
return partner_autoneg
|
|
|
|
def get_common_link_modes(self) -> set[str]:
|
|
common_link_modes = []
|
|
"""Populate common link modes"""
|
|
link_modes = self.get_ethtool_field("supported-link-modes")
|
|
partner_link_modes = self.get_ethtool_field("link-partner-advertised-link-modes")
|
|
if link_modes is None:
|
|
raise KsftSkipEx(f"Link modes not available for {self.cfg.ifname}")
|
|
if partner_link_modes is None:
|
|
raise KsftSkipEx(f"Partner link modes not available for {self.cfg.ifname}")
|
|
common_link_modes = set(link_modes) and set(partner_link_modes)
|
|
return common_link_modes
|
|
|
|
def get_speed_duplex_values(self, link_modes: list[str]) -> tuple[list[str], list[str]]:
|
|
speed = []
|
|
duplex = []
|
|
"""Check the link modes"""
|
|
for data in link_modes:
|
|
parts = data.split('/')
|
|
speed_value = re.match(r'\d+', parts[0])
|
|
if speed_value:
|
|
speed.append(speed_value.group())
|
|
else:
|
|
ksft_pr(f"No speed value found for interface {self.ifname}")
|
|
return None, None
|
|
duplex.append(parts[1].lower())
|
|
return speed, duplex
|
|
|
|
def get_ethtool_field(self, field: str, remote: bool = False) -> Optional[str]:
|
|
process = None
|
|
if not remote:
|
|
"""Get the ethtool field value for the local interface"""
|
|
try:
|
|
process = ethtool(self.cfg.ifname, json=True)
|
|
except Exception as e:
|
|
ksft_pr("Required minimum ethtool version is 6.10")
|
|
ksft_pr(f"Unexpected error occurred while getting ethtool field in local: {e}")
|
|
return None
|
|
else:
|
|
if not self.verify_link_up():
|
|
return None
|
|
"""Get the ethtool field value for the remote interface"""
|
|
self.cfg.require_cmd("ethtool", remote=True)
|
|
if self.partner_netif is None:
|
|
ksft_pr(f"Partner interface name is unavailable.")
|
|
return None
|
|
try:
|
|
process = ethtool(self.partner_netif, json=True, host=self.cfg.remote)
|
|
except Exception as e:
|
|
ksft_pr("Required minimum ethtool version is 6.10")
|
|
ksft_pr(f"Unexpected error occurred while getting ethtool field in remote: {e}")
|
|
return None
|
|
json_data = process[0]
|
|
"""Check if the field exist in the json data"""
|
|
if field not in json_data:
|
|
raise KsftSkipEx(f"Field {field} does not exist in the output of interface {json_data["ifname"]}")
|
|
return json_data[field]
|