Disk finder refactor

Signed-off-by: Katarzyna Treder <katarzyna.treder@h-partners.com>

core/test_run_utils.py

@@ -19,7 +19,7 @@ from core.pair_testing import generate_pair_testing_testcases, register_testcase
 from core.plugins import PluginManager
 from log.base_log import BaseLogResult
 from storage_devices.disk import Disk
-from test_utils import disk_finder
+from test_tools import disk_finder
 from test_utils.dut import Dut
 
 TestRun = core.test_run.TestRun

storage_devices/disk.py

@@ -13,9 +13,9 @@ from enum import IntEnum
 from core.test_run import TestRun
 from storage_devices.device import Device
 from test_tools import disk_utils, fs_utils, nvme_cli
-from test_utils import disk_finder
 from test_tools.common.wait import wait
 from connection.utils.output import Output
+from test_tools.disk_finder import get_block_devices_list, resolve_to_by_id_link
 from types.size import Unit
 
 
@@ -159,7 +159,7 @@ class Disk(Device):
 
     def is_detected(self):
         if self.serial_number:
-            serial_numbers = disk_finder.get_all_serial_numbers()
+            serial_numbers = Disk.get_all_serial_numbers()
             return self.serial_number in serial_numbers
         elif self.path:
             output = fs_utils.ls_item(f"{self.path}")
@@ -214,6 +214,40 @@ class Disk(Device):
         for disk_type in cls.types_registry:
             disk_type.plug_all()
 
+    @staticmethod
+    def get_all_serial_numbers():
+        serial_numbers = {}
+        block_devices = get_block_devices_list()
+        for dev in block_devices:
+            serial = Disk.get_disk_serial_number(dev)
+            try:
+                path = resolve_to_by_id_link(dev)
+            except Exception:
+                continue
+            if serial:
+                serial_numbers[serial] = path
+            else:
+                TestRun.LOGGER.warning(f"Device {path} ({dev}) does not have a serial number.")
+                serial_numbers[path] = path
+        return serial_numbers
+
+    @staticmethod
+    def get_disk_serial_number(dev_path):
+        commands = [
+            f"(udevadm info --query=all --name={dev_path} | grep 'SCSI.*_SERIAL' || "
+            f"udevadm info --query=all --name={dev_path} | grep 'ID_SERIAL_SHORT') | "
+            "awk -F '=' '{print $NF}'",
+            f"sg_inq {dev_path} 2> /dev/null | grep '[Ss]erial number:' | "
+            "awk '{print $NF}'",
+            f"udevadm info --query=all --name={dev_path} | grep 'ID_SERIAL' | "
+            "awk -F '=' '{print $NF}'"
+        ]
+        for command in commands:
+            serial = TestRun.executor.run(command).stdout
+            if serial:
+                return serial.split('\n')[0]
+        return None
+
 
 @static_init
 class NvmeDisk(Disk):

storage_devices/error_device.py

@@ -7,7 +7,7 @@
 from core.test_run import TestRun
 from storage_devices.device import Device
 from test_tools.device_mapper import DmTable, DeviceMapper
-from test_utils.disk_finder import resolve_to_by_id_link
+from test_tools.disk_finder import resolve_to_by_id_link
 
 
 class ErrorDevice(Device):

storage_devices/lvm.py

@@ -11,7 +11,7 @@ from core.test_run import TestRun
 from storage_devices.device import Device
 from storage_devices.disk import Disk
 from test_tools.fs_utils import readlink
-from test_utils.disk_finder import resolve_to_by_id_link, get_system_disks
+from test_tools.disk_finder import resolve_to_by_id_link, get_system_disks
 from test_utils.filesystem.symlink import Symlink
 from types.size import Size
 

storage_devices/raid.py

@@ -10,7 +10,7 @@ from storage_devices.device import Device
 from storage_devices.disk import Disk
 from test_tools.fs_utils import readlink
 from test_tools.mdadm import Mdadm
-from test_utils.disk_finder import resolve_to_by_id_link
+from test_tools.disk_finder import resolve_to_by_id_link
 from types.size import Size, Unit
 
 

test_tools/disk_finder.py

@@ -1,5 +1,6 @@
 #
 # Copyright(c) 2019-2021 Intel Corporation
+# Copyright(c) 2024 Huawei Technologies Co., Ltd.
 # SPDX-License-Identifier: BSD-3-Clause
 #
 import os
@@ -103,40 +104,6 @@ def discover_ssd_devices(block_devices, devices_res):
             block_devices.remove(dev)
 
 
-def get_disk_serial_number(dev_path):
-    commands = [
-        f"(udevadm info --query=all --name={dev_path} | grep 'SCSI.*_SERIAL' || "
-        f"udevadm info --query=all --name={dev_path} | grep 'ID_SERIAL_SHORT') | "
-        "awk -F '=' '{print $NF}'",
-        f"sg_inq {dev_path} 2> /dev/null | grep '[Ss]erial number:' | "
-        "awk '{print $NF}'",
-        f"udevadm info --query=all --name={dev_path} | grep 'ID_SERIAL' | "
-        "awk -F '=' '{print $NF}'"
-    ]
-    for command in commands:
-        serial = TestRun.executor.run(command).stdout
-        if serial:
-            return serial.split('\n')[0]
-    return None
-
-
-def get_all_serial_numbers():
-    serial_numbers = {}
-    block_devices = get_block_devices_list()
-    for dev in block_devices:
-        serial = get_disk_serial_number(dev)
-        try:
-            path = resolve_to_by_id_link(dev)
-        except Exception:
-            continue
-        if serial:
-            serial_numbers[serial] = path
-        else:
-            TestRun.LOGGER.warning(f"Device {path} ({dev}) does not have a serial number.")
-            serial_numbers[path] = path
-    return serial_numbers
-
-
 def get_system_disks():
     system_device = TestRun.executor.run_expect_success('mount | grep " / "').stdout.split()[0]
     readlink_output = readlink(system_device)
@@ -176,7 +143,9 @@ def __get_slaves(device_name: str):
 
 def resolve_to_by_id_link(path):
     by_id_paths = TestRun.executor.run_expect_success("ls /dev/disk/by-id -1").stdout.splitlines()
-    dev_full_paths = [posixpath.join("/dev/disk/by-id", by_id_path) for by_id_path in by_id_paths]
+    dev_full_paths = [
+        posixpath.join("/dev/disk/by-id", by_id_path) for by_id_path in by_id_paths
+    ]
 
     for full_path in dev_full_paths:
         # handle exception for broken links