Merge pull request #34 from katlapinka/kasiat/tf-refactor

TF refactor

connection/base_executor.py

@@ -8,7 +8,7 @@
 from datetime import timedelta
 
 from core.test_run import TestRun
-from test_utils.output import CmdException
+from connection.utils.output import CmdException
 
 
 class BaseExecutor:

connection/local_executor.py

@@ -9,8 +9,8 @@
 
 from connection.base_executor import BaseExecutor
 from core.test_run import TestRun
-from test_tools.fs_utils import copy
-from test_utils.output import Output, CmdException
+from test_tools.fs_tools import copy
+from connection.utils.output import Output, CmdException
 
 
 class LocalExecutor(BaseExecutor):

connection/ssh_executor.py

@@ -3,17 +3,18 @@
 # Copyright(c) 2024 Huawei Technologies Co., Ltd.
 # SPDX-License-Identifier: BSD-3-Clause
 #
+
 import os
 import re
+import paramiko
 import socket
 import subprocess
-from datetime import timedelta, datetime
 
-import paramiko
+from datetime import timedelta, datetime
 
 from connection.base_executor import BaseExecutor
 from core.test_run import TestRun, Blocked
-from test_utils.output import Output
+from connection.utils.output import Output
 
 
 class SshExecutor(BaseExecutor):
