NVMe testcase fixes for ASAP VMs

NVMe testcase fixes for ASAP VMs

lisa/features/nvme.py

@@ -16,6 +16,7 @@
 from lisa.tools import Ls, Lspci, Nvmecli
 from lisa.tools.lspci import PciDevice
 from lisa.util import field_metadata, get_matched_str
+from lisa.util.constants import DEVICE_TYPE_NVME
 
 
 class Nvme(Feature):
@@ -42,6 +43,9 @@ class Nvme(Feature):
     # /dev/nvme0n1p15 -> /dev/nvme0n1
     NVME_NAMESPACE_PATTERN = re.compile(r"/dev/nvme[0-9]+n[0-9]+", re.M)
 
+    # /dev/nvme0n1p15 -> /dev/nvme0n1
+    NVME_DEVICE_PATTERN = re.compile(r"/dev/nvme[0-9]+", re.M)
+
     _pci_device_name = "Non-Volatile memory controller"
     _ls_devices: str = ""
 
@@ -63,6 +67,11 @@ def get_devices(self) -> List[str]:
             matched_result = self._device_pattern.match(row)
             if matched_result:
                 devices_list.append(matched_result.group("device_name"))
+        node_disk = self._node.features[Disk]
+        if node_disk.get_os_disk_controller_type() == schema.DiskControllerType.NVME:
+            os_disk_nvme_device = self.get_os_disk_nvme_device()
+            # Removing OS disk/device from the list.
+            devices_list.remove(os_disk_nvme_device)
         return devices_list
 
     def get_namespaces(self) -> List[str]:
@@ -78,7 +87,13 @@ def get_namespaces(self) -> List[str]:
         return namespaces
 
     def get_namespaces_from_cli(self) -> List[str]:
-        return self._node.tools[Nvmecli].get_namespaces()
+        namespaces_list = self._node.tools[Nvmecli].get_namespaces()
+        node_disk = self._node.features[Disk]
+        if node_disk.get_os_disk_controller_type() == schema.DiskControllerType.NVME:
+            os_disk_nvme_namespace = self.get_os_disk_nvme_namespace()
+            # Removing OS disk/device from the list.
+            namespaces_list.remove(os_disk_nvme_namespace)
+        return namespaces_list
 
     def get_os_disk_nvme_namespace(self) -> str:
         node_disk = self._node.features[Disk]
@@ -93,10 +108,22 @@ def get_os_disk_nvme_namespace(self) -> str:
             )
         return os_partition_namespace
 
+    def get_os_disk_nvme_device(self) -> str:
+        os_disk_nvme_namespace = self.get_os_disk_nvme_namespace()
+        # Sample os_boot_partition when disc controller type is NVMe:
+        # name: /dev/nvme0n1p15, disk: nvme, mount_point: /boot/efi, type: vfat
+        if os_disk_nvme_namespace:
+            os_disk_nvme_device = get_matched_str(
+                os_disk_nvme_namespace,
+                self.NVME_DEVICE_PATTERN,
+            )
+        return os_disk_nvme_device
+
     def get_devices_from_lspci(self) -> List[PciDevice]:
         devices_from_lspci = []
         lspci_tool = self._node.tools[Lspci]
         device_list = lspci_tool.get_devices()
+        device_list = lspci_tool.get_devices_by_type(DEVICE_TYPE_NVME, use_pci_ids=True)
         devices_from_lspci = [
             x for x in device_list if self._pci_device_name == x.device_class
         ]

lisa/tools/lspci.py

@@ -8,12 +8,7 @@
 from lisa.executable import Tool
 from lisa.operating_system import Posix
 from lisa.tools import Echo
-from lisa.util import (
-    LisaException,
-    constants,
-    find_patterns_in_lines,
-    get_matched_str,
-)
+from lisa.util import LisaException, constants, find_patterns_in_lines, get_matched_str
 
 # Example output of lspci command -
 # lspci -m
@@ -129,13 +124,20 @@ def parse(self, raw_str: str, pci_ids: Dict[str, Any]) -> None:
         matched_pci_device_info = PATTERN_PCI_DEVICE.match(raw_str)
         if matched_pci_device_info:
             self.slot = matched_pci_device_info.group("slot")
+            assert self.slot, f"Can not find slot info for: {raw_str}"
             self.device_class = matched_pci_device_info.group("device_class")
+            assert self.device_class, f"Can not find device class for: {raw_str}"
             self.vendor = matched_pci_device_info.group("vendor")
+            assert self.vendor, f"Can not find vendor info for: {raw_str}"
             self.device_info = matched_pci_device_info.group("device")
+            assert self.device_info, f"Can not find device info for: {raw_str}"
             if pci_ids:
                 self.device_id = pci_ids[self.slot]["device_id"]
+                assert self.device_id, f"cannot find device id from {raw_str}"
                 self.vendor_id = pci_ids[self.slot]["vendor_id"]
+                assert self.vendor_id, f"cannot find vendor id from {raw_str}"
                 self.controller_id = pci_ids[self.slot]["controller_id"]
+                assert self.controller_id, f"cannot findcontroller_id from {raw_str}"
         else:
             raise LisaException("cannot find any matched pci devices")
 
@@ -163,13 +165,25 @@ def _install(self) -> bool:
         return self._check_exists()
 
     def get_device_names_by_type(
-        self, device_type: str, force_run: bool = False
+        self, device_type: str, force_run: bool = False, use_pci_ids: bool = False
     ) -> List[str]:
         if device_type.upper() not in DEVICE_TYPE_DICT.keys():
             raise LisaException(f"pci_type '{device_type}' is not recognized.")
         class_names = DEVICE_TYPE_DICT[device_type.upper()]
         devices_list = self.get_devices(force_run)
-        devices_slots = [x.slot for x in devices_list if x.device_class in class_names]
+        devices_slots = []
+        if use_pci_ids:
+            for device in devices_list:
+                if (
+                    device.controller_id in CONTROLLER_ID_DICT[device_type.upper()]
+                    and device.vendor_id in VENDOR_ID_DICT[device_type.upper()]
+                    and device.device_id in DEVICE_ID_DICT[device_type.upper()]
+                ):
+                    devices_slots.append(device.slot)
+        else:
+            devices_slots = [
+                x.slot for x in devices_list if x.device_class in class_names
+            ]
         return devices_slots
 
     def get_devices_by_type(
@@ -196,6 +210,7 @@ def get_devices_by_type(
             ]
         return device_type_list
 
+    @retry(KeyError, tries=10, delay=20)
     def get_devices(self, force_run: bool = False) -> List[PciDevice]:
         if (not self._pci_devices) or force_run:
             self._pci_devices = []
@@ -318,7 +333,7 @@ class LspciBSD(Lspci):
     _disabled_devices: Set[str] = set()
 
     def get_device_names_by_type(
-        self, device_type: str, force_run: bool = False
+        self, device_type: str, force_run: bool = False, use_pci_ids: bool = False
     ) -> List[str]:
         output = self.node.execute("pciconf -l", sudo=True).stdout
         if device_type.upper() not in self._DEVICE_DRIVER_MAPPING.keys():