scsi_debug fixes

storage_devices/device.py

@@ -1,6 +1,7 @@
 #
 # Copyright(c) 2019-2022 Intel Corporation
 # Copyright(c) 2023-2024 Huawei Technologies Co., Ltd.
+# Copyright(c) 2026 Unvertical
 # SPDX-License-Identifier: BSD-3-Clause
 #
 
@@ -116,10 +117,3 @@ def __str__(self):
 
     def __repr__(self):
         return str(self)
-
-    @staticmethod
-    def get_scsi_debug_devices():
-        scsi_debug_devices = TestRun.executor.run_expect_success(
-            "lsscsi --scsi_id | grep scsi_debug").stdout
-        return [Device(f'/dev/disk/by-id/scsi-{device.split()[-1]}')
-                for device in scsi_debug_devices.splitlines()]

test_tools/os_tools.py

@@ -49,6 +49,14 @@ class SystemManagerType(Enum):
     systemd = 1
 
 
+def get_syslog_path():
+    for path in ["/var/log/messages", "/var/log/syslog"]:
+        result = TestRun.executor.run(f"test -f {path}")
+        if result.exit_code == 0:
+            return path
+    raise FileNotFoundError("Could not find syslog file")
+
+
 def get_distro():
     output = TestRun.executor.run(
         "cat /etc/os-release | grep -e \"^ID=\" | awk -F= '{print$2}' | tr -d '\"'"

test_tools/scsi_debug.py

@@ -1,77 +1,94 @@
 #
-# Copyright(c) 2022 Intel Corporation
+# Copyright(c) 2020-2022 Intel Corporation
+# Copyright(c) 2026 Unvertical
 # SPDX-License-Identifier: BSD-3-Clause
 #
 
 import re
+from time import sleep
 
+from connection.utils.output import CmdException
 from core.test_run import TestRun
+from storage_devices.device import Device
+from test_tools.os_tools import (
+    get_syslog_path,
+    is_kernel_module_loaded,
+    load_kernel_module,
+    unload_kernel_module,
+)
 
-syslog_path = "/var/log/messages"
+MODULE_NAME = "scsi_debug"
 
+FLUSH = re.compile(r"scsi_debug:[\s\S]*cmd 35")
+FUA = re.compile(r"scsi_debug:[\s\S]*cmd 2a 08")
 
-class Logs:
-    last_read_line = 1
-    FLUSH = re.compile(r"scsi_debug:[\s\S]*cmd 35")
-    FUA = re.compile(r"scsi_debug:[\s\S]*cmd 2a 08")
 
-    @staticmethod
-    def check_syslog_for_signals():
-        Logs.check_syslog_for_flush()
-        Logs.check_syslog_for_fua()
+class ScsiDebug:
+    def __init__(self, params):
+        self.params = params
+        self.syslog_path = None
+        self.last_read_line = 0
+        self.reload()
 
-    @staticmethod
-    def check_syslog_for_flush():
-        """Check syslog for FLUSH logs"""
-        log_lines = Logs._read_syslog(Logs.last_read_line)
-        flush_logs_counter = Logs._count_logs(log_lines, Logs.FLUSH)
-        log_type = "FLUSH"
-        Logs._validate_logs_amount(flush_logs_counter, log_type)
+    def reload(self):
+        self.unload()
+        sleep(1)
+        load_output = load_kernel_module(MODULE_NAME, self.params)
+        if load_output.exit_code != 0:
+            raise CmdException(f"Failed to load {MODULE_NAME} module", load_output)
+        TestRun.LOGGER.info(f"{MODULE_NAME} loaded successfully.")
+        sleep(10)
 
     @staticmethod
-    def check_syslog_for_fua():
-        """Check syslog for FUA logs"""
-        log_lines = Logs._read_syslog(Logs.last_read_line)
-        fua_logs_counter = Logs._count_logs(log_lines, Logs.FUA)
-        log_type = "FUA"
-        Logs._validate_logs_amount(fua_logs_counter, log_type)
+    def unload():
+        if is_kernel_module_loaded(MODULE_NAME):
+            unload_kernel_module(MODULE_NAME)
 
-    @staticmethod
-    def _read_syslog(last_read_line: int):
-        """Read recent lines in syslog, mark last line and return read lines as list."""
-        log_lines = TestRun.executor.run_expect_success(
-            f"tail -qn +{last_read_line} {syslog_path}"
-        ).stdout.splitlines()
-        # mark last read line to continue next reading from here
-        Logs.last_read_line += len(log_lines)
+    def get_devices(self):
+        scsi_debug_devices = TestRun.executor.run_expect_success(
+            "lsscsi --scsi_id | grep scsi_debug").stdout
+        return [Device(f'/dev/disk/by-id/scsi-{device.split()[-1]}')
+                for device in scsi_debug_devices.splitlines()]
 
-        return log_lines
+    def reset_stats(self):
+        """Set syslog position to current end so subsequent reads only see new entries."""
+        if self.syslog_path is None:
+            self.syslog_path = get_syslog_path()
 
-    @staticmethod
-    def _count_logs(log_lines: list, expected_log):
-        """Count specified log in list and return its amount."""
-        logs_counter = 0
+        line_count = TestRun.executor.run_expect_success(
+            f"wc -l < {self.syslog_path}"
+        ).stdout.strip()
+        self.last_read_line = int(line_count)
 
-        for line in log_lines:
-            is_log_in_line = expected_log.search(line)
-            if is_log_in_line is not None:
-                logs_counter += 1
+    def get_flush_count(self):
+        log_lines = self._read_syslog()
+        flush_count, _ = self._count_logs(log_lines)
+        return flush_count
 
-        return logs_counter
+    def get_fua_count(self):
+        log_lines = self._read_syslog()
+        _, fua_count = self._count_logs(log_lines)
+        return fua_count
 
-    @staticmethod
-    def _validate_logs_amount(logs_counter: int, log_type: str):
-        """Validate amount of logs and return"""
-        if logs_counter == 0:
-            if Logs._is_flush(log_type):
-                TestRun.LOGGER.error(f"{log_type} log not occured")
-            else:
-                TestRun.LOGGER.warning(f"{log_type} log not occured")
-        elif logs_counter == 1:
-            TestRun.LOGGER.warning(f"{log_type} log occured only once.")
-        else:
-            TestRun.LOGGER.info(f"{log_type} log occured {logs_counter} times.")
+    def _read_syslog(self):
+        """Read syslog lines since last mark and advance the position."""
+        if self.syslog_path is None:
+            self.syslog_path = get_syslog_path()
+
+        start_line = self.last_read_line + 1
+        log_lines = TestRun.executor.run_expect_success(
+            f"tail -qn +{start_line} {self.syslog_path}"
+        ).stdout.splitlines()
+        self.last_read_line += len(log_lines)
+        return log_lines
 
     @staticmethod
-    def _is_flush(log_type: str):
-        return log_type == "FLUSH"
+    def _count_logs(log_lines):
+        flush_count = 0
+        fua_count = 0
+        for line in log_lines:
+            if FLUSH.search(line):
+                flush_count += 1
+            if FUA.search(line):
+                fua_count += 1
+        return flush_count, fua_count