kevinveenbirkenbach/pkgmgr
1
0
Fork 0
Code Issues Pull Requests Actions 24 Packages Projects Releases Wiki Activity

feat(install): add --update to re-run active-layer installers and improve Nix refresh logic
Some checks failed
CI / test-unit (push) Has been cancelled
Details
CI / test-integration (push) Has been cancelled
Details
CI / test-env-virtual (push) Has been cancelled
Details
CI / test-env-nix (push) Has been cancelled
Details
CI / test-e2e (push) Has been cancelled
Details
CI / test-virgin-user (push) Has been cancelled
Details
CI / test-virgin-root (push) Has been cancelled
Details
CI / codesniffer-shellcheck (push) Has been cancelled
Details
CI / codesniffer-ruff (push) Has been cancelled
Details

Browse Source 
* 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
This commit is contained in:
Kevin Veen-Birkenbach
2025-12-13 19:30:06 +01:00
parent 0bc7a3ecc0
commit 7f262c6557
16 changed files with 424 additions and 397 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
scripts/test/test-e2e.sh
View File

@@ -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

20
src/pkgmgr/actions/install/__init__.py
View File

@@ -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)

4
src/pkgmgr/actions/install/context.py
View File

@@ -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

59
src/pkgmgr/actions/install/installers/makefile.py
View File

@@ -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.")

321
src/pkgmgr/actions/install/installers/nix_flake.py
View File

@@ -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.
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,
)
def _profile_list_json(self, ctx: "RepoContext") -> dict:
"""
# Reuse supports() to keep logic in one place.
if not self.supports(ctx): # type: ignore[arg-type]
Read current Nix profile entries as JSON (best-effort).
NOTE: Nix versions differ:
- Newer: {"elements": [ { "index": 0, "attrPath": "...", ... }, ... ]}
- Older: {"elements": [ "nixpkgs#hello", ... ]} (strings)
We return {} on failure or in preview mode.
"""
if ctx.preview:
return {}
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
idx = el.get("index")
if idx is None:
continue
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
outputs = self._profile_outputs(ctx) # list of (name, allow_failure)
if not ctx.quiet:
print(
"Nix flake detected in "
f"{ctx.identifier}, attempting to install profile outputs: "
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)
)
# Handle the "already installed" case up-front for the shared profile.
self._ensure_old_profile_removed(ctx) # type: ignore[arg-type]
for output, allow_failure in outputs:
cmd = f"nix profile install {ctx.repo_dir}#{output}"
print(f"[INFO] Running: {cmd}")
ret = os.system(cmd)
# Extract real exit code from os.system() result
if os.WIFEXITED(ret):
exit_code = os.WEXITSTATUS(ret)
if ctx.force_update:
self._force_upgrade_output(ctx, output, allow_failure)
else:
# abnormal termination (signal etc.) – keep raw value
exit_code = ret
if exit_code == 0:
print(f"Nix flake output '{output}' successfully installed.")
continue
print(f"[Error] Failed to install Nix flake output '{output}'")
print(f"[Error] Command exited with code {exit_code}")
if not allow_failure:
raise SystemExit(exit_code)
print(
"[Warning] Continuing despite failure to install "
f"optional output '{output}'."
)
self._install_only(ctx, output, allow_failure)

106
src/pkgmgr/actions/install/installers/python.py
View File

@@ -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/<repo>/.
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/<identifier>/ 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/<repo>/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}.")

79
src/pkgmgr/actions/install/pipeline.py
View File

@@ -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:
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:

23
src/pkgmgr/actions/repository/pull.py
View File

@@ -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 "

23
src/pkgmgr/actions/repository/update.py
View File

@@ -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)

14
src/pkgmgr/cli/commands/repos.py
View File

@@ -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)

14
src/pkgmgr/cli/parser/install_update.py
View File

@@ -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",

30
src/pkgmgr/core/command/layer.py Normal file
View File

@@ -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)

24
tests/e2e/_util.py Normal file
View File

@@ -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

25
tests/e2e/test_install_makefile_three_times.py Normal file
View File

@@ -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",
)

37
tests/e2e/test_install_pkgmgr_three_times_nix.py Normal file
View File

@@ -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,
)

34
tests/e2e/test_install_pkgmgr_three_times_venv.py Normal file
View File

@@ -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,
)