From e4eb36d72fced7c654d8070a9e85377281b98739 Mon Sep 17 00:00:00 2001 From: dw-0 Date: Sat, 11 Jul 2026 00:27:35 +0200 Subject: [PATCH] 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 | 11 +++++- kiauh/utils/common.py | 2 + kiauh/utils/fs_utils.py | 40 +++++++++++++++----- kiauh/utils/instance_type.py | 4 +- kiauh/utils/sys_utils.py | 58 ++++++++++++++++++++++++----- kiauh/utils/tests/conftest.py | 21 ----------- kiauh/utils/tests/test_common.py | 22 +++++++++++ kiauh/utils/tests/test_sys_utils.py | 53 +++++++++++++++++++++++--- 8 files changed, 163 insertions(+), 48 deletions(-) delete mode 100644 kiauh/utils/tests/conftest.py diff --git a/kiauh/core/menus/update_menu.py b/kiauh/core/menus/update_menu.py index 3bb76617..a2d4536a 100644 --- a/kiauh/core/menus/update_menu.py +++ b/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) diff --git a/kiauh/utils/common.py b/kiauh/utils/common.py index da6ebb74..27bf22c8 100644 --- a/kiauh/utils/common.py +++ b/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) diff --git a/kiauh/utils/fs_utils.py b/kiauh/utils/fs_utils.py index 0d141715..fa442e81 100644 --- a/kiauh/utils/fs_utils.py +++ b/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}") diff --git a/kiauh/utils/instance_type.py b/kiauh/utils/instance_type.py index 3ee79104..54f54bf5 100644 --- a/kiauh/utils/instance_type.py +++ b/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", diff --git a/kiauh/utils/sys_utils.py b/kiauh/utils/sys_utils.py index 8f4a25fb..942de501 100644 --- a/kiauh/utils/sys_utils.py +++ b/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: diff --git a/kiauh/utils/tests/conftest.py b/kiauh/utils/tests/conftest.py deleted file mode 100644 index ed492803..00000000 --- a/kiauh/utils/tests/conftest.py +++ /dev/null @@ -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) diff --git a/kiauh/utils/tests/test_common.py b/kiauh/utils/tests/test_common.py index ae7ada04..46478b8e 100644 --- a/kiauh/utils/tests/test_common.py +++ b/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): diff --git a/kiauh/utils/tests/test_sys_utils.py b/kiauh/utils/tests/test_sys_utils.py index 76f305f7..8ab48086 100644 --- a/kiauh/utils/tests/test_sys_utils.py +++ b/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()