Kevin Veen-Birkenbach
37f3057d31
Some checks failed
Mark stable commit / test-unit (push) Has been cancelled
Mark stable commit / test-integration (push) Has been cancelled
Mark stable commit / test-env-virtual (push) Has been cancelled
Mark stable commit / test-env-nix (push) Has been cancelled
Mark stable commit / test-e2e (push) Has been cancelled
Mark stable commit / test-virgin-user (push) Has been cancelled
Mark stable commit / test-virgin-root (push) Has been cancelled
Mark stable commit / codesniffer-shellcheck (push) Has been cancelled
Mark stable commit / codesniffer-ruff (push) Has been cancelled
Mark stable commit / mark-stable (push) Has been cancelled
CI / test-unit (push) Has been cancelled
CI / test-integration (push) Has been cancelled
CI / test-env-virtual (push) Has been cancelled
CI / test-env-nix (push) Has been cancelled
CI / test-e2e (push) Has been cancelled
CI / test-virgin-user (push) Has been cancelled
CI / test-virgin-root (push) Has been cancelled
CI / codesniffer-shellcheck (push) Has been cancelled
CI / codesniffer-ruff (push) Has been cancelled
- Add TYPE_CHECKING imports for RepoContext and CommandRunner to avoid runtime deps - Fix Ruff F821 undefined-name errors in nix installer modules - Refactor legacy NixFlakeInstaller unit tests to mock subprocess.run directly - Remove obsolete run_cmd_mock usage and assert install calls via subprocess calls - Ensure tests run without realtime waits or external nix dependencies https://chatgpt.com/share/693e925d-a79c-800f-b0b6-92b8ba260b11
65 lines
1.9 KiB
Python
65 lines
1.9 KiB
Python
from __future__ import annotations
|
|
|
|
import subprocess
|
|
|
|
from typing import TYPE_CHECKING
|
|
|
|
from .types import RunResult
|
|
|
|
if TYPE_CHECKING:
|
|
from pkgmgr.actions.install.context import RepoContext
|
|
|
|
class CommandRunner:
|
|
"""
|
|
Executes commands (shell=True) inside a repository directory (if provided).
|
|
Supports preview mode and compact failure output logging.
|
|
"""
|
|
|
|
def run(self, ctx: "RepoContext", cmd: str, allow_failure: bool) -> RunResult:
|
|
repo_dir = getattr(ctx, "repo_dir", None) or getattr(ctx, "repo_path", None)
|
|
preview = bool(getattr(ctx, "preview", False))
|
|
quiet = bool(getattr(ctx, "quiet", False))
|
|
|
|
if preview:
|
|
if not quiet:
|
|
print(f"[preview] {cmd}")
|
|
return RunResult(returncode=0, stdout="", stderr="")
|
|
|
|
try:
|
|
p = subprocess.run(
|
|
cmd,
|
|
shell=True,
|
|
cwd=repo_dir,
|
|
check=False,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
)
|
|
except Exception as e:
|
|
if not allow_failure:
|
|
raise
|
|
return RunResult(returncode=1, stdout="", stderr=str(e))
|
|
|
|
res = RunResult(returncode=p.returncode, stdout=p.stdout or "", stderr=p.stderr or "")
|
|
|
|
if res.returncode != 0 and not quiet:
|
|
self._print_compact_failure(res)
|
|
|
|
if res.returncode != 0 and not allow_failure:
|
|
raise SystemExit(res.returncode)
|
|
|
|
return res
|
|
|
|
@staticmethod
|
|
def _print_compact_failure(res: RunResult) -> None:
|
|
out = (res.stdout or "").strip()
|
|
err = (res.stderr or "").strip()
|
|
|
|
if out:
|
|
print("[nix] stdout (last lines):")
|
|
print("\n".join(out.splitlines()[-20:]))
|
|
|
|
if err:
|
|
print("[nix] stderr (last lines):")
|
|
print("\n".join(err.splitlines()[-40:]))
|