diff --git a/kiauh/core/backends/__init__.py b/kiauh/core/backends/__init__.py new file mode 100644 index 00000000..db023100 --- /dev/null +++ b/kiauh/core/backends/__init__.py @@ -0,0 +1,182 @@ +# ======================================================================= # +# Copyright (C) 2020 - 2026 Dominik Willner # +# # +# This file is part of KIAUH - Klipper Installation And Update Helper # +# # +# https://github.com/dw-0/kiauh # +# # +# This file may be distributed under the terms of the GNU GPLv3 license # +# ======================================================================= # +from __future__ import annotations + +import shutil +import subprocess +from pathlib import Path +from typing import Any, List, Protocol, Sequence, cast, runtime_checkable + +# --------------------------------------------------------------------------- # +# Singleton backends # +# --------------------------------------------------------------------------- # +# There is exactly ONE ``command_runner`` and ONE ``filesystem`` global in the +# whole project, owned by this module. They are assigned (with explicit type +# annotations) at the bottom of this file, AFTER the default implementations +# are defined. ``utils.fs_utils`` and ``utils.sys_utils`` delegate to these +# singletons via the wrapper functions below, so tests patch a single place — +# ``core.backends.command_runner`` / ``core.backends.filesystem`` — instead of +# per-module duplicates + + +def run(cmd: str | List[str], **kwargs: Any) -> subprocess.CompletedProcess[str]: + """Run a command through the shared command runner.""" + return command_runner.run(cmd, **kwargs) + + +def check_output(cmd: str | List[str], **kwargs: Any) -> str | bytes: + """Run a command and return its output through the shared command runner.""" + return command_runner.check_output(cmd, **kwargs) + + +def call(cmd: str | List[str], **kwargs: Any) -> int: + """Run a command and return its exit code through the shared command runner.""" + return command_runner.call(cmd, **kwargs) + + +def popen(cmd: str | List[str], **kwargs: Any) -> subprocess.Popen: + """Start a process through the shared command runner.""" + return command_runner.popen(cmd, **kwargs) + + +@runtime_checkable +class CommandRunner(Protocol): + """Pluggable backend for executing system commands.""" + + def run( + self, + cmd: str | Sequence[str], + **kwargs: Any, + ) -> subprocess.CompletedProcess: ... + + def check_output( + self, + cmd: str | Sequence[str], + **kwargs: Any, + ) -> str | bytes: ... + + def call( + self, + cmd: str | Sequence[str], + **kwargs: Any, + ) -> int: ... + + def popen( + self, + cmd: str | Sequence[str], + **kwargs: Any, + ) -> subprocess.Popen: ... + + +class SubprocessRunner: + """Default command runner backed by the standard subprocess module.""" + + def run( + self, + cmd: str | Sequence[str], + **kwargs: Any, + ) -> subprocess.CompletedProcess[str]: + return subprocess.run(cmd, **kwargs) + + def check_output( + self, + cmd: str | Sequence[str], + **kwargs: Any, + ) -> str | bytes: + return cast("str | bytes", subprocess.check_output(cmd, **kwargs)) + + def call( + self, + cmd: str | Sequence[str], + **kwargs: Any, + ) -> int: + return subprocess.call(cmd, **kwargs) + + def popen( + self, + cmd: str | Sequence[str], + **kwargs: Any, + ) -> subprocess.Popen: + return subprocess.Popen(cmd, **kwargs) + + +@runtime_checkable +class FilesystemBackend(Protocol): + """Pluggable backend for filesystem operations.""" + + def exists(self, path: Path) -> bool: ... + + def is_dir(self, path: Path) -> bool: ... + + def is_file(self, path: Path) -> bool: ... + + def is_symlink(self, path: Path) -> bool: ... + + def mkdir( + self, path: Path, *, parents: bool = False, exist_ok: bool = False + ) -> None: ... + + def unlink(self, path: Path) -> None: ... + + def rmtree(self, path: Path) -> None: ... + + def read_text(self, path: Path) -> str: ... + + def write_text(self, path: Path, content: str) -> None: ... + + def copy(self, source: Path, target: Path) -> None: ... + + def home(self) -> Path: ... + + +class LocalFilesystemBackend: + """Default filesystem backend backed by the local filesystem.""" + + def exists(self, path: Path) -> bool: + return path.exists() + + def is_dir(self, path: Path) -> bool: + return path.is_dir() + + def is_file(self, path: Path) -> bool: + return path.is_file() + + def is_symlink(self, path: Path) -> bool: + return path.is_symlink() + + def mkdir( + self, path: Path, *, parents: bool = False, exist_ok: bool = False + ) -> None: + path.mkdir(parents=parents, exist_ok=exist_ok) + + def unlink(self, path: Path) -> None: + path.unlink() + + def rmtree(self, path: Path) -> None: + shutil.rmtree(path) + + def read_text(self, path: Path) -> str: + return path.read_text() + + def write_text(self, path: Path, content: str) -> None: + path.write_text(content) + + def copy(self, source: Path, target: Path) -> None: + if source.is_dir(): + shutil.copytree(source, target) + else: + shutil.copy2(source, target) + + def home(self) -> Path: + return Path.home() + + +command_runner: CommandRunner = SubprocessRunner() +filesystem: FilesystemBackend = LocalFilesystemBackend() diff --git a/kiauh/core/backends/tests/__init__.py b/kiauh/core/backends/tests/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/kiauh/core/backends/tests/test_backends.py b/kiauh/core/backends/tests/test_backends.py new file mode 100644 index 00000000..153049a9 --- /dev/null +++ b/kiauh/core/backends/tests/test_backends.py @@ -0,0 +1,101 @@ +from __future__ import annotations + +import subprocess +from pathlib import Path + +import pytest +from core import backends +from core.backends import LocalFilesystemBackend, SubprocessRunner +from tests.helpers.fake_backends import FakeCommandRunner, FakeFilesystemBackend +from utils import fs_utils, sys_utils + + +class TestSubprocessRunner: + def test_run_executes_command(self) -> None: + runner = SubprocessRunner() + result = runner.run(["true"]) + assert result.returncode == 0 + + def test_check_output_returns_stdout(self) -> None: + runner = SubprocessRunner() + output = runner.check_output(["echo", "hello"], text=True) + assert "hello" in output + + +class TestCommandRunnerInjection: + def test_sys_utils_uses_injected_runner(self, monkeypatch) -> None: + fake = FakeCommandRunner({ + ("some", "cmd"): subprocess.CompletedProcess(["some", "cmd"], 0, "", "") + }) + monkeypatch.setattr(backends, "command_runner", fake) + + sys_utils.run(["some", "cmd"], check=True) + + assert fake.calls[0][0] == ["some", "cmd"] + assert fake.calls[0][1].get("check") is True + + def test_cmd_sysctl_service_records_command(self, monkeypatch) -> None: + expected_cmd = ["sudo", "systemctl", "start", "klipper.service"] + fake = FakeCommandRunner({ + tuple(expected_cmd): subprocess.CompletedProcess(expected_cmd, 0, "", "") + }) + monkeypatch.setattr(backends, "command_runner", fake) + + sys_utils.cmd_sysctl_service("klipper.service", "start") + + assert fake.calls[0][0] == expected_cmd + + @pytest.mark.parametrize("module", [sys_utils, fs_utils]) + def test_single_shared_command_runner_registry(self, monkeypatch, module) -> None: + # there is only ONE ``command_runner`` global to patch. + # Patching ``core.backends.command_runner`` must affect every wrapper + # (sys_utils.run, fs_utils.run, enum helpers) — no per-module duplicates. + fake = FakeCommandRunner({ + ("shared", "cmd"): subprocess.CompletedProcess(["shared", "cmd"], 0, "", "") + }) + monkeypatch.setattr(backends, "command_runner", fake) + + module.run(["shared", "cmd"], check=True) + + assert fake.calls[0][0] == ["shared", "cmd"] + + +class TestLocalFilesystemBackend: + def test_write_and_read_text(self, tmp_path: Path) -> None: + fs = LocalFilesystemBackend() + target = tmp_path / "test.txt" + fs.write_text(target, "hello") + assert fs.read_text(target) == "hello" + + def test_mkdir_and_exists(self, tmp_path: Path) -> None: + fs = LocalFilesystemBackend() + target = tmp_path / "new_dir" + assert not fs.exists(target) + fs.mkdir(target) + assert fs.exists(target) + assert fs.is_dir(target) + + +class TestFilesystemBackendInjection: + def test_create_folders_uses_injected_fs(self, monkeypatch) -> None: + fake = FakeFilesystemBackend() + monkeypatch.setattr(backends, "filesystem", fake) + + fs_utils.create_folders([Path("/tmp/a"), Path("/tmp/b")]) + + assert fake.exists(Path("/tmp/a")) + assert fake.exists(Path("/tmp/b")) + + def test_run_remove_routines_uses_injected_fs(self, monkeypatch) -> None: + fake = FakeFilesystemBackend() + fake.add_file(Path("/tmp/file.txt"), "x") + monkeypatch.setattr(backends, "filesystem", fake) + + assert fs_utils.run_remove_routines(Path("/tmp/file.txt")) is True + assert not fake.exists(Path("/tmp/file.txt")) + + def test_run_remove_routines_skips_missing_file(self, monkeypatch) -> None: + fake = FakeFilesystemBackend() + monkeypatch.setattr(backends, "filesystem", fake) + + assert fs_utils.run_remove_routines(Path("/tmp/missing")) is False diff --git a/kiauh/tests/__init__.py b/kiauh/tests/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/kiauh/tests/helpers/__init__.py b/kiauh/tests/helpers/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/kiauh/tests/helpers/fake_backends.py b/kiauh/tests/helpers/fake_backends.py new file mode 100644 index 00000000..1042be80 --- /dev/null +++ b/kiauh/tests/helpers/fake_backends.py @@ -0,0 +1,178 @@ +# ======================================================================= # +# Test-only backends. Not imported by production code. # +# ======================================================================= # +from __future__ import annotations + +import subprocess +from pathlib import Path +from typing import Any, Dict, List, Sequence, Tuple + + +class FakeCommandRunner: + """Command runner for tests. Records calls and returns scripted responses. + + By default, running a command that was not explicitly scripted raises an + error so missing mocks are caught during development. Pass + ``strict=False`` to restore the legacy "default success" behavior. + """ + + def __init__( + self, + responses: Dict[Tuple[str, ...], subprocess.CompletedProcess] | None = None, + *, + strict: bool = True, + ) -> None: + self.calls: List[Tuple[str | Sequence[str], Dict[str, Any]]] = [] + self.responses = responses or {} + self.strict = strict + + @staticmethod + def _key(cmd: str | Sequence[str]) -> Tuple[str, ...]: + if isinstance(cmd, str): + return (cmd,) + return tuple(str(c) for c in cmd) + + def _make_response( + self, cmd: str | Sequence[str], returncode: int = 0 + ) -> subprocess.CompletedProcess: + return subprocess.CompletedProcess( + args=cmd, + returncode=returncode, + stdout="", + stderr="", + ) + + def _unscripted(self, cmd: str | Sequence[str]) -> subprocess.CompletedProcess: + if self.strict: + raise RuntimeError(f"Unscripted command: {cmd}") + return self._make_response(cmd) + + def run( + self, + cmd: str | Sequence[str], + **kwargs: Any, + ) -> subprocess.CompletedProcess: + self.calls.append((cmd, kwargs)) + key = self._key(cmd) + if key in self.responses: + return self.responses[key] + return self._unscripted(cmd) + + def check_output( + self, + cmd: str | Sequence[str], + **kwargs: Any, + ) -> str | bytes: + self.calls.append((cmd, kwargs)) + key = self._key(cmd) + if key in self.responses: + return self.responses[key].stdout # type: ignore[no-any-return] + if self.strict: + raise RuntimeError(f"Unscripted command: {cmd}") + return "" + + def call( + self, + cmd: str | Sequence[str], + **kwargs: Any, + ) -> int: + self.calls.append((cmd, kwargs)) + key = self._key(cmd) + if key in self.responses: + return self.responses[key].returncode + if self.strict: + raise RuntimeError(f"Unscripted command: {cmd}") + return 0 + + def popen( + self, + cmd: str | Sequence[str], + **kwargs: Any, + ) -> subprocess.Popen: + raise NotImplementedError("FakeCommandRunner.popen is not implemented") + + +class FakeFilesystemBackend: + """In-memory filesystem backend for tests.""" + + def __init__(self) -> None: + self.dirs: set[str] = set() + self.files: Dict[str, str] = {} + self.symlinks: Dict[str, str] = {} + self._home: Path = Path("/home/test") + + def _path(self, path: Path) -> str: + return str(Path(path).resolve()) + + def exists(self, path: Path) -> bool: + key = self._path(path) + return key in self.dirs or key in self.files or key in self.symlinks + + def is_dir(self, path: Path) -> bool: + return self._path(path) in self.dirs + + def is_file(self, path: Path) -> bool: + return self._path(path) in self.files + + def is_symlink(self, path: Path) -> bool: + return self._path(path) in self.symlinks + + def mkdir( + self, path: Path, *, parents: bool = False, exist_ok: bool = False + ) -> None: + key = self._path(path) + if key in self.files and not exist_ok: + raise FileExistsError(key) + if key in self.dirs and not exist_ok: + raise FileExistsError(key) + if parents: + for parent in reversed(Path(key).parents): + self.dirs.add(str(parent)) + self.dirs.add(key) + + def unlink(self, path: Path) -> None: + key = self._path(path) + if key in self.files: + del self.files[key] + elif key in self.symlinks: + del self.symlinks[key] + else: + raise FileNotFoundError(key) + + def rmtree(self, path: Path) -> None: + key = self._path(path) + if key not in self.dirs: + raise FileNotFoundError(key) + prefix = key + "/" + self.dirs = {d for d in self.dirs if not (d == key or d.startswith(prefix))} + self.files = {k: v for k, v in self.files.items() if not k.startswith(prefix)} + self.symlinks = { + k: v for k, v in self.symlinks.items() if not k.startswith(prefix) + } + + def read_text(self, path: Path) -> str: + key = self._path(path) + if key not in self.files: + raise FileNotFoundError(key) + return self.files[key] + + def write_text(self, path: Path, content: str) -> None: + key = self._path(path) + self.files[key] = content + self.dirs.discard(key) + + def copy(self, source: Path, target: Path) -> None: + content = self.read_text(source) + self.write_text(target, content) + + def home(self) -> Path: + return self._home + + def add_dir(self, path: Path) -> None: + self.dirs.add(self._path(path)) + + def add_file(self, path: Path, content: str = "") -> None: + self.files[self._path(path)] = content + + def add_symlink(self, path: Path, target: Path) -> None: + self.symlinks[self._path(path)] = str(target) diff --git a/kiauh/tests/test_main.py b/kiauh/tests/test_main.py new file mode 100644 index 00000000..5cc91a1a --- /dev/null +++ b/kiauh/tests/test_main.py @@ -0,0 +1,98 @@ +# ======================================================================= # +# Copyright (C) 2020 - 2026 Dominik Willner # +# # +# This file is part of KIAUH - Klipper Installation And Update Helper # +# https://github.com/dw-0/kiauh # +# # +# This file may be distributed under the terms of the GNU GPLv3 license # +# ======================================================================= # +from __future__ import annotations + +from typing import List + +import main as main_module +import pytest + + +class _FakeMainMenu: + """Minimal stand-in for ``core.menus.main_menu.MainMenu``.""" + + instances: List["_FakeMainMenu"] = [] + + def __init__(self) -> None: + self._run = False + type(self).instances.append(self) + + def run(self) -> None: + self._run = True + + @classmethod + def reset(cls) -> None: + cls.instances = [] + + +@pytest.fixture(autouse=True) +def _reset_fake_menu() -> None: + _FakeMainMenu.reset() + yield + _FakeMainMenu.reset() + + +def _patch_tui_seeds(monkeypatch: pytest.MonkeyPatch) -> None: + """Neutralise the heavyweight side-effects triggered when launching the TUI.""" + monkeypatch.setattr(main_module, "KiauhSettings", lambda: None) + monkeypatch.setattr(main_module, "ensure_encoding", lambda: None) + monkeypatch.setattr(main_module, "MainMenu", _FakeMainMenu) + + +class TestMainDispatch: + def test_no_command_launches_tui(self, monkeypatch: pytest.MonkeyPatch) -> None: + # rc == -1 means "fall back to the TUI": ``MainMenu().run()`` is called. + monkeypatch.setattr(main_module, "run_cli", lambda: -1) + _patch_tui_seeds(monkeypatch) + + main_module.main() + + assert _FakeMainMenu.instances + assert all(m._run for m in _FakeMainMenu.instances) + + def test_cli_success_returns_cleanly( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + # rc == 0 means the CLI succeeded; the TUI must NOT start and main must + # NOT call sys.exit. + monkeypatch.setattr(main_module, "run_cli", lambda: 0) + _patch_tui_seeds(monkeypatch) + + main_module.main() # must not raise SystemExit + + assert _FakeMainMenu.instances == [] + + def test_cli_failure_exits_nonzero( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + # rc > 0 means the CLI reported a failure; main must propagate via sys.exit. + monkeypatch.setattr(main_module, "run_cli", lambda: 2) + _patch_tui_seeds(monkeypatch) + + with pytest.raises(SystemExit) as exc: + main_module.main() + + assert exc.value.code == 2 + assert _FakeMainMenu.instances == [] + + def test_tui_keyboard_interrupt_is_absorbed( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + # A Ctrl-C while the TUI runs must be caught and printed friendly + # instead of crashing with a traceback. + class _InterruptingMenu(_FakeMainMenu): + def run(self) -> None: + raise KeyboardInterrupt() + + monkeypatch.setattr(main_module, "run_cli", lambda: -1) + monkeypatch.setattr(main_module, "KiauhSettings", lambda: None) + monkeypatch.setattr(main_module, "ensure_encoding", lambda: None) + monkeypatch.setattr(main_module, "MainMenu", _InterruptingMenu) + + main_module.main() # must not raise; KeyboardInterrupt is absorbed \ No newline at end of file