feat: modify fault parameters manually

src/imgtests/exec/debugfs.py

@@ -0,0 +1,94 @@
+import logging
+from pathlib import Path
+from typing import TYPE_CHECKING, Final
+
+from imgtests.exec.exec import ExecResult, SSHClient, common_run_command
+
+if TYPE_CHECKING:
+    from imgtests.exec.exec import ExecResult, SSHClient
+
+DEBUGFS_PATH: Final = Path("/sys/kernel/debug")
+MAX_FAULT_PROBABILITY: Final = 100
+
+logger = logging.getLogger(__name__)
+
+
+def ensure_debugfs(ssh_client: SSHClient | None) -> ExecResult:
+    """Ensures that debugfs is created and mounted."""
+    result = common_run_command(("sudo", "mkdir", "-p", str(DEBUGFS_PATH)), ssh_client)
+    if result.returncode:
+        return result
+    mount_pattern = f"[[:space:]]{DEBUGFS_PATH}[[:space:]]debugfs[[:space:]]"
+    result = common_run_command(
+        ("sudo", "grep", "-qs", mount_pattern, "/proc/mounts"),
+        ssh_client,
+    )
+    if result.returncode == 0 or result.returncode != 1:
+        return result
+    logger.info("Mounting debugfs to '%s'.", str(DEBUGFS_PATH))
+    result = common_run_command(
+        ("sudo", "mount", "-t", "debugfs", "debugfs", str(DEBUGFS_PATH)),
+        ssh_client,
+    )
+
+    if result.returncode:
+        logger.info("Unmounting debugfs from '%s'.", str(DEBUGFS_PATH))
+        common_run_command(("sudo", "umount", str(DEBUGFS_PATH)), ssh_client)
+    return result
+
+
+def validate_fault_probability(fault_prob: int) -> None:
+    """Checks if fault injection probability is in between borders."""
+    if not 0 <= fault_prob <= MAX_FAULT_PROBABILITY:
+        err_msg = f"fault_probability must be in range 0..{MAX_FAULT_PROBABILITY}."
+        raise ValueError(err_msg)
+
+
+def change_fault_parameters(
+    client: SSHClient | None,
+    fault_probability: int,
+    fault_interval: int,
+) -> ExecResult:
+    result = ensure_debugfs(client)
+    if result.returncode:
+        return result
+
+    result = common_run_command(["ls", str(DEBUGFS_PATH)], ssh_client=client)
+    if result.returncode:
+        logger.error("Failed to list debugfs directory.")
+        return result
+
+    dirs = [
+        dir_name
+        for dir_name in result.stdout.splitlines()
+        if ("fail" in dir_name or "fault" in dir_name) and dir_name != "fault_around_bytes"
+    ]
+    if not dirs:
+        logger.warning("No fault-injection debugfs entries found under %s", str(DEBUGFS_PATH))
+        return result
+
+    # When fault injection is enabled, times should be set to -1, so that it would run infinitely
+    # When fault injection is disabled, times set to default 1
+    times = -1 if fault_probability > 0 else 1
+    last_result = result
+    for directory in dirs:
+        dir_path = DEBUGFS_PATH / directory
+        updates = (
+            ("interval", fault_interval),
+            ("space", 0),
+            ("times", times),
+            ("probability", fault_probability),
+        )
+
+        for parameter, value in updates:
+            write_cmd = ["echo", f"{value}", ">", f"{dir_path}/{parameter}"]
+            result = common_run_command(write_cmd, ssh_client=client)
+            last_result = result
+            if result.returncode:
+                logger.error(
+                    "Failed to update fault-injection parameter %s in %s",
+                    parameter,
+                    dir_path,
+                )
+                return result
+    return last_result

src/imgtests/exec/loaders/kirk.py

@@ -6,6 +6,7 @@
 from typing import TYPE_CHECKING, Any
 
 from imgtests.exec.base_util import GenericUtil
+from imgtests.exec.debugfs import ensure_debugfs, validate_fault_probability
 from imgtests.exec.exec import ExecResult, SSHClient, common_run_command
 from imgtests.exec.osinfo import get_os_release
 from imgtests.exec.pkgmgrs.zypper import Zypper
