Use ReadLink in NIC.py

Author: adityagesh

lisa/nic.py

@@ -13,7 +13,7 @@
 from assertpy import assert_that
 from retry import retry
 
-from lisa.tools import Cat, Ip, KernelConfig, Ls, Lspci, Modprobe, Tee
+from lisa.tools import Cat, Ip, KernelConfig, Ls, Lspci, Modprobe, Readlink, Tee
 from lisa.util import InitializableMixin, LisaException, constants, find_groups_in_lines
 
 if TYPE_CHECKING:
@@ -121,11 +121,6 @@ class Nics(InitializableMixin):
         )
     )
 
-    # PCI slot pattern: XXXX:XX:XX.X (e.g., 34da:00:02.0)
-    __pci_slot_pattern = re.compile(
-        r"([a-zA-Z0-9]{4}:[a-zA-Z0-9]{2}:[a-zA-Z0-9]{2}\.[a-zA-Z0-9])"
-    )
-
     _file_not_exist = re.compile(r"No such file or directory", re.MULTILINE)
 
     # ConnectX-3 uses mlx4_core
@@ -232,10 +227,12 @@ def get_nic_driver(self, nic_name: str) -> str:
         # get the current driver for the nic from the node
         # sysfs provides a link to the driver entry at device/driver
         nic = self.get_nic(nic_name)
-        cmd = f"readlink -f /sys/class/net/{nic_name}/device/driver"
+        readlink = self._node.tools[Readlink]
         # ex return value:
         # /sys/bus/vmbus/drivers/hv_netvsc
-        found_link = self._node.execute(cmd, expected_exit_code=0).stdout
+        found_link = readlink.get_canonical_path(
+            f"/sys/class/net/{nic_name}/device/driver"
+        )
         assert_that(found_link).described_as(
             f"sysfs check for NIC device {nic_name} driver returned no output"
         ).is_not_equal_to("")
@@ -400,8 +397,9 @@ def _get_nic_names(self) -> List[str]:
         return non_virtual_nics
 
     def _get_nic_uuid(self, nic_name: str) -> str:
-        full_dev_path = self._node.execute(f"readlink /sys/class/net/{nic_name}/device")
-        uuid = os.path.basename(full_dev_path.stdout.strip())
+        readlink = self._node.tools[Readlink]
+        full_dev_path = readlink.get_target(f"/sys/class/net/{nic_name}/device")
+        uuid = os.path.basename(full_dev_path)
         self._node.log.debug(f"{nic_name} UUID:{uuid}")
         return uuid
 
@@ -519,6 +517,7 @@ def _discover_standalone_pci_nics(self, lspci: Lspci) -> None:
         """
         # Get unpaired NICs that might have PCI devices
         unpaired_nics = self.get_unpaired_devices()
+        readlink = self._node.tools[Readlink]
 
         for nic_name in unpaired_nics:
             nic = self.nics[nic_name]
@@ -527,16 +526,15 @@ def _discover_standalone_pci_nics(self, lspci: Lspci) -> None:
                 continue
 
             # Try to find the PCI slot for this NIC by checking its device path
-            result = self._node.execute(
-                f"readlink -f /sys/class/net/{nic_name}/device",
-                shell=True,
+            device_path = readlink.get_canonical_path(
+                f"/sys/class/net/{nic_name}/device",
+                no_error_log=True,
             )
-            if result.exit_code == 0 and result.stdout.strip():
-                # Extract PCI slot from device path
+            if device_path:
+                # Extract PCI slot from device path using lspci tool
                 # Path format: /sys/devices/.../XXXX:XX:XX.X/net/nicname
-                match = self.__pci_slot_pattern.search(result.stdout)
-                if match:
-                    pci_slot = match.group(1)
+                pci_slot = lspci.get_pci_slot_from_device_path(device_path)
+                if pci_slot:
                     # Get the module name for this PCI device
                     try:
                         module_name = lspci.get_used_module(pci_slot)
@@ -561,9 +559,10 @@ def _discover_standalone_pci_nics(self, lspci: Lspci) -> None:
                 else:
                     self._node.log.debug(
                         f"Could not extract PCI slot from device path for {nic_name}: "
-                        f"{result.stdout.strip()}"
+                        f"{device_path}"
                     )
 
+
     def _get_default_nic(self) -> None:
         self.default_nic: str = ""
         self.default_nic_route: str = ""

lisa/tools/__init__.py

@@ -104,6 +104,7 @@
 from .python import Pip, Python
 from .qemu import Qemu
 from .qemu_img import QemuImg
+from .readlink import Readlink
 from .reboot import Reboot
 from .remote_copy import RemoteCopy
 from .resize_partition import ResizePartition
@@ -242,6 +243,7 @@
     "Qemu",
     "QemuImg",
     "Reboot",
+    "Readlink",
     "RemoteCopy",
     "ResizePartition",
     "Rpm",

lisa/tools/lspci.py

@@ -138,6 +138,12 @@
 # Kernel driver in use: mlx5_core\r
 PATTERN_MODULE_IN_USE = re.compile(r"Kernel driver in use: ([A-Za-z0-9_-]*)", re.M)
 
+# PCI slot pattern for extracting slot from device paths
+# Example: /sys/devices/pci0000:00/0000:00:02.0/net/eth0 -> 0000:00:02.0
+PATTERN_PCI_SLOT = re.compile(
+    r"([a-fA-F0-9]{4}:[a-fA-F0-9]{2}:[a-fA-F0-9]{2}\.[a-fA-F0-9])"
+)
+
 
 class PciDevice:
     def __init__(self, pci_device_raw: str) -> None:
@@ -402,6 +408,13 @@ def get_devices_by_vendor_device_id(
                 devices_list.append(device)
         return devices_list
 
+    def get_pci_slot_from_device_path(self, device_path: str) -> Optional[str]:
+        """
+        Extract PCI slot information from a device path.
+        """
+        pci_slot = get_matched_str(device_path, PATTERN_PCI_SLOT)
+        return pci_slot if pci_slot else None
+
 
 class LspciBSD(Lspci):
     _DEVICE_DRIVER_MAPPING: Dict[str, Pattern[str]] = {