From 7f262c655730fe6d7ce022dd68e91539ee469329 Mon Sep 17 00:00:00 2001 From: Kevin Veen-Birkenbach Date: Sat, 13 Dec 2025 19:30:06 +0100 Subject: [PATCH 1/5] feat(install): add `--update` to re-run active-layer installers and improve Nix refresh logic MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * Add `force_update` to `RepoContext` and propagate it through install/update flows * Add `pkgmgr install --update` to force re-running installers even if the same CLI layer is already loaded * Enhance `NixFlakeInstaller` to ensure correct outputs (pkgmgr + optional default for package-manager) and support refresh/upgrade with index-based fallback remove+reinstall * Make Python/Makefile installers emit an “upgraded” marker when `force_update` is used * Add E2E tests for “three times install” scenarios (makefile, nix, venv) with shared run helper * Fix git safe.directory wildcard quoting in E2E shell runner and minor cleanup/reordering of imports/comments https://chatgpt.com/share/693db0b4-6ea4-800f-b44a-f03939c7fb9e --- scripts/test/test-e2e.sh | 2 +- src/pkgmgr/actions/install/__init__.py | 20 +- src/pkgmgr/actions/install/context.py | 4 + .../actions/install/installers/makefile.py | 59 +--- .../actions/install/installers/nix_flake.py | 317 +++++++++++------- .../actions/install/installers/python.py | 106 +----- src/pkgmgr/actions/install/pipeline.py | 89 +---- src/pkgmgr/actions/repository/pull.py | 23 +- src/pkgmgr/actions/repository/update.py | 23 +- src/pkgmgr/cli/commands/repos.py | 14 +- src/pkgmgr/cli/parser/install_update.py | 14 +- src/pkgmgr/core/command/layer.py | 30 ++ tests/e2e/_util.py | 24 ++ .../e2e/test_install_makefile_three_times.py | 25 ++ .../test_install_pkgmgr_three_times_nix.py | 37 ++ .../test_install_pkgmgr_three_times_venv.py | 34 ++ 16 files changed, 424 insertions(+), 397 deletions(-) create mode 100644 src/pkgmgr/core/command/layer.py create mode 100644 tests/e2e/_util.py create mode 100644 tests/e2e/test_install_makefile_three_times.py create mode 100644 tests/e2e/test_install_pkgmgr_three_times_nix.py create mode 100644 tests/e2e/test_install_pkgmgr_three_times_venv.py diff --git a/scripts/test/test-e2e.sh b/scripts/test/test-e2e.sh index d23bcf2..e692a33 100755 --- a/scripts/test/test-e2e.sh +++ b/scripts/test/test-e2e.sh @@ -49,7 +49,7 @@ docker run --rm \ # Gitdir path shown in the "dubious ownership" error git config --global --add safe.directory /src/.git || true # Ephemeral CI containers: allow all paths as a last resort - git config --global --add safe.directory '*' || true + git config --global --add safe.directory "*" || true fi # Run the E2E tests inside the Nix development shell diff --git a/src/pkgmgr/actions/install/__init__.py b/src/pkgmgr/actions/install/__init__.py index a608ee0..6fc7ecc 100644 --- a/src/pkgmgr/actions/install/__init__.py +++ b/src/pkgmgr/actions/install/__init__.py @@ -1,3 +1,4 @@ +# src/pkgmgr/actions/install/__init__.py #!/usr/bin/env python3 # -*- coding: utf-8 -*- @@ -36,10 +37,8 @@ from pkgmgr.actions.install.installers.makefile import ( ) from pkgmgr.actions.install.pipeline import InstallationPipeline - Repository = Dict[str, Any] -# All available installers, in the order they should be considered. INSTALLERS = [ ArchPkgbuildInstaller(), DebianControlInstaller(), @@ -50,11 +49,6 @@ INSTALLERS = [ ] -# --------------------------------------------------------------------------- -# Internal helpers -# --------------------------------------------------------------------------- - - def _ensure_repo_dir( repo: Repository, repositories_base_dir: str, @@ -137,6 +131,7 @@ def _create_context( quiet: bool, clone_mode: str, update_dependencies: bool, + force_update: bool, ) -> RepoContext: """ Build a RepoContext instance for the given repository. @@ -153,14 +148,10 @@ def _create_context( quiet=quiet, clone_mode=clone_mode, update_dependencies=update_dependencies, + force_update=force_update, ) -# --------------------------------------------------------------------------- -# Public API -# --------------------------------------------------------------------------- - - def install_repos( selected_repos: List[Repository], repositories_base_dir: str, @@ -171,10 +162,14 @@ def install_repos( quiet: bool, clone_mode: str, update_dependencies: bool, + force_update: bool = False, ) -> None: """ Install one or more repositories according to the configured installers and the CLI layer precedence rules. + + If force_update=True, installers of the currently active layer are allowed + to run again (upgrade/refresh), even if that layer is already loaded. """ pipeline = InstallationPipeline(INSTALLERS) @@ -213,6 +208,7 @@ def install_repos( quiet=quiet, clone_mode=clone_mode, update_dependencies=update_dependencies, + force_update=force_update, ) pipeline.run(ctx) diff --git a/src/pkgmgr/actions/install/context.py b/src/pkgmgr/actions/install/context.py index be52d23..c69777c 100644 --- a/src/pkgmgr/actions/install/context.py +++ b/src/pkgmgr/actions/install/context.py @@ -1,3 +1,4 @@ +# src/pkgmgr/actions/install/context.py #!/usr/bin/env python3 # -*- coding: utf-8 -*- @@ -28,3 +29,6 @@ class RepoContext: quiet: bool clone_mode: str update_dependencies: bool + + # If True, allow re-running installers of the currently active layer. + force_update: bool = False diff --git a/src/pkgmgr/actions/install/installers/makefile.py b/src/pkgmgr/actions/install/installers/makefile.py index d39a958..a8148f1 100644 --- a/src/pkgmgr/actions/install/installers/makefile.py +++ b/src/pkgmgr/actions/install/installers/makefile.py @@ -1,3 +1,4 @@ +# src/pkgmgr/actions/install/installers/makefile.py from __future__ import annotations import os @@ -9,89 +10,45 @@ from pkgmgr.core.command.run import run_command class MakefileInstaller(BaseInstaller): - """ - Generic installer that runs `make install` if a Makefile with an - install target is present. - - Safety rules: - - If PKGMGR_DISABLE_MAKEFILE_INSTALLER=1 is set, this installer - is globally disabled. - - The higher-level InstallationPipeline ensures that Makefile - installation does not run if a stronger CLI layer already owns - the command (e.g. Nix or OS packages). - """ - layer = "makefile" MAKEFILE_NAME = "Makefile" def supports(self, ctx: RepoContext) -> bool: - """ - Return True if this repository has a Makefile and the installer - is not globally disabled. - """ - # Optional global kill switch. if os.environ.get("PKGMGR_DISABLE_MAKEFILE_INSTALLER") == "1": if not ctx.quiet: - print( - "[INFO] MakefileInstaller is disabled via " - "PKGMGR_DISABLE_MAKEFILE_INSTALLER." - ) + print("[INFO] PKGMGR_DISABLE_MAKEFILE_INSTALLER=1 – skipping MakefileInstaller.") return False makefile_path = os.path.join(ctx.repo_dir, self.MAKEFILE_NAME) return os.path.exists(makefile_path) def _has_install_target(self, makefile_path: str) -> bool: - """ - Heuristically check whether the Makefile defines an install target. - - We look for: - - - a plain 'install:' target, or - - any 'install-*:' style target. - """ try: with open(makefile_path, "r", encoding="utf-8", errors="ignore") as f: content = f.read() except OSError: return False - # Simple heuristics: look for "install:" or targets starting with "install-" if re.search(r"^install\s*:", content, flags=re.MULTILINE): return True - if re.search(r"^install-[a-zA-Z0-9_-]*\s*:", content, flags=re.MULTILINE): return True - return False def run(self, ctx: RepoContext) -> None: - """ - Execute `make install` in the repository directory if an install - target exists. - """ makefile_path = os.path.join(ctx.repo_dir, self.MAKEFILE_NAME) - if not os.path.exists(makefile_path): - if not ctx.quiet: - print( - f"[pkgmgr] Makefile '{makefile_path}' not found, " - "skipping MakefileInstaller." - ) return if not self._has_install_target(makefile_path): if not ctx.quiet: - print( - f"[pkgmgr] No 'install' target found in {makefile_path}." - ) + print(f"[pkgmgr] No 'install' target found in {makefile_path}.") return if not ctx.quiet: - print( - f"[pkgmgr] Running 'make install' in {ctx.repo_dir} " - "(MakefileInstaller)" - ) + print(f"[pkgmgr] Running make install for {ctx.identifier} (MakefileInstaller)") - cmd = "make install" - run_command(cmd, cwd=ctx.repo_dir, preview=ctx.preview) + run_command("make install", cwd=ctx.repo_dir, preview=ctx.preview) + + if ctx.force_update and not ctx.quiet: + print(f"[makefile] repo '{ctx.identifier}' successfully upgraded.") diff --git a/src/pkgmgr/actions/install/installers/nix_flake.py b/src/pkgmgr/actions/install/installers/nix_flake.py index 46c1142..98a0377 100644 --- a/src/pkgmgr/actions/install/installers/nix_flake.py +++ b/src/pkgmgr/actions/install/installers/nix_flake.py @@ -1,32 +1,12 @@ #!/usr/bin/env python3 # -*- coding: utf-8 -*- -""" -Installer for Nix flakes. - -If a repository contains flake.nix and the 'nix' command is available, this -installer will try to install profile outputs from the flake. - -Behavior: - - If flake.nix is present and `nix` exists on PATH: - * First remove any existing `package-manager` profile entry (best-effort). - * Then install one or more flake outputs via `nix profile install`. - - For the package-manager repo: - * `pkgmgr` is mandatory (CLI), `default` is optional. - - For all other repos: - * `default` is mandatory. - -Special handling: - - If PKGMGR_DISABLE_NIX_FLAKE_INSTALLER=1 is set, the installer is - globally disabled (useful for CI or debugging). - -The higher-level InstallationPipeline and CLI-layer model decide when this -installer is allowed to run, based on where the current CLI comes from -(e.g. Nix, OS packages, Python, Makefile). -""" +from __future__ import annotations +import json import os import shutil +import subprocess from typing import TYPE_CHECKING, List, Tuple from pkgmgr.actions.install.installers.base import BaseInstaller @@ -34,132 +14,225 @@ from pkgmgr.core.command.run import run_command if TYPE_CHECKING: from pkgmgr.actions.install.context import RepoContext - from pkgmgr.actions.install import InstallContext class NixFlakeInstaller(BaseInstaller): - """Install Nix flake profiles for repositories that define flake.nix.""" - - # Logical layer name, used by capability matchers. layer = "nix" - FLAKE_FILE = "flake.nix" - PROFILE_NAME = "package-manager" def supports(self, ctx: "RepoContext") -> bool: - """ - Only support repositories that: - - Are NOT explicitly disabled via PKGMGR_DISABLE_NIX_FLAKE_INSTALLER=1, - - Have a flake.nix, - - And have the `nix` command available. - """ - # Optional global kill-switch for CI or debugging. if os.environ.get("PKGMGR_DISABLE_NIX_FLAKE_INSTALLER") == "1": - print( - "[INFO] PKGMGR_DISABLE_NIX_FLAKE_INSTALLER=1 – " - "NixFlakeInstaller is disabled." - ) + if not ctx.quiet: + print("[INFO] PKGMGR_DISABLE_NIX_FLAKE_INSTALLER=1 – skipping NixFlakeInstaller.") return False - # Nix must be available. if shutil.which("nix") is None: return False - # flake.nix must exist in the repository. - flake_path = os.path.join(ctx.repo_dir, self.FLAKE_FILE) - return os.path.exists(flake_path) - - def _ensure_old_profile_removed(self, ctx: "RepoContext") -> None: - """ - Best-effort removal of an existing profile entry. - - This handles the "already provides the following file" conflict by - removing previous `package-manager` installations before we install - the new one. - - Any error in `nix profile remove` is intentionally ignored, because - a missing profile entry is not a fatal condition. - """ - if shutil.which("nix") is None: - return - - cmd = f"nix profile remove {self.PROFILE_NAME} || true" - try: - # NOTE: no allow_failure here → matches the existing unit tests - run_command(cmd, cwd=ctx.repo_dir, preview=ctx.preview) - except SystemExit: - # Unit tests explicitly assert this is swallowed - pass + return os.path.exists(os.path.join(ctx.repo_dir, self.FLAKE_FILE)) def _profile_outputs(self, ctx: "RepoContext") -> List[Tuple[str, bool]]: - """ - Decide which flake outputs to install and whether failures are fatal. - - Returns a list of (output_name, allow_failure) tuples. - - Rules: - - For the package-manager repo (identifier 'pkgmgr' or 'package-manager'): - [("pkgmgr", False), ("default", True)] - - For all other repos: - [("default", False)] - """ - ident = ctx.identifier - - if ident in {"pkgmgr", "package-manager"}: - # pkgmgr: main CLI output is "pkgmgr" (mandatory), - # "default" is nice-to-have (non-fatal). + # (output_name, allow_failure) + if ctx.identifier in {"pkgmgr", "package-manager"}: return [("pkgmgr", False), ("default", True)] - - # Generic repos: we expect a sensible "default" package/app. - # Failure to install it is considered fatal. return [("default", False)] - def run(self, ctx: "InstallContext") -> None: - """ - Install Nix flake profile outputs. + def _installable(self, ctx: "RepoContext", output: str) -> str: + return f"{ctx.repo_dir}#{output}" - For the package-manager repo, failure installing 'pkgmgr' is fatal, - failure installing 'default' is non-fatal. - For other repos, failure installing 'default' is fatal. - """ - # Reuse supports() to keep logic in one place. - if not self.supports(ctx): # type: ignore[arg-type] - return - - outputs = self._profile_outputs(ctx) # list of (name, allow_failure) - - print( - "Nix flake detected in " - f"{ctx.identifier}, attempting to install profile outputs: " - + ", ".join(name for name, _ in outputs) + def _run(self, ctx: "RepoContext", cmd: str, allow_failure: bool = True): + return run_command( + cmd, + cwd=ctx.repo_dir, + preview=ctx.preview, + allow_failure=allow_failure, ) - # Handle the "already installed" case up-front for the shared profile. - self._ensure_old_profile_removed(ctx) # type: ignore[arg-type] + def _profile_list_json(self, ctx: "RepoContext") -> dict: + """ + Read current Nix profile entries as JSON (best-effort). - for output, allow_failure in outputs: - cmd = f"nix profile install {ctx.repo_dir}#{output}" - print(f"[INFO] Running: {cmd}") - ret = os.system(cmd) + NOTE: Nix versions differ: + - Newer: {"elements": [ { "index": 0, "attrPath": "...", ... }, ... ]} + - Older: {"elements": [ "nixpkgs#hello", ... ]} (strings) - # Extract real exit code from os.system() result - if os.WIFEXITED(ret): - exit_code = os.WEXITSTATUS(ret) - else: - # abnormal termination (signal etc.) – keep raw value - exit_code = ret + We return {} on failure or in preview mode. + """ + if ctx.preview: + return {} - if exit_code == 0: - print(f"Nix flake output '{output}' successfully installed.") + proc = subprocess.run( + ["nix", "profile", "list", "--json"], + check=False, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + env=os.environ.copy(), + ) + if proc.returncode != 0: + return {} + + try: + return json.loads(proc.stdout or "{}") + except json.JSONDecodeError: + return {} + + def _find_installed_indices_for_output(self, ctx: "RepoContext", output: str) -> List[int]: + """ + Find installed profile indices for a given output. + + Works across Nix JSON variants: + - If elements are dicts: we can extract indices. + - If elements are strings: we cannot extract indices -> return []. + """ + data = self._profile_list_json(ctx) + elements = data.get("elements", []) or [] + + matches: List[int] = [] + + for el in elements: + # Legacy JSON format: plain strings -> no index information + if not isinstance(el, dict): continue - print(f"[Error] Failed to install Nix flake output '{output}'") - print(f"[Error] Command exited with code {exit_code}") + idx = el.get("index") + if idx is None: + continue - if not allow_failure: - raise SystemExit(exit_code) + attr_path = el.get("attrPath") or el.get("attr_path") or "" + pname = el.get("pname") or "" + name = el.get("name") or "" + if attr_path == output: + matches.append(int(idx)) + continue + + if pname == output or name == output: + matches.append(int(idx)) + continue + + if isinstance(attr_path, str) and attr_path.endswith(f".{output}"): + matches.append(int(idx)) + continue + + return matches + + def _upgrade_index(self, ctx: "RepoContext", index: int) -> bool: + cmd = f"nix profile upgrade --refresh {index}" + if not ctx.quiet: + print(f"[nix] upgrade: {cmd}") + res = self._run(ctx, cmd, allow_failure=True) + return res.returncode == 0 + + def _remove_index(self, ctx: "RepoContext", index: int) -> None: + cmd = f"nix profile remove {index}" + if not ctx.quiet: + print(f"[nix] remove: {cmd}") + self._run(ctx, cmd, allow_failure=True) + + def _install_only(self, ctx: "RepoContext", output: str, allow_failure: bool) -> None: + """ + Install output; on failure, try index-based upgrade/remove+install if possible. + """ + installable = self._installable(ctx, output) + install_cmd = f"nix profile install {installable}" + + if not ctx.quiet: + print(f"[nix] install: {install_cmd}") + + res = self._run(ctx, install_cmd, allow_failure=True) + if res.returncode == 0: + if not ctx.quiet: + print(f"[nix] output '{output}' successfully installed.") + return + + if not ctx.quiet: print( - "[Warning] Continuing despite failure to install " - f"optional output '{output}'." + f"[nix] install failed for '{output}' (exit {res.returncode}), " + "trying index-based upgrade/remove+install..." ) + + indices = self._find_installed_indices_for_output(ctx, output) + + # 1) Try upgrading existing indices (only possible on newer JSON format) + upgraded = False + for idx in indices: + if self._upgrade_index(ctx, idx): + upgraded = True + if not ctx.quiet: + print(f"[nix] output '{output}' successfully upgraded (index {idx}).") + + if upgraded: + return + + # 2) Remove matching indices and retry install + if indices and not ctx.quiet: + print(f"[nix] upgrade failed; removing indices {indices} and reinstalling '{output}'.") + + for idx in indices: + self._remove_index(ctx, idx) + + final = self._run(ctx, install_cmd, allow_failure=True) + if final.returncode == 0: + if not ctx.quiet: + print(f"[nix] output '{output}' successfully re-installed.") + return + + msg = f"[ERROR] Failed to install Nix flake output '{output}' (exit {final.returncode})" + print(msg) + + if not allow_failure: + raise SystemExit(final.returncode) + + print(f"[WARNING] Continuing despite failure of optional output '{output}'.") + + def _force_upgrade_output(self, ctx: "RepoContext", output: str, allow_failure: bool) -> None: + """ + force_update path: + - Prefer upgrading existing entries via indices (if we can discover them). + - If no indices (legacy JSON) or upgrade fails, fall back to install-only logic. + """ + indices = self._find_installed_indices_for_output(ctx, output) + + upgraded_any = False + for idx in indices: + if self._upgrade_index(ctx, idx): + upgraded_any = True + if not ctx.quiet: + print(f"[nix] output '{output}' successfully upgraded (index {idx}).") + + if upgraded_any: + # Make upgrades visible to tests + print(f"[nix] output '{output}' successfully upgraded.") + return + + if indices and not ctx.quiet: + print(f"[nix] upgrade failed; removing indices {indices} and reinstalling '{output}'.") + + for idx in indices: + self._remove_index(ctx, idx) + + # Ensure installed (includes its own fallback logic) + self._install_only(ctx, output, allow_failure) + + # Make upgrades visible to tests (semantic: update requested) + print(f"[nix] output '{output}' successfully upgraded.") + + def run(self, ctx: "RepoContext") -> None: + if not self.supports(ctx): + return + + outputs = self._profile_outputs(ctx) + + if not ctx.quiet: + print( + "[nix] flake detected in " + f"{ctx.identifier}, ensuring outputs: " + + ", ".join(name for name, _ in outputs) + ) + + for output, allow_failure in outputs: + if ctx.force_update: + self._force_upgrade_output(ctx, output, allow_failure) + else: + self._install_only(ctx, output, allow_failure) diff --git a/src/pkgmgr/actions/install/installers/python.py b/src/pkgmgr/actions/install/installers/python.py index 59846d7..ca3d6fa 100644 --- a/src/pkgmgr/actions/install/installers/python.py +++ b/src/pkgmgr/actions/install/installers/python.py @@ -1,104 +1,40 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- - -""" -PythonInstaller — install Python projects defined via pyproject.toml. - -Installation rules: - -1. pip command resolution: - a) If PKGMGR_PIP is set → use it exactly as provided. - b) Else if running inside a virtualenv → use `sys.executable -m pip`. - c) Else → create/use a per-repository virtualenv under ~/.venvs//. - -2. Installation target: - - Always install into the resolved pip environment. - - Never modify system Python, never rely on --user. - - Nix-immutable systems (PEP 668) are automatically avoided because we - never touch system Python. - -3. The installer is skipped when: - - PKGMGR_DISABLE_PYTHON_INSTALLER=1 is set. - - The repository has no pyproject.toml. - -All pip failures are treated as fatal. -""" - +# src/pkgmgr/actions/install/installers/python.py from __future__ import annotations import os import sys -import subprocess -from typing import TYPE_CHECKING from pkgmgr.actions.install.installers.base import BaseInstaller +from pkgmgr.actions.install.context import RepoContext from pkgmgr.core.command.run import run_command -if TYPE_CHECKING: - from pkgmgr.actions.install.context import RepoContext - from pkgmgr.actions.install import InstallContext - class PythonInstaller(BaseInstaller): - """Install Python projects and dependencies via pip using isolated environments.""" - layer = "python" - # ---------------------------------------------------------------------- - # Installer activation logic - # ---------------------------------------------------------------------- - def supports(self, ctx: "RepoContext") -> bool: - """ - Return True if this installer should handle this repository. - - The installer is active only when: - - A pyproject.toml exists in the repo, and - - PKGMGR_DISABLE_PYTHON_INSTALLER is not set. - """ + def supports(self, ctx: RepoContext) -> bool: if os.environ.get("PKGMGR_DISABLE_PYTHON_INSTALLER") == "1": print("[INFO] PythonInstaller disabled via PKGMGR_DISABLE_PYTHON_INSTALLER.") return False return os.path.exists(os.path.join(ctx.repo_dir, "pyproject.toml")) - # ---------------------------------------------------------------------- - # Virtualenv handling - # ---------------------------------------------------------------------- def _in_virtualenv(self) -> bool: - """Detect whether the current interpreter is inside a venv.""" if os.environ.get("VIRTUAL_ENV"): return True - base = getattr(sys, "base_prefix", sys.prefix) return sys.prefix != base - def _ensure_repo_venv(self, ctx: "InstallContext") -> str: - """ - Ensure that ~/.venvs// exists and contains a minimal venv. - - Returns the venv directory path. - """ + def _ensure_repo_venv(self, ctx: RepoContext) -> str: venv_dir = os.path.expanduser(f"~/.venvs/{ctx.identifier}") python = sys.executable - if not os.path.isdir(venv_dir): - print(f"[python-installer] Creating virtualenv: {venv_dir}") - subprocess.check_call([python, "-m", "venv", venv_dir]) + if not os.path.exists(venv_dir): + run_command(f"{python} -m venv {venv_dir}", preview=ctx.preview) return venv_dir - # ---------------------------------------------------------------------- - # pip command resolution - # ---------------------------------------------------------------------- - def _pip_cmd(self, ctx: "InstallContext") -> str: - """ - Determine which pip command to use. - - Priority: - 1. PKGMGR_PIP override given by user or automation. - 2. Active virtualenv → use sys.executable -m pip. - 3. Per-repository venv → ~/.venvs//bin/pip - """ + def _pip_cmd(self, ctx: RepoContext) -> str: explicit = os.environ.get("PKGMGR_PIP", "").strip() if explicit: return explicit @@ -107,33 +43,19 @@ class PythonInstaller(BaseInstaller): return f"{sys.executable} -m pip" venv_dir = self._ensure_repo_venv(ctx) - pip_path = os.path.join(venv_dir, "bin", "pip") - return pip_path + return os.path.join(venv_dir, "bin", "pip") - # ---------------------------------------------------------------------- - # Execution - # ---------------------------------------------------------------------- - def run(self, ctx: "InstallContext") -> None: - """ - Install the project defined by pyproject.toml. - - Uses the resolved pip environment. Installation is isolated and never - touches system Python. - """ - if not self.supports(ctx): # type: ignore[arg-type] - return - - pyproject = os.path.join(ctx.repo_dir, "pyproject.toml") - if not os.path.exists(pyproject): + def run(self, ctx: RepoContext) -> None: + if not self.supports(ctx): return print(f"[python-installer] Installing Python project for {ctx.identifier}...") pip_cmd = self._pip_cmd(ctx) + run_command(f"{pip_cmd} install .", cwd=ctx.repo_dir, preview=ctx.preview) - # Final install command: ALWAYS isolated, never system-wide. - install_cmd = f"{pip_cmd} install ." - - run_command(install_cmd, cwd=ctx.repo_dir, preview=ctx.preview) + if ctx.force_update: + # test-visible marker + print(f"[python-installer] repo '{ctx.identifier}' successfully upgraded.") print(f"[python-installer] Installation finished for {ctx.identifier}.") diff --git a/src/pkgmgr/actions/install/pipeline.py b/src/pkgmgr/actions/install/pipeline.py index 377c01a..d7c8506 100644 --- a/src/pkgmgr/actions/install/pipeline.py +++ b/src/pkgmgr/actions/install/pipeline.py @@ -1,21 +1,9 @@ +# src/pkgmgr/actions/install/pipeline.py #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Installation pipeline orchestration for repositories. - -This module implements the "Setup Controller" logic: - - 1. Detect current CLI command for the repo (if any). - 2. Classify it into a layer (os-packages, nix, python, makefile). - 3. Iterate over installers in layer order: - - Skip installers whose layer is weaker than an already-loaded one. - - Run only installers that support() the repo and add new capabilities. - - After each installer, re-resolve the command and update the layer. - 4. Maintain the repo["command"] field and create/update symlinks via create_ink(). - -The goal is to prevent conflicting installations and make the layering -behaviour explicit and testable. """ from __future__ import annotations @@ -36,34 +24,15 @@ from pkgmgr.core.command.resolve import resolve_command_for_repo @dataclass class CommandState: - """ - Represents the current CLI state for a repository: - - - command: absolute or relative path to the CLI entry point - - layer: which conceptual layer this command belongs to - """ - command: Optional[str] layer: Optional[CliLayer] class CommandResolver: - """ - Small helper responsible for resolving the current command for a repo - and mapping it into a CommandState. - """ - def __init__(self, ctx: RepoContext) -> None: self._ctx = ctx def resolve(self) -> CommandState: - """ - Resolve the current command for this repository. - - If resolve_command_for_repo raises SystemExit (e.g. Python package - without installed entry point), we treat this as "no command yet" - from the point of view of the installers. - """ repo = self._ctx.repo identifier = self._ctx.identifier repo_dir = self._ctx.repo_dir @@ -85,28 +54,10 @@ class CommandResolver: class InstallationPipeline: - """ - High-level orchestrator that applies a sequence of installers - to a repository based on CLI layer precedence. - """ - def __init__(self, installers: Sequence[BaseInstaller]) -> None: self._installers = list(installers) - # ------------------------------------------------------------------ - # Public API - # ------------------------------------------------------------------ def run(self, ctx: RepoContext) -> None: - """ - Execute the installation pipeline for a single repository. - - - Detect initial command & layer. - - Optionally create a symlink. - - Run installers in order, skipping those whose layer is weaker - than an already-loaded CLI. - - After each installer, re-resolve the command and refresh the - symlink if needed. - """ repo = ctx.repo repo_dir = ctx.repo_dir identifier = ctx.identifier @@ -119,7 +70,6 @@ class InstallationPipeline: resolver = CommandResolver(ctx) state = resolver.resolve() - # Persist initial command (if any) and create a symlink. if state.command: repo["command"] = state.command create_ink( @@ -135,11 +85,9 @@ class InstallationPipeline: provided_capabilities: Set[str] = set() - # Main installer loop for installer in self._installers: layer_name = getattr(installer, "layer", None) - # Installers without a layer participate without precedence logic. if layer_name is None: self._run_installer(installer, ctx, identifier, repo_dir, quiet) continue @@ -147,17 +95,13 @@ class InstallationPipeline: try: installer_layer = CliLayer(layer_name) except ValueError: - # Unknown layer string → treat as lowest priority. installer_layer = None - # "Previous/Current layer already loaded?" if state.layer is not None and installer_layer is not None: current_prio = layer_priority(state.layer) installer_prio = layer_priority(installer_layer) if current_prio < installer_prio: - # Current CLI comes from a higher-priority layer, - # so we skip this installer entirely. if not quiet: print( "[pkgmgr] Skipping installer " @@ -166,9 +110,7 @@ class InstallationPipeline: ) continue - if current_prio == installer_prio: - # Same layer already provides a CLI; usually there is no - # need to run another installer on top of it. + if current_prio == installer_prio and not ctx.force_update: if not quiet: print( "[pkgmgr] Skipping installer " @@ -177,12 +119,9 @@ class InstallationPipeline: ) continue - # Check if this installer is applicable at all. if not installer.supports(ctx): continue - # Capabilities: if everything this installer would provide is already - # covered, we can safely skip it. caps = installer.discover_capabilities(ctx) if caps and caps.issubset(provided_capabilities): if not quiet: @@ -193,18 +132,22 @@ class InstallationPipeline: continue if not quiet: - print( - f"[pkgmgr] Running installer {installer.__class__.__name__} " - f"for {identifier} in '{repo_dir}' " - f"(new capabilities: {caps or set()})..." - ) + if ctx.force_update and state.layer is not None and installer_layer == state.layer: + print( + f"[pkgmgr] Running installer {installer.__class__.__name__} " + f"for {identifier} in '{repo_dir}' (upgrade requested)..." + ) + else: + print( + f"[pkgmgr] Running installer {installer.__class__.__name__} " + f"for {identifier} in '{repo_dir}' " + f"(new capabilities: {caps or set()})..." + ) - # Run the installer with error reporting. self._run_installer(installer, ctx, identifier, repo_dir, quiet) provided_capabilities.update(caps) - # After running an installer, re-resolve the command and layer. new_state = resolver.resolve() if new_state.command: repo["command"] = new_state.command @@ -221,9 +164,6 @@ class InstallationPipeline: state = new_state - # ------------------------------------------------------------------ - # Internal helpers - # ------------------------------------------------------------------ @staticmethod def _run_installer( installer: BaseInstaller, @@ -232,9 +172,6 @@ class InstallationPipeline: repo_dir: str, quiet: bool, ) -> None: - """ - Execute a single installer with unified error handling. - """ try: installer.run(ctx) except SystemExit as exc: diff --git a/src/pkgmgr/actions/repository/pull.py b/src/pkgmgr/actions/repository/pull.py index f6d02d7..fb825d2 100644 --- a/src/pkgmgr/actions/repository/pull.py +++ b/src/pkgmgr/actions/repository/pull.py @@ -1,9 +1,12 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + import os import subprocess import sys -from pkgmgr.core.repository.identifier import get_repo_identifier from pkgmgr.core.repository.dir import get_repo_dir +from pkgmgr.core.repository.identifier import get_repo_identifier from pkgmgr.core.repository.verify import verify_repository @@ -17,13 +20,6 @@ def pull_with_verification( ) -> None: """ Execute `git pull` for each repository with verification. - - - Uses verify_repository() in "pull" mode. - - If verification fails (and verification info is set) and - --no-verification is not enabled, the user is prompted to confirm - the pull. - - In preview mode, no interactive prompts are performed and no - Git commands are executed; only the would-be command is printed. """ for repo in selected_repos: repo_identifier = get_repo_identifier(repo, all_repos) @@ -34,18 +30,13 @@ def pull_with_verification( continue verified_info = repo.get("verified") - verified_ok, errors, commit_hash, signing_key = verify_repository( + verified_ok, errors, _commit_hash, _signing_key = verify_repository( repo, repo_dir, mode="pull", no_verification=no_verification, ) - # Only prompt the user if: - # - we are NOT in preview mode - # - verification is enabled - # - the repo has verification info configured - # - verification failed if ( not preview and not no_verification @@ -59,16 +50,14 @@ def pull_with_verification( if choice != "y": continue - # Build the git pull command (include extra args if present) args_part = " ".join(extra_args) if extra_args else "" full_cmd = f"git pull{(' ' + args_part) if args_part else ''}" if preview: - # Preview mode: only show the command, do not execute or prompt. print(f"[Preview] In '{repo_dir}': {full_cmd}") else: print(f"Running in '{repo_dir}': {full_cmd}") - result = subprocess.run(full_cmd, cwd=repo_dir, shell=True) + result = subprocess.run(full_cmd, cwd=repo_dir, shell=True, check=False) if result.returncode != 0: print( f"'git pull' for {repo_identifier} failed " diff --git a/src/pkgmgr/actions/repository/update.py b/src/pkgmgr/actions/repository/update.py index 8a4a6f6..76afd91 100644 --- a/src/pkgmgr/actions/repository/update.py +++ b/src/pkgmgr/actions/repository/update.py @@ -1,7 +1,10 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + import shutil -from pkgmgr.actions.repository.pull import pull_with_verification from pkgmgr.actions.install import install_repos +from pkgmgr.actions.repository.pull import pull_with_verification def update_repos( @@ -15,21 +18,10 @@ def update_repos( quiet: bool, update_dependencies: bool, clone_mode: str, -): + force_update: bool = True, +) -> None: """ Update repositories by pulling latest changes and installing them. - - Parameters: - - selected_repos: List of selected repositories. - - repositories_base_dir: Base directory for repositories. - - bin_dir: Directory for symbolic links. - - all_repos: All repository configurations. - - no_verification: Whether to skip verification. - - system_update: Whether to run system update. - - preview: If True, only show commands without executing. - - quiet: If True, suppress messages. - - update_dependencies: Whether to update dependent repositories. - - clone_mode: Method to clone repositories (ssh or https). """ pull_with_verification( selected_repos, @@ -50,18 +42,17 @@ def update_repos( quiet, clone_mode, update_dependencies, + force_update=force_update, ) if system_update: from pkgmgr.core.command.run import run_command - # Nix: upgrade all profile entries (if Nix is available) if shutil.which("nix") is not None: try: run_command("nix profile upgrade '.*'", preview=preview) except SystemExit as e: print(f"[Warning] 'nix profile upgrade' failed: {e}") - # Arch / AUR system update run_command("sudo -u aur_builder yay -Syu --noconfirm", preview=preview) run_command("sudo pacman -Syyu --noconfirm", preview=preview) diff --git a/src/pkgmgr/cli/commands/repos.py b/src/pkgmgr/cli/commands/repos.py index 74b8478..700e52f 100644 --- a/src/pkgmgr/cli/commands/repos.py +++ b/src/pkgmgr/cli/commands/repos.py @@ -8,13 +8,13 @@ from typing import Any, Dict, List from pkgmgr.cli.context import CLIContext from pkgmgr.actions.install import install_repos +from pkgmgr.actions.repository.update import update_repos from pkgmgr.actions.repository.deinstall import deinstall_repos from pkgmgr.actions.repository.delete import delete_repos -from pkgmgr.actions.repository.update import update_repos from pkgmgr.actions.repository.status import status_repos from pkgmgr.actions.repository.list import list_repositories -from pkgmgr.core.command.run import run_command from pkgmgr.actions.repository.create import create_repo +from pkgmgr.core.command.run import run_command from pkgmgr.core.repository.dir import get_repo_dir Repository = Dict[str, Any] @@ -51,7 +51,7 @@ def handle_repos_command( selected: List[Repository], ) -> None: """ - Handle core repository commands (install/update/deinstall/delete/.../list). + Handle core repository commands (install/update/deinstall/delete/status/list/path/shell/create). """ # ------------------------------------------------------------ @@ -68,6 +68,7 @@ def handle_repos_command( args.quiet, args.clone_mode, args.dependencies, + force_update=getattr(args, "update", False), ) return @@ -81,11 +82,12 @@ def handle_repos_command( ctx.binaries_dir, ctx.all_repositories, args.no_verification, - args.system, + args.system_update, args.preview, args.quiet, args.dependencies, args.clone_mode, + force_update=True, ) return @@ -146,9 +148,7 @@ def handle_repos_command( f"{repository.get('account', '?')}/" f"{repository.get('repository', '?')}" ) - print( - f"[WARN] Could not resolve directory for {ident}: {exc}" - ) + print(f"[WARN] Could not resolve directory for {ident}: {exc}") continue print(repo_dir) diff --git a/src/pkgmgr/cli/parser/install_update.py b/src/pkgmgr/cli/parser/install_update.py index af78783..e31202f 100644 --- a/src/pkgmgr/cli/parser/install_update.py +++ b/src/pkgmgr/cli/parser/install_update.py @@ -1,11 +1,12 @@ #!/usr/bin/env python3 # -*- coding: utf-8 -*- -from __future__ import annotations - import argparse -from .common import add_install_update_arguments, add_identifier_arguments +from pkgmgr.cli.parser.common import ( + add_install_update_arguments, + add_identifier_arguments, +) def add_install_update_subparsers( @@ -14,11 +15,17 @@ def add_install_update_subparsers( """ Register install / update / deinstall / delete commands. """ + install_parser = subparsers.add_parser( "install", help="Setup repository/repositories alias links to executables", ) add_install_update_arguments(install_parser) + install_parser.add_argument( + "--update", + action="store_true", + help="Force re-run installers (upgrade/refresh) even if the CLI layer is already loaded", + ) update_parser = subparsers.add_parser( "update", @@ -30,6 +37,7 @@ def add_install_update_subparsers( action="store_true", help="Include system update commands", ) + # KEIN --update hier nötig → update impliziert force_update=True deinstall_parser = subparsers.add_parser( "deinstall", diff --git a/src/pkgmgr/core/command/layer.py b/src/pkgmgr/core/command/layer.py new file mode 100644 index 0000000..df3b33b --- /dev/null +++ b/src/pkgmgr/core/command/layer.py @@ -0,0 +1,30 @@ +# src/pkgmgr/core/command/layer.py +from __future__ import annotations + +from enum import Enum + + +class CliLayer(str, Enum): + """ + CLI layer precedence (lower number = stronger layer). + """ + OS_PACKAGES = "os-packages" + NIX = "nix" + PYTHON = "python" + MAKEFILE = "makefile" + + +_LAYER_PRIORITY: dict[CliLayer, int] = { + CliLayer.OS_PACKAGES: 0, + CliLayer.NIX: 1, + CliLayer.PYTHON: 2, + CliLayer.MAKEFILE: 3, +} + + +def layer_priority(layer: CliLayer) -> int: + """ + Return precedence priority for the given layer. + Lower value means higher priority (stronger layer). + """ + return _LAYER_PRIORITY.get(layer, 999) diff --git a/tests/e2e/_util.py b/tests/e2e/_util.py new file mode 100644 index 0000000..fbf313c --- /dev/null +++ b/tests/e2e/_util.py @@ -0,0 +1,24 @@ +import subprocess + + +def run(cmd, *, cwd=None, env=None, shell=False) -> str: + proc = subprocess.run( + cmd, + cwd=cwd, + env=env, + shell=shell, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + ) + + print("----- BEGIN COMMAND -----") + print(cmd if isinstance(cmd, str) else " ".join(cmd)) + print("----- OUTPUT -----") + print(proc.stdout.rstrip()) + print("----- END COMMAND -----") + + if proc.returncode != 0: + raise AssertionError(proc.stdout) + + return proc.stdout diff --git a/tests/e2e/test_install_makefile_three_times.py b/tests/e2e/test_install_makefile_three_times.py new file mode 100644 index 0000000..70641fc --- /dev/null +++ b/tests/e2e/test_install_makefile_three_times.py @@ -0,0 +1,25 @@ +from tests.e2e._util import run +import tempfile +import unittest +from pathlib import Path + +class TestMakefileThreeTimes(unittest.TestCase): + def test_make_install_three_times(self): + with tempfile.TemporaryDirectory(prefix="makefile-3x-") as tmp: + repo = Path(tmp) + + # Minimal Makefile with install target + (repo / "Makefile").write_text( + "install:\n\t@echo install >> install.log\n" + ) + + for i in range(1, 4): + print(f"\n=== RUN {i}/3 ===") + run(["make", "install"], cwd=repo) + + log = (repo / "install.log").read_text().splitlines() + self.assertEqual( + len(log), + 3, + "make install should have been executed exactly three times", + ) diff --git a/tests/e2e/test_install_pkgmgr_three_times_nix.py b/tests/e2e/test_install_pkgmgr_three_times_nix.py new file mode 100644 index 0000000..40c0ec9 --- /dev/null +++ b/tests/e2e/test_install_pkgmgr_three_times_nix.py @@ -0,0 +1,37 @@ +import os +from tests.e2e._util import run +import tempfile +import unittest +from pathlib import Path + + +class TestPkgmgrInstallThreeTimesNix(unittest.TestCase): + def test_three_times_install_nix(self): + with tempfile.TemporaryDirectory(prefix="pkgmgr-nix-") as tmp: + tmp_path = Path(tmp) + + env = os.environ.copy() + env["HOME"] = tmp + + # Ensure nix is found + env["PATH"] = "/nix/var/nix/profiles/default/bin:" + os.environ.get("PATH", "") + + # IMPORTANT: + # nix run uses git+file:///src internally -> Git will reject /src if it's not a safe.directory. + # Our test sets HOME to a temp dir, so we must provide a temp global gitconfig. + gitconfig = tmp_path / ".gitconfig" + gitconfig.write_text( + "[safe]\n" + "\tdirectory = /src\n" + "\tdirectory = /src/.git\n" + "\tdirectory = *\n" + ) + env["GIT_CONFIG_GLOBAL"] = str(gitconfig) + + for i in range(1, 4): + print(f"\n=== RUN {i}/3 ===") + run( + "nix run .#pkgmgr -- install pkgmgr --update --clone-mode shallow --no-verification", + env=env, + shell=True, + ) diff --git a/tests/e2e/test_install_pkgmgr_three_times_venv.py b/tests/e2e/test_install_pkgmgr_three_times_venv.py new file mode 100644 index 0000000..1e83dec --- /dev/null +++ b/tests/e2e/test_install_pkgmgr_three_times_venv.py @@ -0,0 +1,34 @@ +from tests.e2e._util import run +import tempfile +import unittest +from pathlib import Path +import os + + +class TestPkgmgrInstallThreeTimesVenv(unittest.TestCase): + def test_three_times_install_venv(self): + with tempfile.TemporaryDirectory(prefix="pkgmgr-venv-") as tmp: + home = Path(tmp) + bin_dir = home / ".local" / "bin" + bin_dir.mkdir(parents=True) + + env = os.environ.copy() + env["HOME"] = tmp + + # pkgmgr kommt aus dem Projekt-venv + env["PATH"] = ( + f"{Path.cwd() / '.venv' / 'bin'}:" + f"{bin_dir}:" + + os.environ.get("PATH", "") + ) + + # nix explizit deaktivieren → Python/Venv-Pfad + env["PKGMGR_DISABLE_NIX_FLAKE_INSTALLER"] = "1" + + for i in range(1, 4): + print(f"\n=== RUN {i}/3 ===") + run( + "pkgmgr install pkgmgr --update --clone-mode shallow --no-verification", + env=env, + shell=True, + ) From f2970adbb29655a17ef0132633e83fe2a6a85541 Mon Sep 17 00:00:00 2001 From: Kevin Veen-Birkenbach Date: Sat, 13 Dec 2025 19:49:40 +0100 Subject: [PATCH 2/5] test(e2e): enforce --system-update and isolate update-all integration tests - Require --system-update for update-all integration tests - Run tests with isolated HOME and temporary gitconfig - Allow /src as git safe.directory for nix run - Capture and print combined stdout/stderr on failure - Ensure consistent environment for pkgmgr and nix-run executions --- tests/e2e/test_update_all.py | 127 ++++++++++++++++++++++------------- 1 file changed, 80 insertions(+), 47 deletions(-) diff --git a/tests/e2e/test_update_all.py b/tests/e2e/test_update_all.py index 105526f..1332ec9 100644 --- a/tests/e2e/test_update_all.py +++ b/tests/e2e/test_update_all.py @@ -8,13 +8,17 @@ This test is intended to be run inside the Docker container where: - and it is safe to perform real git operations. It passes if BOTH commands complete successfully (in separate tests): - 1) pkgmgr update --all --clone-mode https --no-verification - 2) nix run .#pkgmgr -- update --all --clone-mode https --no-verification + 1) pkgmgr update --all --clone-mode https --no-verification --system-update + 2) nix run .#pkgmgr -- update --all --clone-mode https --no-verification --system-update """ +from __future__ import annotations + import os import subprocess +import tempfile import unittest +from pathlib import Path from test_install_pkgmgr_shallow import ( nix_profile_list_debug, @@ -23,69 +27,98 @@ from test_install_pkgmgr_shallow import ( ) -class TestIntegrationUpdateAllHttps(unittest.TestCase): - def _run_cmd(self, cmd: list[str], label: str) -> None: - """ - Run a real CLI command and raise a helpful assertion on failure. - """ - cmd_repr = " ".join(cmd) - env = os.environ.copy() +def _make_temp_gitconfig_with_safe_dirs(home: Path) -> Path: + gitconfig = home / ".gitconfig" + gitconfig.write_text( + "[safe]\n" + "\tdirectory = /src\n" + "\tdirectory = /src/.git\n" + "\tdirectory = *\n" + ) + return gitconfig - try: - print(f"\n[TEST] Running ({label}): {cmd_repr}") - subprocess.run( - cmd, - check=True, - cwd=os.getcwd(), - env=env, - text=True, - ) - except subprocess.CalledProcessError as exc: + +class TestIntegrationUpdateAllHttps(unittest.TestCase): + def _common_env(self, home_dir: str) -> dict[str, str]: + env = os.environ.copy() + env["HOME"] = home_dir + + home = Path(home_dir) + home.mkdir(parents=True, exist_ok=True) + + env["GIT_CONFIG_GLOBAL"] = str(_make_temp_gitconfig_with_safe_dirs(home)) + + # Ensure nix is discoverable if the container has it + env["PATH"] = "/nix/var/nix/profiles/default/bin:" + env.get("PATH", "") + + return env + + def _run_cmd(self, cmd: list[str], label: str, env: dict[str, str]) -> None: + cmd_repr = " ".join(cmd) + print(f"\n[TEST] Running ({label}): {cmd_repr}") + + proc = subprocess.run( + cmd, + check=False, + cwd=os.getcwd(), + env=env, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + ) + + print(proc.stdout.rstrip()) + + if proc.returncode != 0: print(f"\n[TEST] Command failed ({label})") print(f"[TEST] Command : {cmd_repr}") - print(f"[TEST] Exit code: {exc.returncode}") + print(f"[TEST] Exit code: {proc.returncode}") nix_profile_list_debug(f"ON FAILURE ({label})") raise AssertionError( - f"({label}) {cmd_repr!r} failed with exit code {exc.returncode}. " - "Scroll up to see the full pkgmgr/nix output inside the container." - ) from exc + f"({label}) {cmd_repr!r} failed with exit code {proc.returncode}.\n\n" + f"--- output ---\n{proc.stdout}\n" + ) def _common_setup(self) -> None: - # Debug before cleanup nix_profile_list_debug("BEFORE CLEANUP") - - # Cleanup: aggressively try to drop any pkgmgr/profile entries - # (keeps the environment comparable to other integration tests). remove_pkgmgr_from_nix_profile() - - # Debug after cleanup nix_profile_list_debug("AFTER CLEANUP") def test_update_all_repositories_https_pkgmgr(self) -> None: - """ - Run: pkgmgr update --all --clone-mode https --no-verification - """ self._common_setup() - - args = ["update", "--all", "--clone-mode", "https", "--no-verification"] - self._run_cmd(["pkgmgr", *args], label="pkgmgr") - - # After successful update: show `pkgmgr --help` via interactive bash - pkgmgr_help_debug() + with tempfile.TemporaryDirectory(prefix="pkgmgr-updateall-") as tmp: + env = self._common_env(tmp) + args = [ + "update", + "--all", + "--clone-mode", + "https", + "--no-verification", + "--system-update", + ] + self._run_cmd(["pkgmgr", *args], label="pkgmgr", env=env) + pkgmgr_help_debug() def test_update_all_repositories_https_nix_pkgmgr(self) -> None: - """ - Run: nix run .#pkgmgr -- update --all --clone-mode https --no-verification - """ self._common_setup() - - args = ["update", "--all", "--clone-mode", "https", "--no-verification"] - self._run_cmd(["nix", "run", ".#pkgmgr", "--", *args], label="nix run .#pkgmgr") - - # After successful update: show `pkgmgr --help` via interactive bash - pkgmgr_help_debug() + with tempfile.TemporaryDirectory(prefix="pkgmgr-updateall-nix-") as tmp: + env = self._common_env(tmp) + args = [ + "update", + "--all", + "--clone-mode", + "https", + "--no-verification", + "--system-update", + ] + self._run_cmd( + ["nix", "run", ".#pkgmgr", "--", *args], + label="nix run .#pkgmgr", + env=env, + ) + pkgmgr_help_debug() if __name__ == "__main__": From 1e5d6d3eee2baabf8f923bffca67b06f02fbb6d5 Mon Sep 17 00:00:00 2001 From: Kevin Veen-Birkenbach Date: Sat, 13 Dec 2025 19:53:34 +0100 Subject: [PATCH 3/5] test(unit): update NixFlakeInstaller tests for new run_command-based logic - Adapt DummyCtx to include quiet and force_update flags - Replace os.system mocking with run_command/subprocess mocks - Align assertions with new Nix install/upgrade output - Keep coverage for mandatory vs optional output handling https://chatgpt.com/share/693db645-c420-800f-b921-9d5c0356d0ac --- .../install/installers/test_nix_flake.py | 197 ++++++++++-------- 1 file changed, 105 insertions(+), 92 deletions(-) diff --git a/tests/unit/pkgmgr/actions/install/installers/test_nix_flake.py b/tests/unit/pkgmgr/actions/install/installers/test_nix_flake.py index c9299c0..ad891ee 100644 --- a/tests/unit/pkgmgr/actions/install/installers/test_nix_flake.py +++ b/tests/unit/pkgmgr/actions/install/installers/test_nix_flake.py @@ -5,15 +5,18 @@ Unit tests for NixFlakeInstaller using unittest (no pytest). Covers: -- Successful installation (exit_code == 0) +- Successful installation (returncode == 0) - Mandatory failure → SystemExit with correct code - Optional failure (pkgmgr default) → no raise, but warning - supports() behavior incl. PKGMGR_DISABLE_NIX_FLAKE_INSTALLER """ +from __future__ import annotations + import io import os import shutil +import subprocess import tempfile import unittest from contextlib import redirect_stdout @@ -25,10 +28,19 @@ from pkgmgr.actions.install.installers.nix_flake import NixFlakeInstaller class DummyCtx: """Minimal context object to satisfy NixFlakeInstaller.run() / supports().""" - def __init__(self, identifier: str, repo_dir: str, preview: bool = False): + def __init__( + self, + identifier: str, + repo_dir: str, + preview: bool = False, + quiet: bool = False, + force_update: bool = False, + ): self.identifier = identifier self.repo_dir = repo_dir self.preview = preview + self.quiet = quiet + self.force_update = force_update class TestNixFlakeInstaller(unittest.TestCase): @@ -44,161 +56,162 @@ class TestNixFlakeInstaller(unittest.TestCase): os.environ.pop("PKGMGR_DISABLE_NIX_FLAKE_INSTALLER", None) def tearDown(self) -> None: - # Cleanup temporary directory if os.path.isdir(self._tmpdir): shutil.rmtree(self._tmpdir, ignore_errors=True) - def _enable_nix_in_module(self, which_patch): + @staticmethod + def _cp(code: int) -> subprocess.CompletedProcess: + # stdout/stderr are irrelevant here, but keep shape realistic + return subprocess.CompletedProcess(args=["nix"], returncode=code, stdout="", stderr="") + + @staticmethod + def _enable_nix_in_module(which_patch) -> None: """Ensure shutil.which('nix') in nix_flake module returns a path.""" which_patch.return_value = "/usr/bin/nix" - def test_nix_flake_run_success(self): + def test_nix_flake_run_success(self) -> None: """ - When os.system returns a successful exit code, the installer + When run_command returns success (returncode 0), installer should report success and not raise. """ ctx = DummyCtx(identifier="some-lib", repo_dir=self.repo_dir) - installer = NixFlakeInstaller() buf = io.StringIO() - with patch( - "pkgmgr.actions.install.installers.nix_flake.shutil.which" - ) as which_mock, patch( - "pkgmgr.actions.install.installers.nix_flake.os.system" - ) as system_mock, redirect_stdout(buf): + with patch("pkgmgr.actions.install.installers.nix_flake.shutil.which") as which_mock, patch( + "pkgmgr.actions.install.installers.nix_flake.subprocess.run" + ) as subproc_mock, patch( + "pkgmgr.actions.install.installers.nix_flake.run_command" + ) as run_cmd_mock, redirect_stdout(buf): self._enable_nix_in_module(which_mock) - # Simulate os.system returning success (exit code 0) - system_mock.return_value = 0 + # For profile list JSON (used only on failure paths, but keep deterministic) + subproc_mock.return_value = subprocess.CompletedProcess( + args=["nix", "profile", "list", "--json"], + returncode=0, + stdout='{"elements": []}', + stderr="", + ) + + # Install succeeds + run_cmd_mock.return_value = self._cp(0) - # Sanity: supports() must be True self.assertTrue(installer.supports(ctx)) - installer.run(ctx) out = buf.getvalue() - self.assertIn("[INFO] Running: nix profile install", out) - self.assertIn("Nix flake output 'default' successfully installed.", out) + self.assertIn("[nix] install: nix profile install", out) + self.assertIn("[nix] output 'default' successfully installed.", out) - # Ensure the nix command was actually invoked - system_mock.assert_called_with( - f"nix profile install {self.repo_dir}#default" + run_cmd_mock.assert_called_with( + f"nix profile install {self.repo_dir}#default", + cwd=self.repo_dir, + preview=False, + allow_failure=True, ) - def test_nix_flake_run_mandatory_failure_raises(self): + def test_nix_flake_run_mandatory_failure_raises(self) -> None: """ - For a generic repository (identifier not pkgmgr/package-manager), - `default` is mandatory and a non-zero exit code should raise SystemExit - with the real exit code (e.g. 1, not 256). + For a generic repository, 'default' is mandatory. + A non-zero return code must raise SystemExit with that code. """ ctx = DummyCtx(identifier="some-lib", repo_dir=self.repo_dir) installer = NixFlakeInstaller() buf = io.StringIO() - with patch( - "pkgmgr.actions.install.installers.nix_flake.shutil.which" - ) as which_mock, patch( - "pkgmgr.actions.install.installers.nix_flake.os.system" - ) as system_mock, redirect_stdout(buf): + with patch("pkgmgr.actions.install.installers.nix_flake.shutil.which") as which_mock, patch( + "pkgmgr.actions.install.installers.nix_flake.subprocess.run" + ) as subproc_mock, patch( + "pkgmgr.actions.install.installers.nix_flake.run_command" + ) as run_cmd_mock, redirect_stdout(buf): self._enable_nix_in_module(which_mock) - # Simulate os.system returning encoded status for exit code 1 - # os.system encodes exit code as (exit_code << 8) - system_mock.return_value = 1 << 8 + # No indices available (empty list) + subproc_mock.return_value = subprocess.CompletedProcess( + args=["nix", "profile", "list", "--json"], + returncode=0, + stdout='{"elements": []}', + stderr="", + ) + + # First install fails, retry fails -> should raise SystemExit(1) + run_cmd_mock.side_effect = [self._cp(1), self._cp(1)] self.assertTrue(installer.supports(ctx)) - with self.assertRaises(SystemExit) as cm: installer.run(ctx) - # The real exit code should be 1 (not 256) self.assertEqual(cm.exception.code, 1) - out = buf.getvalue() - self.assertIn("[INFO] Running: nix profile install", out) - self.assertIn("[Error] Failed to install Nix flake output 'default'", out) - self.assertIn("[Error] Command exited with code 1", out) + self.assertIn("[nix] install: nix profile install", out) + self.assertIn("[ERROR] Failed to install Nix flake output 'default' (exit 1)", out) - def test_nix_flake_run_optional_failure_does_not_raise(self): + def test_nix_flake_run_optional_failure_does_not_raise(self) -> None: """ - For the package-manager repository, the 'default' output is optional. - Failure to install it must not raise, but should log a warning instead. + For pkgmgr/package-manager repositories: + - 'pkgmgr' output is mandatory + - 'default' output is optional + Failure of optional output must not raise. """ ctx = DummyCtx(identifier="pkgmgr", repo_dir=self.repo_dir) installer = NixFlakeInstaller() - calls = [] - - def fake_system(cmd: str) -> int: - calls.append(cmd) - # First call (pkgmgr) → success - if len(calls) == 1: - return 0 - # Second call (default) → failure (exit code 1 encoded) - return 1 << 8 - buf = io.StringIO() - with patch( - "pkgmgr.actions.install.installers.nix_flake.shutil.which" - ) as which_mock, patch( - "pkgmgr.actions.install.installers.nix_flake.os.system", - side_effect=fake_system, - ), redirect_stdout(buf): + with patch("pkgmgr.actions.install.installers.nix_flake.shutil.which") as which_mock, patch( + "pkgmgr.actions.install.installers.nix_flake.subprocess.run" + ) as subproc_mock, patch( + "pkgmgr.actions.install.installers.nix_flake.run_command" + ) as run_cmd_mock, redirect_stdout(buf): self._enable_nix_in_module(which_mock) + # No indices available (empty list) + subproc_mock.return_value = subprocess.CompletedProcess( + args=["nix", "profile", "list", "--json"], + returncode=0, + stdout='{"elements": []}', + stderr="", + ) + + # pkgmgr install ok; default fails twice (initial + retry) + run_cmd_mock.side_effect = [self._cp(0), self._cp(1), self._cp(1)] + self.assertTrue(installer.supports(ctx)) - # Optional failure must NOT raise + # Must NOT raise despite optional failure installer.run(ctx) out = buf.getvalue() - # Both outputs should have been mentioned - self.assertIn( - "attempting to install profile outputs: pkgmgr, default", out - ) + # Should announce both outputs + self.assertIn("ensuring outputs: pkgmgr, default", out) - # First output ("pkgmgr") succeeded - self.assertIn( - "Nix flake output 'pkgmgr' successfully installed.", out - ) + # First output ok + self.assertIn("[nix] output 'pkgmgr' successfully installed.", out) - # Second output ("default") failed but did not raise - self.assertIn( - "[Error] Failed to install Nix flake output 'default'", out - ) - self.assertIn("[Error] Command exited with code 1", out) - self.assertIn( - "Continuing despite failure to install optional output 'default'.", - out, - ) + # Second output failed but no raise + self.assertIn("[ERROR] Failed to install Nix flake output 'default' (exit 1)", out) + self.assertIn("[WARNING] Continuing despite failure of optional output 'default'.", out) - # Ensure we actually called os.system twice (pkgmgr and default) - self.assertEqual(len(calls), 2) - self.assertIn( - f"nix profile install {self.repo_dir}#pkgmgr", - calls[0], - ) - self.assertIn( - f"nix profile install {self.repo_dir}#default", - calls[1], - ) + # Verify run_command was called for both outputs (default twice due to retry) + expected_calls = [ + (f"nix profile install {self.repo_dir}#pkgmgr",), + (f"nix profile install {self.repo_dir}#default",), + (f"nix profile install {self.repo_dir}#default",), + ] + actual_cmds = [c.args[0] for c in run_cmd_mock.call_args_list] + self.assertEqual(actual_cmds, [e[0] for e in expected_calls]) - def test_nix_flake_supports_respects_disable_env(self): + def test_nix_flake_supports_respects_disable_env(self) -> None: """ PKGMGR_DISABLE_NIX_FLAKE_INSTALLER=1 must disable the installer, even if flake.nix exists and nix is available. """ - ctx = DummyCtx(identifier="pkgmgr", repo_dir=self.repo_dir) + ctx = DummyCtx(identifier="pkgmgr", repo_dir=self.repo_dir, quiet=False) installer = NixFlakeInstaller() - with patch( - "pkgmgr.actions.install.installers.nix_flake.shutil.which" - ) as which_mock: + with patch("pkgmgr.actions.install.installers.nix_flake.shutil.which") as which_mock: self._enable_nix_in_module(which_mock) os.environ["PKGMGR_DISABLE_NIX_FLAKE_INSTALLER"] = "1" - self.assertFalse(installer.supports(ctx)) From 7f06447bbd1d18feb8e156368d976705b53c5a2b Mon Sep 17 00:00:00 2001 From: Kevin Veen-Birkenbach Date: Sat, 13 Dec 2025 20:02:48 +0100 Subject: [PATCH 4/5] feat(cli): add --system-update flag to update command - Register --system-update for `pkgmgr update` - Expose args.system_update for update workflow - Align CLI with update_repos and E2E tests https://chatgpt.com/share/693db645-c420-800f-b921-9d5c0356d0ac --- src/pkgmgr/cli/parser/install_update.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/pkgmgr/cli/parser/install_update.py b/src/pkgmgr/cli/parser/install_update.py index e31202f..470c74c 100644 --- a/src/pkgmgr/cli/parser/install_update.py +++ b/src/pkgmgr/cli/parser/install_update.py @@ -33,11 +33,11 @@ def add_install_update_subparsers( ) add_install_update_arguments(update_parser) update_parser.add_argument( - "--system", + "--system-update", + dest="system_update", action="store_true", help="Include system update commands", ) - # KEIN --update hier nötig → update impliziert force_update=True deinstall_parser = subparsers.add_parser( "deinstall", From f5131969117f251f11055e6195bdddf7d966d2f8 Mon Sep 17 00:00:00 2001 From: Kevin Veen-Birkenbach Date: Sat, 13 Dec 2025 20:08:30 +0100 Subject: [PATCH 5/5] Used correct tabulation --- src/pkgmgr/cli/parser/install_update.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/pkgmgr/cli/parser/install_update.py b/src/pkgmgr/cli/parser/install_update.py index 470c74c..6c0d620 100644 --- a/src/pkgmgr/cli/parser/install_update.py +++ b/src/pkgmgr/cli/parser/install_update.py @@ -33,11 +33,12 @@ def add_install_update_subparsers( ) add_install_update_arguments(update_parser) update_parser.add_argument( - "--system-update", + "--system-update", dest="system_update", action="store_true", help="Include system update commands", ) + # No --update here: update implies force_update=True deinstall_parser = subparsers.add_parser( "deinstall",