refactor(utils): delegate subprocess and filesystem calls to shared backends

Route fs_utils and sys_utils subprocess/filesystem operations through

core.backends so tests can substitute a fake runner or filesystem from one place.

Make create_python_venv, update_python_pip and update_system_package_lists

raise RuntimeError on failure so callers decide whether to fail fast or recover.

Catch apt update failures in UpdateMenu to keep the menu usable.

kiauh/core/menus/update_menu.py

@@ -254,7 +254,16 @@ class UpdateMenu(BaseMenu):
         self._fetch_system_package_update_status()
 
     def _fetch_system_package_update_status(self) -> None:
-        update_system_package_lists(silent=True)
+        # Treat apt update failures as non-fatal here so the menu remains usable
+        # even when package metadata is unavailable. Dependency installation still
+        # fails fast elsewhere.
+        try:
+            update_system_package_lists(silent=True)
+        except RuntimeError as exc:
+            Logger.print_warn(
+                "Could not update the system package lists; "
+                f"system package status may be incomplete. ({exc})"
+            )
         self.packages = get_upgradable_packages()
         self.package_count = len(self.packages)
 

kiauh/utils/common.py

@@ -89,6 +89,8 @@ def check_install_dependencies(
         Logger.print_info("The following packages need installation:")
         for r in requirements:
             print(Color.apply(f"● {r}", Color.CYAN))
+        # Installing against stale or missing package metadata is unsafe, so abort
+        # here instead of swallowing the error like the update menu does.
         update_system_package_lists(silent=False)
         install_system_packages(requirements)
 

kiauh/utils/fs_utils.py

@@ -12,15 +12,34 @@ from __future__ import annotations
 
 import os
 import re
-import shutil
+import subprocess
 from pathlib import Path
-from subprocess import DEVNULL, PIPE, CalledProcessError, call, check_output, run
+from subprocess import DEVNULL, PIPE, CalledProcessError
 from typing import List
 from zipfile import ZipFile
 
+from core import backends
 from core.decorators import deprecated
 from core.logger import Logger
 
+# Delegate to the shared backends module so tests can substitute
+# command_runner/filesystem from one location instead of patching module globals.
+
+
+def run(cmd: str | List[str], **kwargs) -> subprocess.CompletedProcess[str]:
+    """Run a command through the shared command runner."""
+    return backends.command_runner.run(cmd, **kwargs)
+
+
+def check_output(cmd: str | List[str], **kwargs) -> str | bytes:
+    """Run a command and return its output through the shared command runner."""
+    return backends.command_runner.check_output(cmd, **kwargs)
+
+
+def call(cmd: str | List[str], **kwargs) -> int:
+    """Run a command and return its exit code through the shared command runner."""
+    return backends.command_runner.call(cmd, **kwargs)
+
 
 def check_file_exist(file_path: Path, sudo=False) -> bool:
     """
@@ -103,14 +122,14 @@ def remove_file(file_path: Path, sudo=False) -> None:
 
 def run_remove_routines(file: Path) -> bool:
     try:
-        if not file.is_symlink() and not file.exists():
+        if not backends.filesystem.is_symlink(file) and not backends.filesystem.exists(file):
             Logger.print_info(f"File '{file}' does not exist. Skipped ...")
             return False
 
-        if file.is_dir():
-            shutil.rmtree(file)
-        elif file.is_file() or file.is_symlink():
-            file.unlink()
+        if backends.filesystem.is_dir(file):
+            backends.filesystem.rmtree(file)
+        elif backends.filesystem.is_file(file) or backends.filesystem.is_symlink(file):
+            backends.filesystem.unlink(file)
         else:
             Logger.print_error(f"File '{file}' is neither a file nor a directory!")
             return False
@@ -127,6 +146,9 @@ def run_remove_routines(file: Path) -> bool:
             Logger.print_error(f"Error deleting '{file}' with sudo:\n{e}")
             Logger.print_error("Remove this directory manually!")
             return False
+    # Direct and sudo removal both failed without raising; return a boolean so
+    # callers get a predictable result.
+    return False
 
 
 def unzip(filepath: Path, target_dir: Path) -> None:
@@ -143,9 +165,9 @@ def unzip(filepath: Path, target_dir: Path) -> None:
 def create_folders(dirs: List[Path]) -> None:
     try:
         for _dir in dirs:
-            if _dir.exists():
+            if backends.filesystem.exists(_dir):
                 continue
-            _dir.mkdir(exist_ok=True)
+            backends.filesystem.mkdir(_dir, exist_ok=True)
             Logger.print_ok(f"Created directory '{_dir}'!")
     except OSError as e:
         Logger.print_error(f"Error creating directories: {e}")

kiauh/utils/instance_type.py

@@ -12,10 +12,10 @@ from typing import TypeVar
 from components.klipper.klipper import Klipper
 from components.moonraker.moonraker import Moonraker
 from extensions.obico.moonraker_obico import MoonrakerObico
-from extensions.octoeverywhere.octoeverywhere import Octoeverywhere
 from extensions.octoapp.octoapp import Octoapp
-from extensions.telegram_bot.moonraker_telegram_bot import MoonrakerTelegramBot
+from extensions.octoeverywhere.octoeverywhere import Octoeverywhere
 from extensions.octoprint.octoprint import Octoprint
+from extensions.telegram_bot.moonraker_telegram_bot import MoonrakerTelegramBot
 
 InstanceType = TypeVar(
     "InstanceType",

kiauh/utils/sys_utils.py

@@ -13,14 +13,16 @@ import re
 import select
 import shutil
 import socket
+import subprocess
 import sys
 import time
 import urllib.error
 import urllib.request
 from pathlib import Path
-from subprocess import DEVNULL, PIPE, CalledProcessError, Popen, check_output, run
+from subprocess import DEVNULL, PIPE, CalledProcessError, Popen
 from typing import List, Literal, Set, Tuple
 
+from core import backends
 from core.constants import SYSTEMD
 from core.logger import Logger
 from utils.fs_utils import check_file_exist, remove_with_sudo
@@ -38,6 +40,29 @@ SysCtlServiceAction = Literal[
 ]
 SysCtlManageAction = Literal["daemon-reload", "reset-failed"]
 
+# Delegate to the shared backends module so tests can substitute command_runner
+# from one location instead of patching module globals.
+
+
+def run(cmd: str | List[str], **kwargs) -> subprocess.CompletedProcess[str]:
+    """Run a command through the shared command runner."""
+    return backends.command_runner.run(cmd, **kwargs)
+
+
+def check_output(cmd: str | List[str], **kwargs) -> str | bytes:
+    """Run a command and return its output through the shared command runner."""
+    return backends.command_runner.check_output(cmd, **kwargs)
+
+
+def call(cmd: str | List[str], **kwargs) -> int:
+    """Run a command and return its exit code through the shared command runner."""
+    return backends.command_runner.call(cmd, **kwargs)
+
+
+def popen(cmd: str | List[str], **kwargs) -> Popen:
+    """Start a process through the shared command runner."""
+    return backends.command_runner.popen(cmd, **kwargs)
+
 
 class VenvCreationFailedException(Exception):
     pass
@@ -95,7 +120,8 @@ def create_python_venv(
     target: Path,
     force: bool = False,
     allow_access_to_system_site_packages: bool = False,
-    use_python_binary: str | None = None
+    use_python_binary: str | None = None,
+    interactive: bool = True,
 ) -> bool:
     """
     Create a python 3 virtualenv at the provided target destination.
@@ -105,6 +131,9 @@ def create_python_venv(
     :param force: Force recreation of the virtualenv
     :param allow_access_to_system_site_packages: give the virtual environment access to the system site-packages dir
     :param use_python_binary: allows to override default python binary
+    :param interactive: When False (headless), an existing venv is left untouched
+        instead of prompting for confirmation: a non-interactive run must never
+        destroy a working venv, and must never block on ``read``.
     :return: bool
     """
     Logger.print_status("Set up Python virtual environment ...")
@@ -116,7 +145,7 @@ def create_python_venv(
     ) if allow_access_to_system_site_packages else None
 
     n = 2
-    while(n > 0):
+    while n > 0:
         if not target.exists():
             try:
                 run(cmd, check=True)
@@ -131,11 +160,20 @@ def create_python_venv(
                 # but the function should still behave correctly
                 Logger.print_error("Virtualenv still exists after deletion.")
                 return False
-            if not force and not get_confirm(
-                "Virtualenv already exists. Re-create?", default_choice=False
-            ):
-                Logger.print_info("Skipping re-creation of virtualenv ...")
-                return False
+            if not force:
+                if not interactive:
+                    # Headless mode must never destroy an existing venv or block
+                    # on input; skip it so requirements are only installed into
+                    # freshly created environments.
+                    Logger.print_info(
+                        "Virtualenv already exists; skipping re-creation ..."
+                    )
+                    return False
+                if not get_confirm(
+                    "Virtualenv already exists. Re-create?", default_choice=False
+                ):
+                    Logger.print_info("Skipping re-creation of virtualenv ...")
+                    return False
 
             try:
                 shutil.rmtree(target)
@@ -165,7 +203,7 @@ def update_python_pip(target: Path) -> None:
         if result.returncode != 0 or result.stderr:
             Logger.print_error(f"{result.stderr}", False)
             Logger.print_error("Updating pip failed!")
-            return
+            raise RuntimeError("Updating pip failed!")
 
         Logger.print_ok("Updating pip successful!")
     except FileNotFoundError as e:
@@ -268,7 +306,7 @@ def update_system_package_lists(silent: bool, rls_info_change=False) -> None:
         if result.returncode != 0 or result.stderr:
             Logger.print_error(f"{result.stderr}", False)
             Logger.print_error("Updating system package list failed!")
-            return
+            raise RuntimeError("Updating system package list failed!")
 
         Logger.print_ok("System package list update successful!")
     except CalledProcessError as e:

kiauh/utils/tests/conftest.py

@@ -1,21 +0,0 @@
-import sys
-from pathlib import Path
-
-import pytest
-
-PROJECT_ROOT = Path(__file__).resolve().parents[3]
-if str(PROJECT_ROOT) not in sys.path:
-    sys.path.insert(0, str(PROJECT_ROOT))
-
-
-@pytest.fixture(autouse=True)
-def silence_logger(monkeypatch: pytest.MonkeyPatch) -> None:
-    for name in (
-        "print_info",
-        "print_ok",
-        "print_warn",
-        "print_error",
-        "print_status",
-        "print_dialog",
-    ):
-        monkeypatch.setattr(f"core.logger.Logger.{name}", lambda *a, **k: None)

kiauh/utils/tests/test_common.py

@@ -104,6 +104,28 @@ class TestCheckInstallDependencies:
 
         check_install_dependencies({"pkg"})
 
+    def test_propagates_runtime_error_from_package_list_update(
+        self, monkeypatch
+    ) -> None:
+        # Installing dependencies must propagate apt update failures rather than
+        # swallow them, because continuing with broken package metadata is unsafe.
+        monkeypatch.setattr(
+            "utils.common.check_package_install",
+            lambda *_a, **_k: ["missing-pkg"],
+        )
+
+        def _raise(*_a, **_k):
+            raise RuntimeError("apt-get update failed")
+
+        monkeypatch.setattr("utils.common.update_system_package_lists", _raise)
+        monkeypatch.setattr(
+            "utils.common.install_system_packages",
+            lambda *_a, **_k: pytest.fail("should not install on broken apt update"),
+        )
+
+        with pytest.raises(RuntimeError):
+            check_install_dependencies({"pkg"})
+
 
 class _FakeInstanceType:
     def __init__(self, suffix: str):

kiauh/utils/tests/test_sys_utils.py

@@ -158,6 +158,25 @@ class TestCreatePythonVenv:
         assert create_python_venv(target, force=True) is True
         assert removed == [target]
 
+    def test_headless_skips_recreate_without_prompting(self, monkeypatch) -> None:
+        target = Path("/tmp/venv")
+
+        monkeypatch.setattr("utils.sys_utils.Path.exists", lambda self: self == target)
+        monkeypatch.setattr(
+            "utils.sys_utils.get_confirm",
+            lambda *a, **k: pytest.fail("should not prompt in headless mode"),
+        )
+        monkeypatch.setattr(
+            "utils.sys_utils.shutil.rmtree",
+            lambda *a, **k: pytest.fail("should not rmtree in headless mode"),
+        )
+        monkeypatch.setattr(
+            "utils.sys_utils.run",
+            lambda *a, **k: pytest.fail("should not recreate in headless mode"),
+        )
+
+        assert create_python_venv(target, interactive=False) is False
+
     def test_creation_failure(self, monkeypatch) -> None:
         monkeypatch.setattr(
             "utils.sys_utils.run",
@@ -200,15 +219,25 @@ class TestUpdatePythonPip:
         update_python_pip(Path("/tmp/venv"))
         assert runs == [["/tmp/venv/bin/pip", "install", "-U", "pip"]]
 
-    def test_logs_stderr(self, monkeypatch, capsys) -> None:
+    def test_succeeds_when_returncode_and_stderr_are_clean(self, monkeypatch) -> None:
         def fake_run(cmd: List[str], **kwargs: Any) -> Any:
-            return type("R", (), {"returncode": 0, "stderr": "some warning"})()
+            return type("R", (), {"returncode": 0, "stderr": ""})()
 
         monkeypatch.setattr("utils.sys_utils.check_file_exist", lambda *a, **k: True)
         monkeypatch.setattr("utils.sys_utils.run", fake_run)
 
         update_python_pip(Path("/tmp/venv"))
 
+    def test_failure_raises(self, monkeypatch) -> None:
+        def fake_run(cmd: List[str], **kwargs: Any) -> Any:
+            return type("R", (), {"returncode": 1, "stderr": "nope"})()
+
+        monkeypatch.setattr("utils.sys_utils.check_file_exist", lambda *a, **k: True)
+        monkeypatch.setattr("utils.sys_utils.run", fake_run)
+
+        with pytest.raises(RuntimeError, match="Updating pip failed"):
+            update_python_pip(Path("/tmp/venv"))
+
 
 class TestInstallPythonRequirements:
     def test_success(self, monkeypatch) -> None:
@@ -281,6 +310,18 @@ class TestUpdateSystemPackageLists:
 
         assert runs == [["sudo", "apt-get", "update", "--allow-releaseinfo-change"]]
 
+    def test_failure_raises(self, monkeypatch) -> None:
+        monkeypatch.setattr("utils.sys_utils.time.time", lambda: 100_000)
+        monkeypatch.setattr("utils.sys_utils.os.path.getmtime", lambda p: 0)
+
+        def fake_run(cmd: List[str], **kwargs: Any) -> Any:
+            return type("R", (), {"returncode": 1, "stderr": "apt failed"})()
+
+        monkeypatch.setattr("utils.sys_utils.run", fake_run)
+
+        with pytest.raises(RuntimeError, match="Updating system package list failed"):
+            update_system_package_lists(silent=True)
+
 
 class TestGetUpgradablePackages:
     def test_parses_apt_list(self, monkeypatch) -> None:
@@ -413,9 +454,11 @@ class TestSetNginxPermissions:
     def test_no_change_when_executable(self, monkeypatch) -> None:
         monkeypatch.setattr(
             "utils.sys_utils.run",
-            lambda cmd, **kwargs: type("R", (), {"stdout": "drwxr-xr-x"})()
-            if "ls" in cmd
-            else pytest.fail("should not chmod"),
+            lambda cmd, **kwargs: (
+                type("R", (), {"stdout": "drwxr-xr-x"})()
+                if "ls" in cmd
+                else pytest.fail("should not chmod")
+            ),
         )
         set_nginx_permissions()