@@ -46,7 +47,7 @@ def connect(self, user=None, port=None, timeout: timedelta = timedelta(seconds=3
             hostname = target["hostname"]
             key_filename = target.get("identityfile", None)
             user = target.get("user", user)
-            port = target.get("port", port)
+            port = int(target.get("port", port))
             if target.get("proxyjump", None) is not None:
                 proxy = config.lookup(target["proxyjump"])
                 jump = paramiko.SSHClient()

connection/utils/__init__.py

@@ -0,0 +1,4 @@
+#
+# Copyright(c) 2024 Huawei Technologies Co., Ltd.
+# SPDX-License-Identifier: BSD-3-Clause
+#

test_utils/asynchronous.py → connection/utils/asynchronous.py

@@ -1,9 +1,10 @@
 #
 # Copyright(c) 2020-2021 Intel Corporation
+# Copyright(c) 2024 Huawei Technologies Co., Ltd.
 # SPDX-License-Identifier: BSD-3-Clause
 #
 
-import concurrent
+from concurrent.futures import ThreadPoolExecutor
 
 
 def start_async_func(func, *args):
@@ -14,5 +15,5 @@ def start_async_func(func, *args):
     - done() method returns True when task ended (have a result or ended with an exception)
     otherwise returns False
     """
-    executor = concurrent.futures.ThreadPoolExecutor()
+    executor = ThreadPoolExecutor()
     return executor.submit(func, *args)

test_utils/output.py → connection/utils/output.py

@@ -1,5 +1,6 @@
 #
 # Copyright(c) 2019-2021 Intel Corporation
+# Copyright(c) 2024 Huawei Technologies Co., Ltd.
 # SPDX-License-Identifier: BSD-3-Clause
 #
 

test_utils/retry.py → connection/utils/retry.py

@@ -1,5 +1,6 @@
 #
 # Copyright(c) 2021 Intel Corporation
+# Copyright(c) 2024 Huawei Technologies Co., Ltd.
 # SPDX-License-Identifier: BSD-3-Clause
 #
 
@@ -33,7 +34,7 @@ def wrapped_func():
             try:
                 result = func()
                 return True
-            except:
+            except Exception:
                 return False
 
         cls.run_while_false(wrapped_func, retries=retries, timeout=timeout)

core/test_run_utils.py

@@ -19,7 +19,7 @@
 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

internal_plugins/scsi_debug/__init__.py

@@ -6,8 +6,8 @@
 
 from core.test_run_utils import TestRun
 from storage_devices.device import Device
-from test_utils import os_utils
-from test_utils.output import CmdException
+from connection.utils.output import CmdException
+from test_tools.os_tools import load_kernel_module, is_kernel_module_loaded, unload_kernel_module
 
 
 class ScsiDebug:
@@ -24,16 +24,16 @@ def post_setup(self):
     def reload(self):
         self.teardown()
         sleep(1)
-        load_output = os_utils.load_kernel_module(self.module_name, self.params)
+        load_output = load_kernel_module(self.module_name, self.params)
         if load_output.exit_code != 0:
             raise CmdException(f"Failed to load {self.module_name} module", load_output)
         TestRun.LOGGER.info(f"{self.module_name} loaded successfully.")
         sleep(10)
         TestRun.scsi_debug_devices = Device.get_scsi_debug_devices()
 
     def teardown(self):
-        if os_utils.is_kernel_module_loaded(self.module_name):
-            os_utils.unload_kernel_module(self.module_name)
+        if is_kernel_module_loaded(self.module_name):
+            unload_kernel_module(self.module_name)
 
 
 plugin_class = ScsiDebug

log/logger.py

@@ -15,8 +15,8 @@
 from log.html_log_config import HtmlLogConfig
 from log.html_log_manager import HtmlLogManager
 from log.html_presentation_policy import html_policy
-from test_utils.output import Output
-from test_utils.singleton import Singleton
+from connection.utils.output import Output
+from test_utils.common.singleton import Singleton
 
 
 def create_log(log_base_path, test_module, additional_args=None):
@@ -188,7 +188,7 @@ def step_info(self, step_name):
 
     def get_additional_logs(self):
         from core.test_run import TestRun
-        from test_tools.fs_utils import check_if_file_exists
+        from test_tools.fs_tools import check_if_file_exists
         messages_log = "/var/log/messages"
         if not check_if_file_exists(messages_log):
             messages_log = "/var/log/syslog"

scripts/__init__.py

@@ -0,0 +1,4 @@
+#
+# Copyright(c) 2024 Huawei Technologies Co., Ltd.
+# SPDX-License-Identifier: BSD-3-Clause
+#

storage_devices/device.py

@@ -3,29 +3,32 @@
 # Copyright(c) 2023-2024 Huawei Technologies Co., Ltd.
 # SPDX-License-Identifier: BSD-3-Clause
 #
+
 import posixpath
 
 from core.test_run import TestRun
-from test_tools import disk_utils, fs_utils
-from test_tools.disk_utils import get_device_filesystem_type, get_sysfs_path
+from test_tools import disk_tools
+from test_tools.disk_tools import get_sysfs_path, validate_dev_path, get_size
+from test_tools.fs_tools import (get_device_filesystem_type, Filesystem, wipefs,
+                                 readlink, write_file, mkfs, ls, parse_ls_output)
 from test_utils.io_stats import IoStats
-from test_utils.size import Size, Unit
+from type_def.size import Size, Unit
 
 
 class Device:
     def __init__(self, path):
-        disk_utils.validate_dev_path(path)
+        validate_dev_path(path)
         self.path = path
-        self.size = Size(disk_utils.get_size(self.get_device_id()), Unit.Byte)
+        self.size = Size(get_size(self.get_device_id()), Unit.Byte)
         self.filesystem = get_device_filesystem_type(self.get_device_id())
         self.mount_point = None
 
-    def create_filesystem(self, fs_type: disk_utils.Filesystem, force=True, blocksize=None):
-        disk_utils.create_filesystem(self, fs_type, force, blocksize)
+    def create_filesystem(self, fs_type: Filesystem, force=True, blocksize=None):
+        mkfs(self, fs_type, force, blocksize)
         self.filesystem = fs_type
 
     def wipe_filesystem(self, force=True):
-        disk_utils.wipe_filesystem(self, force)
+        wipefs(self, force)
         self.filesystem = None
 
     def is_mounted(self):
@@ -34,49 +37,48 @@ def is_mounted(self):
             return False
         else:
             mount_point_line = output.stdout.split('\n')[1]
-            device_path = fs_utils.readlink(self.path)
+            device_path = readlink(self.path)
             self.mount_point = mount_point_line[0:mount_point_line.find(device_path)].strip()
             return True
 
     def mount(self, mount_point, options: [str] = None):
         if not self.is_mounted():
-            if disk_utils.mount(self, mount_point, options):
+            if disk_tools.mount(self, mount_point, options):
                 self.mount_point = mount_point
         else:
             raise Exception(f"Device is already mounted! Actual mount point: {self.mount_point}")
 
     def unmount(self):
         if not self.is_mounted():
             TestRun.LOGGER.info("Device is not mounted.")
-        elif disk_utils.unmount(self):
+        elif disk_tools.unmount(self):
             self.mount_point = None
 
     def get_device_link(self, directory: str):
         items = self.get_all_device_links(directory)
         return next(i for i in items if i.full_path.startswith(directory))
 
     def get_device_id(self):
-        return fs_utils.readlink(self.path).split('/')[-1]
+        return readlink(self.path).split('/')[-1]
 
     def get_all_device_links(self, directory: str):
-        from test_tools import fs_utils
-        output = fs_utils.ls(f"$(find -L {directory} -samefile {self.path})")
-        return fs_utils.parse_ls_output(output, self.path)
+        output = ls(f"$(find -L {directory} -samefile {self.path})")
+        return parse_ls_output(output, self.path)
 
     def get_io_stats(self):
         return IoStats.get_io_stats(self.get_device_id())
 
     def get_sysfs_property(self, property_name):
-        path = posixpath.join(disk_utils.get_sysfs_path(self.get_device_id()),
+        path = posixpath.join(get_sysfs_path(self.get_device_id()),
                               "queue", property_name)
         return TestRun.executor.run_expect_success(f"cat {path}").stdout
 
     def set_sysfs_property(self, property_name, value):
         TestRun.LOGGER.info(
             f"Setting {property_name} for device {self.get_device_id()} to {value}.")
-        path = posixpath.join(disk_utils.get_sysfs_path(self.get_device_id()), "queue",
+        path = posixpath.join(get_sysfs_path(self.get_device_id()), "queue",
                             property_name)
-        fs_utils.write_file(path, str(value))
+        write_file(path, str(value))
 
     def set_max_io_size(self, new_max_io_size: Size):
         self.set_sysfs_property("max_sectors_kb",