mirror of
https://github.com/dw-0/kiauh.git
synced 2026-08-03 04:47:56 +05:00
feat(tests): add pluggable subprocess and filesystem backends
Introduce core.backends with CommandRunner and FilesystemBackend protocols plus default subprocess/local filesystem implementations. Add fake backends in tests.helpers.fake_backends for isolated unit tests.
This commit is contained in:
dw-0
@@ -0,0 +1,182 @@
|
||||
# ======================================================================= #
|
||||
# Copyright (C) 2020 - 2026 Dominik Willner <th33xitus@gmail.com> #
|
||||
# #
|
||||
# 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()
|
||||
@@ -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
|
||||
@@ -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)
|
||||
@@ -0,0 +1,98 @@
|
||||
# ======================================================================= #
|
||||
# Copyright (C) 2020 - 2026 Dominik Willner <th33xitus@gmail.com> #
|
||||
# #
|
||||
# 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
|
||||
Reference in New Issue
Block a user