@@ -19,8 +20,6 @@
 logger = logging.getLogger(__name__)
 
 DEFAULT_LTP_RESULTS_DIR = Path("/var/tmp/ltp-results")  # noqa: S108
-DEBUGFS_MOUNTPOINT = Path("/sys/kernel/debug")
-MAX_FAULT_PROBABILITY = 100
 
 
 class Kirk(GenericUtil):
@@ -90,36 +89,9 @@ def list_suites(
 
         return tuple(line.strip() for line in res.stdout.splitlines() if line.strip())
 
-    @staticmethod
-    def _validate_fault_probability(fault_prob: int) -> None:
-        """Checks if fault injection probability is in between borders."""
-        if not 0 <= fault_prob <= MAX_FAULT_PROBABILITY:
-            err_msg = f"fault_probability must be in range 0..{MAX_FAULT_PROBABILITY}."
-            raise ValueError(err_msg)
-
-    def _ensure_debugfs(self) -> ExecResult:
-        """Ensures that debugfs is created and mounted."""
-        debugfs_path = str(DEBUGFS_MOUNTPOINT)
-        result = common_run_command(("sudo", "mkdir", "-p", debugfs_path), self.ssh_client)
-        if result.returncode:
-            return result
-        mount_pattern = f"[[:space:]]{debugfs_path}[[:space:]]debugfs[[:space:]]"
-        result = common_run_command(
-            ("sudo", "grep", "-qs", mount_pattern, "/proc/mounts"),
-            self.ssh_client,
-        )
-        if result.returncode == 0 or result.returncode != 1:
-            return result
-        logger.info("Mounting debugfs to '%s'.", debugfs_path)
-        result = common_run_command(
-            ("sudo", "mount", "-t", "debugfs", "debugfs", debugfs_path),
-            self.ssh_client,
-        )
-
-        if result.returncode:
-            logger.info("Unmounting debugfs from '%s'.", debugfs_path)
-            common_run_command(("sudo", "umount", debugfs_path), self.ssh_client)
-        return result
+    def is_suites_available(self, required_suites: tuple[str, ...]) -> bool:
+        available_suites = self.list_suites()
+        return all(suite in available_suites for suite in required_suites)
 
     def run(  # noqa: PLR0911, PLR0913
         self,
@@ -152,9 +124,9 @@ def run(  # noqa: PLR0911, PLR0913
             raise ValueError(scenarios_empty_msg)
 
         if fault_prob is not None:
-            self._validate_fault_probability(fault_prob)
+            validate_fault_probability(fault_prob)
 
-            debugfs_res = self._ensure_debugfs()
+            debugfs_res = ensure_debugfs(self.ssh_client)
             if debugfs_res.returncode:
                 return debugfs_res, None
 

src/imgtests/runner.py

@@ -172,20 +172,27 @@ def handle_tests(
                 test_instance = test_class
             else:
                 test_instance = test_class(timeout=self.__test_config.test_duration)
-            for result in test_instance(self._executor, self._client):
-                self._database.insert_util_run_result(
-                    experiment_id=experiment_id,
-                    # TODO: fill util_type with the correct value
-                    util_type="loader",
-                    # TODO: fill descriptions and adds into TestResult class
-                    description="",
-                    result=result.metrics,
-                    command=result.command,
-                    started_at=result.started_at,
-                    ended_at=result.ended_at,
+            try:
+                for result in test_instance(self._executor, self._client):
+                    self._database.insert_util_run_result(
+                        experiment_id=experiment_id,
+                        # TODO: fill util_type with the correct value
+                        util_type="loader",
+                        # TODO: fill descriptions and adds into TestResult class
+                        description="",
+                        result=result.metrics,
+                        command=result.command,
+                        started_at=result.started_at,
+                        ended_at=result.ended_at,
+                    )
+                    self.__tests_statuses[result.status] += 1
+                    self.__tests_cnt += 1
+            except Exception:
+                self._logger.exception(
+                    "Test '%s' failed with exception.",
+                    test_instance.description,
                 )
-                self.__tests_statuses[result.status] += 1
-                self.__tests_cnt += 1
+                self.__tests_statuses[TestStatus.BROKEN] += 1
             self._collect_system_errors(
                 experiment_id=experiment_id,
                 since=test_started_at,