Compare commits

...

19 Commits

Author SHA1 Message Date
Kevin Veen-Birkenbach
fcf9d4b59b **Aur builder: add retry logic for yay clone to recover from GitHub 504 errors**
Some checks failed
Mark stable commit / test-unit (push) Has been cancelled
Mark stable commit / test-integration (push) Has been cancelled
Mark stable commit / test-container (push) Has been cancelled
Mark stable commit / test-e2e (push) Has been cancelled
Mark stable commit / test-virgin-user (push) Has been cancelled
Mark stable commit / test-virgin-root (push) Has been cancelled
Mark stable commit / mark-stable (push) Has been cancelled
Implemented a robust retry mechanism for cloning the yay AUR helper during Arch dependency installation.
The new logic retries the git clone operation for up to 5 minutes with a 20-second pause between attempts, allowing the build to proceed even when GitHub intermittently returns HTTP 504 errors.

This improves the stability of Arch container builds, especially under network pressure or transient upstream outages.
The yay build process now only starts once the clone step completes successfully.

https://chatgpt.com/share/693b102b-fdb0-800f-9f2e-d4840f14d329
2025-12-11 19:40:25 +01:00
Kevin Veen-Birkenbach
b483dbfaad **fix(init-nix): ensure nixbld group/users exist on Ubuntu root-without-systemd installs**
Some checks failed
Mark stable commit / test-unit (push) Has been cancelled
Mark stable commit / test-integration (push) Has been cancelled
Mark stable commit / test-container (push) Has been cancelled
Mark stable commit / test-e2e (push) Has been cancelled
Mark stable commit / test-virgin-user (push) Has been cancelled
Mark stable commit / test-virgin-root (push) Has been cancelled
Mark stable commit / mark-stable (push) Has been cancelled
Implement `ensure_nix_build_group()` and use it in all code paths where Nix is installed as root.
This resolves Nix installation failures on Ubuntu containers (root, no systemd) where the installer aborts with:

```
error: the group 'nixbld' specified in 'build-users-group' does not exist
```

The fix standardizes creation of the `nixbld` group and `nixbld1..10` build users across:

* container root mode
* systemd host daemon installs
* root-on-host without systemd (Debian/Ubuntu CI case)

This makes Nix initialization deterministic across all test distros and fixes failing Ubuntu E2E runs.

https://chatgpt.com/share/693b0e1a-e5d4-800f-8a89-7d91108b0368
2025-12-11 19:31:25 +01:00
Kevin Veen-Birkenbach
9630917570 **refactor(nix-flake): replace run_command wrapper with direct os.system execution and extend test coverage**
Some checks failed
Mark stable commit / test-unit (push) Has been cancelled
Mark stable commit / test-integration (push) Has been cancelled
Mark stable commit / test-container (push) Has been cancelled
Mark stable commit / test-e2e (push) Has been cancelled
Mark stable commit / test-virgin-user (push) Has been cancelled
Mark stable commit / test-virgin-root (push) Has been cancelled
Mark stable commit / mark-stable (push) Has been cancelled
This commit removes the `run_command`-based execution model for Nix flake
installations and replaces it with a direct `os.system` invocation.
This ensures that *all* Nix diagnostics (stdout/stderr) are fully visible and
no longer suppressed by wrapper logic.

Key changes:

* Directly run `nix profile install` via `os.system` for full error output
* Correctly decode real exit codes via `os.WIFEXITED` / `os.WEXITSTATUS`
* Preserve mandatory/optional behavior for flake outputs
* Update unit tests to the new execution model using `unittest`
* Add complete coverage for:

  * successful installs
  * mandatory failures → raise SystemExit(code)
  * optional failures → warn and continue
  * environment-based disabling via `PKGMGR_DISABLE_NIX_FLAKE_INSTALLER`
* Remove obsolete mocks and legacy test logic that assumed `run_command`

Overall, this improves transparency, debuggability, and correctness of the
Nix flake installer while maintaining full backward compatibility at the
interface level.

https://chatgpt.com/share/693b0a20-99f4-800f-b789-b00a50413612
2025-12-11 19:14:25 +01:00
Kevin Veen-Birkenbach
6a4432dd04 Added required sudo to debian
Some checks failed
Mark stable commit / test-unit (push) Has been cancelled
Mark stable commit / test-integration (push) Has been cancelled
Mark stable commit / test-container (push) Has been cancelled
Mark stable commit / test-e2e (push) Has been cancelled
Mark stable commit / test-virgin-user (push) Has been cancelled
Mark stable commit / test-virgin-root (push) Has been cancelled
Mark stable commit / mark-stable (push) Has been cancelled
2025-12-11 18:42:33 +01:00
Kevin Veen-Birkenbach
cfb91d825a Release version 0.10.1
Some checks failed
Mark stable commit / test-unit (push) Has been cancelled
Mark stable commit / test-integration (push) Has been cancelled
Mark stable commit / test-container (push) Has been cancelled
Mark stable commit / test-e2e (push) Has been cancelled
Mark stable commit / test-virgin-user (push) Has been cancelled
Mark stable commit / test-virgin-root (push) Has been cancelled
Mark stable commit / mark-stable (push) Has been cancelled
2025-12-11 18:38:15 +01:00
Kevin Veen-Birkenbach
a3b21f23fc pkgmgr-wrapper: improve Nix detection and auto-initialization
Some checks failed
Mark stable commit / test-unit (push) Has been cancelled
Mark stable commit / test-integration (push) Has been cancelled
Mark stable commit / test-container (push) Has been cancelled
Mark stable commit / test-e2e (push) Has been cancelled
Mark stable commit / test-virgin-user (push) Has been cancelled
Mark stable commit / test-virgin-root (push) Has been cancelled
Mark stable commit / mark-stable (push) Has been cancelled
- Extend PATH probing to include /home/nix/.nix-profile/bin/nix (container mode).
- Automatically invoke init-nix.sh when nix is not found before first run.
- Ensure pkgmgr always attempts a one-time Nix initialization instead of failing prematurely.
- Improve error message to clarify that nix was still missing *after* initialization attempt.
- Keep existing flake-based execution path unchanged (exec nix run …).

This makes the wrapper fully reliable across Debian/Ubuntu package installs,
fresh containers, and minimal systems where Nix is not yet initialized.

https://chatgpt.com/share/693b005d-b250-800f-8830-ab71685f51b3
2025-12-11 18:33:02 +01:00
Kevin Veen-Birkenbach
e49dd85200 Release version 0.10.0
Some checks failed
Mark stable commit / test-unit (push) Has been cancelled
Mark stable commit / test-integration (push) Has been cancelled
Mark stable commit / test-container (push) Has been cancelled
Mark stable commit / test-e2e (push) Has been cancelled
Mark stable commit / test-virgin-user (push) Has been cancelled
Mark stable commit / test-virgin-root (push) Has been cancelled
Mark stable commit / mark-stable (push) Has been cancelled
2025-12-11 18:17:21 +01:00
Kevin Veen-Birkenbach
c9dec5ecd6 Merge branch 'feature/mirror'
Some checks failed
Mark stable commit / test-unit (push) Has been cancelled
Mark stable commit / test-integration (push) Has been cancelled
Mark stable commit / test-container (push) Has been cancelled
Mark stable commit / test-e2e (push) Has been cancelled
Mark stable commit / test-virgin-user (push) Has been cancelled
Mark stable commit / test-virgin-root (push) Has been cancelled
Mark stable commit / mark-stable (push) Has been cancelled
2025-12-11 17:50:53 +01:00
Kevin Veen-Birkenbach
f3c5460e48 feat(mirror): support SSH MIRRORS, multi-push origin and remote probe
Some checks failed
CI / test-unit (push) Has been cancelled
CI / test-integration (push) Has been cancelled
CI / test-container (push) Has been cancelled
CI / test-e2e (push) Has been cancelled
CI / test-virgin-user (push) Has been cancelled
CI / test-virgin-root (push) Has been cancelled
- Switch MIRRORS to SSH-based URLs including custom ports/domains
  (GitHub, git.veen.world, code.cymais.cloud)
- Extend mirror IO:
  - load_config_mirrors filters empty values
  - read_mirrors_file now supports:
    * "name url" lines
    * "url" lines with auto-generated names from URL host (host[:port])
  - write_mirrors_file prints full preview content
- Enhance git_remote:
  - determine_primary_remote_url used for origin bootstrap
  - ensure_origin_remote keeps existing origin URL and
    adds all mirror URLs as additional push URLs
  - add is_remote_reachable() helper based on `git ls-remote --exit-code`
- Implement non-destructive remote mirror checks in setup_cmd:
  - `_probe_mirror()` wraps `git ls-remote` and returns (ok, message)
  - `pkgmgr mirror setup --remote` now probes each mirror URL and
    prints [OK]/[WARN] with details instead of placeholder text
- Add unit tests for mirror actions:
  - test_git_remote: default SSH URL building and primary URL selection
  - test_io: config + MIRRORS parsing including auto-named URL-only entries
  - test_setup_cmd: probe_mirror success/failure handling

https://chatgpt.com/share/693adee0-aa3c-800f-b72a-98473fdaf760
2025-12-11 17:49:31 +01:00
Kevin Veen-Birkenbach
39b16b87a8 CI: Add debugging instrumentation to identify container build/run anomalies
Some checks failed
CI / test-unit (push) Has been cancelled
CI / test-integration (push) Has been cancelled
CI / test-container (push) Has been cancelled
CI / test-e2e (push) Has been cancelled
CI / test-virgin-user (push) Has been cancelled
CI / test-virgin-root (push) Has been cancelled
- Added `git rev-parse HEAD` to test-container workflow to confirm the exact
  commit SHA used during CI runs.
- Updated Dockerfile to print BASE_IMAGE and OS release information during
  build for better reproducibility diagnostics.
- Extended test-container script to dump the first 40 lines of
  `docker image inspect` output, allowing verification of the image ID,
  creation time, and applied build args.

These additions help trace discrepancies between local builds and GitHub
Actions, ensuring we can detect mismatches in commit SHA, base image,
or container metadata.

https://chatgpt.com/share/693ae07a-8c58-800f-88e6-254cdb00b676
2025-12-11 17:27:57 +01:00
Kevin Veen-Birkenbach
26c9d79814 Added mirrors 2025-12-11 16:47:23 +01:00
Kevin Veen-Birkenbach
2776d18a42 Implemented arch support 2025-12-11 16:31:00 +01:00
Kevin Veen-Birkenbach
7057ccfb95 CI: Always rebuild test images with --no-cache before container and E2E tests
This ensures that GitHub Actions never reuses outdated Docker layers and that
each test run starts from a fully clean environment. The workflows for
test-container and test-e2e now explicitly invoke:

    distro="${{ matrix.distro }}" make build-no-cache

before executing the actual tests.
This aligns the CI behaviour with local testing, eliminates hidden caching
differences, and guarantees deterministic test results across all distros.

https://chatgpt.com/share/693ae07a-8c58-800f-88e6-254cdb00b676
2025-12-11 16:17:10 +01:00
Kevin Veen-Birkenbach
1807949c6f Add mirror management commands and refactor CLI parser into modules
- Implement new mirror actions:
  - list_mirrors: show mirrors from config, MIRRORS file, or merged view
  - diff_mirrors: compare config mirrors with MIRRORS file (ONLY IN CONFIG,
    ONLY IN FILE, URL MISMATCH, OK)
  - merge_mirrors: merge mirrors between config and MIRRORS file in both
    directions, with preview mode and user config writing via save_user_config
  - setup_mirrors: prepare local Git remotes (ensure origin) and print
    provider-URL suggestions for remote repositories
- Introduce mirror utilities:
  - RepoMirrorContext with resolved_mirrors (config + file, file wins)
  - load_config_mirrors supporting dict and list-of-dicts shapes
  - read/write MIRRORS file with simple "name url" format and preview mode
  - helper for building default SSH URLs from provider/account/repository
- Wire mirror commands into CLI:
  - Add handle_mirror_command and integrate "mirror" into dispatch
  - Add dedicated CLI parser modules under pkgmgr.cli.parser:
    * common, install_update, config_cmd, navigation_cmd,
      branch_cmd, release_cmd, version_cmd, changelog_cmd,
      list_cmd, make_cmd, mirror_cmd
  - Replace old flat cli/parser.py with modular parser package and
    SortedSubParsersAction in common.py
- Update TODO.md to mark MIRROR as implemented
- Add E2E tests for mirror commands:
  - test_mirror_help
  - test_mirror_list_preview_all
  - test_mirror_diff_preview_all
  - test_mirror_merge_config_to_file_preview_all
  - test_mirror_setup_preview_all

https://chatgpt.com/share/693adee0-aa3c-800f-b72a-98473fdaf760
2025-12-11 16:10:19 +01:00
Kevin Veen-Birkenbach
d611720b8f Solved bug when volumes don't exist 2025-12-11 15:46:45 +01:00
Kevin Veen-Birkenbach
bf871650a8 Added purge option to makefile 2025-12-11 15:29:51 +01:00
Kevin Veen-Birkenbach
5ca1adda7b Refactor CI distro handling and container build scripts
- Introduce a GitHub Actions matrix for `test-container` and `test-e2e`
  to run against arch, debian, ubuntu, fedora, and centos
- Run unit and integration tests only in the Arch container by passing
  `distro="arch"` via make in the corresponding workflows
- Replace the global DISTROS loop with a single `distro` variable in
  the Makefile, defaulting to `arch`, and export it for all scripts
- Update build scripts (build-image, build-image-no-cache, build-image-missing)
  to build images for the selected distro only
- Simplify test-container script to validate a single distro image using
  the `distro` environment variable
- Simplify E2E, unit, and integration test scripts to run against a
  single distro container instead of iterating over all distros

https://chatgpt.com/share/693acbba-9e30-800f-94fb-fea4489e9078
2025-12-11 14:48:36 +01:00
Kevin Veen-Birkenbach
acb18adf76 test: restore Dockerfile ENTRYPOINT for all test runs (fix Nix TLS on CentOS)
All test scripts (unit, integration, e2e) previously overwrote the Docker
ENTRYPOINT by using `--entrypoint bash`, which bypassed the container’s
startup logic in `docker-entry.sh`.

`docker-entry.sh` performs essential initialization steps such as:

- CA bundle auto-detection (NIX_SSL_CERT_FILE, SSL_CERT_FILE, etc.)
- Nix environment setup
- PATH adjustments and distro logging

By removing the explicit `--entrypoint bash` and invoking:

  bash -lc '...'

directly as the container command, the Dockerfile’s ENTRYPOINT is restored
and runs as intended before executing the test logic.

This fixes TLS issues in CentOS E2E runs where Nix was unable to fetch
flake inputs due to missing CA configuration.

https://chatgpt.com/share/693ac1f3-fb7c-800f-9e5c-b40c351a9f04
2025-12-11 14:06:39 +01:00
Kevin Veen-Birkenbach
c18490f5d3 deb: remove hard dependency on distro-provided Nix
The Debian Nix package causes flake builds to fail inside the test and
container environment due to sandboxing and patched Nix behavior.

To ensure consistent behaviour across all distributions and align
container logic with production logic, pkgmgr now relies on its own
`init-nix.sh` bootstrap script instead of the distro’s `nix` package.

Dropping `Depends: nix` guarantees that both Debian containers and real
Debian systems install and initialize Nix via the upstream installer,
matching the behaviour on Arch, Fedora, and Ubuntu.

https://chatgpt.com/share/693ab9bf-e6ac-800f-83ba-a4abd1bfe407
2025-12-11 13:31:56 +01:00
56 changed files with 2625 additions and 815 deletions

View File

@@ -7,13 +7,22 @@ jobs:
test-container:
runs-on: ubuntu-latest
timeout-minutes: 30
strategy:
fail-fast: false
matrix:
distro: [arch, debian, ubuntu, fedora, centos]
steps:
- name: Checkout repository
uses: actions/checkout@v4
- name: Show commit SHA
run: git rev-parse HEAD
- name: Show Docker version
run: docker version
- name: Run container tests
run: make test-container
- name: Run container tests (${{ matrix.distro }})
run: |
set -euo pipefail
distro="${{ matrix.distro }}" make test-container

View File

@@ -6,7 +6,11 @@ on:
jobs:
test-e2e:
runs-on: ubuntu-latest
timeout-minutes: 60 # E2E + all distros can be heavier
timeout-minutes: 60 # E2E can be heavier
strategy:
fail-fast: false
matrix:
distro: [arch, debian, ubuntu, fedora, centos]
steps:
- name: Checkout repository
@@ -15,5 +19,7 @@ jobs:
- name: Show Docker version
run: docker version
- name: Run E2E tests via make (all distros)
run: make test-e2e
- name: Run E2E tests via make (${{ matrix.distro }})
run: |
set -euo pipefail
distro="${{ matrix.distro }}" make test-e2e

View File

@@ -16,4 +16,4 @@ jobs:
run: docker version
- name: Run integration tests via make (Arch container)
run: make test-integration DISTROS="arch"
run: make test-integration distro="arch"

View File

@@ -16,4 +16,4 @@ jobs:
run: docker version
- name: Run unit tests via make (Arch container)
run: make test-unit DISTROS="arch"
run: make test-unit distro="arch"

View File

@@ -1,3 +1,50 @@
## [0.10.1] - 2025-12-11
* Fixed Debian\Ubuntu to pass container e2e tests
## [0.10.0] - 2025-12-11
* **Changes since v0.9.1**
**Mirror System**
* Added SSH mirror support including multi-push and remote probing
* Introduced mirror management commands and refactored the CLI parser into modules
**CI/CD**
* Migrated to reusable workflows with improved debugging instrumentation
* Made stable-tag automation reliable for workflow_run events and permissions
* Ensured deterministic test results by rebuilding all test containers with no-cache
**E2E and Container Tests**
* Fixed Git safe.directory handling across all containers
* Restored Dockerfile ENTRYPOINT to resolve Nix TLS issues
* Fixed missing volume errors and hardened the E2E runner
* Added full Nix flake E2E test matrix across all distro containers
* Disabled Nix sandboxing for cross-distro builds where required
**Nix and Python Environment**
* Unified Nix Python environment and introduced lazy CLI imports
* Ensured PyYAML availability and improved Python 3.13 compatibility
* Refactored flake.nix to remove side effects and rely on generic python3
**Packaging**
* Removed Debians hard dependency on Nix
* Restructured packaging layout and refined build paths
* Excluded assets from Arch PKGBUILD rsync
* Cleaned up obsolete ignore files
**Repository Layout**
* Restructured repository to align local, Nix-based, and distro-based build workflows
* Added Arch support and refined build/purge scripts
## [0.9.1] - 2025-12-10
* * Refactored installer: new `venv-create.sh`, cleaner root/user setup flow, updated README with architecture map.

View File

@@ -1,9 +1,12 @@
# ------------------------------------------------------------
# Base image selector — overridden by Makefile
# ------------------------------------------------------------
ARG BASE_IMAGE=archlinux:latest
ARG BASE_IMAGE
FROM ${BASE_IMAGE}
RUN echo "BASE_IMAGE=${BASE_IMAGE}" && \
cat /etc/os-release || true
# ------------------------------------------------------------
# Nix environment defaults
#

3
MIRRORS Normal file
View File

@@ -0,0 +1,3 @@
git@github.com:kevinveenbirkenbach/package-manager.git
ssh://git@git.veen.world:2201/kevinveenbirkenbach/pkgmgr.git
ssh://git@code.cymais.cloud:2201/kevinveenbirkenbach/pkgmgr.git

View File

@@ -2,11 +2,15 @@
test build build-no-cache test-unit test-e2e test-integration \
test-container
# Distro
# Options: arch debian ubuntu fedora centos
distro ?= arch
export distro
# ------------------------------------------------------------
# Distro list and base images
# Base images
# (kept for documentation/reference; actual build logic is in scripts/build)
# ------------------------------------------------------------
DISTROS := arch debian ubuntu fedora centos
BASE_IMAGE_ARCH := archlinux:latest
BASE_IMAGE_DEBIAN := debian:stable-slim
BASE_IMAGE_UBUNTU := ubuntu:latest
@@ -14,7 +18,6 @@ BASE_IMAGE_FEDORA := fedora:latest
BASE_IMAGE_CENTOS := quay.io/centos/centos:stream9
# Make them available in scripts
export DISTROS
export BASE_IMAGE_ARCH
export BASE_IMAGE_DEBIAN
export BASE_IMAGE_UBUNTU
@@ -65,6 +68,11 @@ build-missing:
# Combined test target for local + CI (unit + integration + e2e)
test: test-container test-unit test-integration test-e2e
delete-volumes:
@docker volume rm pkgmgr_nix_store_${distro} pkgmgr_nix_cache_${distro} || true
purge: delete-volumes build-no-cache
# ------------------------------------------------------------
# System install (native packages, calls scripts/installation/run-package.sh)
# ------------------------------------------------------------

View File

@@ -3,5 +3,4 @@
For the following checkout the implementation map:
- Implement TAGS
- Implement MIRROR
- Implement SIGNING_KEY

View File

@@ -36,7 +36,7 @@
rec {
pkgmgr = pyPkgs.buildPythonApplication {
pname = "package-manager";
version = "0.9.1";
version = "0.10.1";
# Use the git repo as source
src = ./.;

View File

@@ -9,7 +9,7 @@ Homepage: https://github.com/kevinveenbirkenbach/package-manager
Package: package-manager
Architecture: any
Depends: nix, ${misc:Depends}
Depends: sudo, ${misc:Depends}
Description: Wrapper that runs Kevin's package-manager via Nix flake
This package provides the `pkgmgr` command, which runs Kevin's package
manager via a local Nix flake

View File

@@ -7,7 +7,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "package-manager"
version = "0.9.1"
version = "0.10.1"
description = "Kevin's package-manager tool (pkgmgr)"
readme = "README.md"
requires-python = ">=3.11"

View File

@@ -4,32 +4,21 @@ set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
source "${SCRIPT_DIR}/resolve-base-image.sh"
echo "============================================================"
echo ">>> Building ONLY missing container images"
echo "============================================================"
IMAGE="package-manager-test-$distro"
BASE_IMAGE="$(resolve_base_image "$distro")"
for distro in $DISTROS; do
IMAGE="package-manager-test-$distro"
BASE_IMAGE="$(resolve_base_image "$distro")"
if docker image inspect "$IMAGE" >/dev/null 2>&1; then
echo "[build-missing] Image already exists: $IMAGE (skipping)"
continue
fi
echo
echo "------------------------------------------------------------"
echo "[build-missing] Building missing image: $IMAGE"
echo "BASE_IMAGE = $BASE_IMAGE"
echo "------------------------------------------------------------"
docker build \
--build-arg BASE_IMAGE="$BASE_IMAGE" \
-t "$IMAGE" \
.
done
if docker image inspect "$IMAGE" >/dev/null 2>&1; then
echo "[build-missing] Image already exists: $IMAGE (skipping)"
exit 0
fi
echo
echo "============================================================"
echo ">>> build-missing: Done"
echo "============================================================"
echo "------------------------------------------------------------"
echo "[build-missing] Building missing image: $IMAGE"
echo "BASE_IMAGE = $BASE_IMAGE"
echo "------------------------------------------------------------"
docker build \
--build-arg BASE_IMAGE="$BASE_IMAGE" \
-t "$IMAGE" \
.

View File

@@ -4,14 +4,12 @@ set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
source "${SCRIPT_DIR}/resolve-base-image.sh"
for distro in $DISTROS; do
base_image="$(resolve_base_image "$distro")"
base_image="$(resolve_base_image "$distro")"
echo ">>> Building test image for distro '$distro' with NO CACHE (BASE_IMAGE=$base_image)..."
echo ">>> Building test image for distro '$distro' with NO CACHE (BASE_IMAGE=$base_image)..."
docker build \
--no-cache \
--build-arg BASE_IMAGE="$base_image" \
-t "package-manager-test-$distro" \
.
done
docker build \
--no-cache \
--build-arg BASE_IMAGE="$base_image" \
-t "package-manager-test-$distro" \
.

View File

@@ -4,13 +4,11 @@ set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
source "${SCRIPT_DIR}/resolve-base-image.sh"
for distro in $DISTROS; do
base_image="$(resolve_base_image "$distro")"
base_image="$(resolve_base_image "$distro")"
echo ">>> Building test image for distro '$distro' (BASE_IMAGE=$base_image)..."
echo ">>> Building test image for distro '$distro' (BASE_IMAGE=$base_image)..."
docker build \
--build-arg BASE_IMAGE="$base_image" \
-t "package-manager-test-$distro" \
.
done
docker build \
--build-arg BASE_IMAGE="$base_image" \
-t "package-manager-test-$distro" \
.

View File

@@ -45,6 +45,26 @@ ensure_nix_on_path() {
fi
}
# ---------------------------------------------------------------------------
# Helper: ensure Nix build group and users exist (build-users-group = nixbld)
# ---------------------------------------------------------------------------
ensure_nix_build_group() {
# Ensure nixbld group (build-users-group for Nix)
if ! getent group nixbld >/dev/null 2>&1; then
echo "[init-nix] Creating group 'nixbld'..."
groupadd -r nixbld
fi
# Ensure Nix build users (nixbld1..nixbld10) as members of nixbld
for i in $(seq 1 10); do
if ! id "nixbld$i" >/dev/null 2>&1; then
echo "[init-nix] Creating build user nixbld$i..."
# -r: system account, -g: primary group, -G: supplementary (ensures membership is listed)
useradd -r -g nixbld -G nixbld -s /usr/sbin/nologin "nixbld$i"
fi
done
}
# ---------------------------------------------------------------------------
# Fast path: Nix already available
# ---------------------------------------------------------------------------
@@ -76,20 +96,8 @@ fi
if [[ "${IN_CONTAINER}" -eq 1 && "${EUID:-0}" -eq 0 ]]; then
echo "[init-nix] Running as root inside a container using dedicated 'nix' user."
# Ensure nixbld group (required by Nix)
if ! getent group nixbld >/dev/null 2>&1; then
echo "[init-nix] Creating group 'nixbld'..."
groupadd -r nixbld
fi
# Ensure Nix build users (nixbld1..nixbld10) as members of nixbld
for i in $(seq 1 10); do
if ! id "nixbld$i" >/dev/null 2>&1; then
echo "[init-nix] Creating build user nixbld$i..."
# -r: system account, -g: primary group, -G: supplementary (ensures membership is listed)
useradd -r -g nixbld -G nixbld -s /usr/sbin/nologin "nixbld$i"
fi
done
# Ensure build group/users for Nix
ensure_nix_build_group
# Ensure "nix" user (home at /home/nix)
if ! id nix >/dev/null 2>&1; then
@@ -187,14 +195,25 @@ if [[ "${IN_CONTAINER}" -eq 0 ]]; then
# Real host
if command -v systemctl >/dev/null 2>&1; then
echo "[init-nix] Host with systemd using multi-user install (--daemon)."
if [[ "${EUID:-0}" -eq 0 ]]; then
# Prepare build-users-group for Nix daemon installs
ensure_nix_build_group
fi
sh <(curl -L https://nixos.org/nix/install) --daemon
else
if [[ "${EUID:-0}" -eq 0 ]]; then
echo "[init-nix] WARNING: Running as root without systemd on host."
echo "[init-nix] Falling back to single-user install (--no-daemon), but this is not recommended."
# IMPORTANT: This is where Debian/Ubuntu inside your CI end up.
# We must ensure 'nixbld' exists before running the installer,
# otherwise modern Nix fails with: "the group 'nixbld' ... does not exist".
ensure_nix_build_group
sh <(curl -L https://nixos.org/nix/install) --no-daemon
else
echo "[init-nix] Non-root host without systemd using single-user install (--no-daemon)."
# Non-root cannot create nixbld group; rely on upstream defaults
sh <(curl -L https://nixos.org/nix/install) --no-daemon
fi
fi

View File

@@ -45,8 +45,42 @@ else
fi
echo "[aur-builder-setup] Ensuring yay is installed for aur_builder..."
if ! "${RUN_AS_AUR[@]}" 'command -v yay >/dev/null 2>&1'; then
"${RUN_AS_AUR[@]}" 'cd ~ && rm -rf yay && git clone https://aur.archlinux.org/yay.git && cd yay && makepkg -si --noconfirm'
echo "[aur-builder-setup] yay not found starting retry sequence for download..."
MAX_TIME=300 # 5 minutes
SLEEP_INTERVAL=20 # 20 seconds
ELAPSED=0
while true; do
if "${RUN_AS_AUR[@]}" '
set -euo pipefail
cd ~
rm -rf yay || true
git clone https://aur.archlinux.org/yay.git yay
'; then
echo "[aur-builder-setup] yay repository cloned successfully."
break
fi
echo "[aur-builder-setup] git clone failed (likely 504). Retrying in ${SLEEP_INTERVAL}s..."
sleep "${SLEEP_INTERVAL}"
ELAPSED=$((ELAPSED + SLEEP_INTERVAL))
if (( ELAPSED >= MAX_TIME )); then
echo "[aur-builder-setup] ERROR: Aborted after 5 minutes of retry attempts."
exit 1
fi
done
# Now build yay after successful clone
"${RUN_AS_AUR[@]}" '
set -euo pipefail
cd ~/yay
makepkg -si --noconfirm
'
else
echo "[aur-builder-setup] yay already installed."
fi

View File

@@ -8,6 +8,12 @@ source "${SCRIPT_DIR}/lib.sh"
OS_ID="$(detect_os_id)"
# Map Manjaro to Arch
if [[ "${OS_ID}" == "manjaro" ]]; then
echo "[run-package] Mapping OS 'manjaro' → 'arch'"
OS_ID="arch"
fi
echo "[run-package] Detected OS: ${OS_ID}"
case "${OS_ID}" in

View File

@@ -8,19 +8,18 @@ fi
FLAKE_DIR="/usr/lib/package-manager"
# ------------------------------------------------------------
# Try to ensure that "nix" is on PATH
# ------------------------------------------------------------
# ---------------------------------------------------------------------------
# Try to ensure that "nix" is on PATH (common locations + container user)
# ---------------------------------------------------------------------------
if ! command -v nix >/dev/null 2>&1; then
# Common locations for Nix installations
CANDIDATES=(
"/nix/var/nix/profiles/default/bin/nix"
"${HOME:-/root}/.nix-profile/bin/nix"
"/home/nix/.nix-profile/bin/nix"
)
for candidate in "${CANDIDATES[@]}"; do
if [[ -x "$candidate" ]]; then
# Prepend the directory of the candidate to PATH
PATH="$(dirname "$candidate"):${PATH}"
export PATH
break
@@ -28,13 +27,22 @@ if ! command -v nix >/dev/null 2>&1; then
done
fi
# ------------------------------------------------------------
# Primary (and only) path: use Nix flake if available
# ------------------------------------------------------------
# ---------------------------------------------------------------------------
# If nix is still missing, try to run init-nix.sh once
# ---------------------------------------------------------------------------
if ! command -v nix >/dev/null 2>&1; then
if [[ -x "${FLAKE_DIR}/init-nix.sh" ]]; then
"${FLAKE_DIR}/init-nix.sh" || true
fi
fi
# ---------------------------------------------------------------------------
# Primary path: use Nix flake if available
# ---------------------------------------------------------------------------
if command -v nix >/dev/null 2>&1; then
exec nix run "${FLAKE_DIR}#pkgmgr" -- "$@"
fi
echo "[pkgmgr-wrapper] ERROR: 'nix' binary not found on PATH."
echo "[pkgmgr-wrapper] ERROR: 'nix' binary not found on PATH after init."
echo "[pkgmgr-wrapper] Nix is required to run pkgmgr (no Python fallback)."
exit 1

View File

@@ -1,41 +1,32 @@
#!/usr/bin/env bash
set -euo pipefail
echo "============================================================"
echo ">>> Running sanity test: verifying test containers start"
echo "============================================================"
for distro in $DISTROS; do
IMAGE="package-manager-test-$distro"
echo
echo "------------------------------------------------------------"
echo ">>> Testing container: $IMAGE"
echo "------------------------------------------------------------"
echo "[test-container] Running: docker run --rm --entrypoint pkgmgr $IMAGE --help"
echo
# Run the command and capture the output
if OUTPUT=$(docker run --rm \
-e PKGMGR_DEV=1 \
-v pkgmgr_nix_store_${distro}:/nix \
-v "$(pwd):/src" \
-v "pkgmgr_nix_cache_${distro}:/root/.cache/nix" \
"$IMAGE" 2>&1); then
echo "$OUTPUT"
echo
echo "[test-container] SUCCESS: $IMAGE responded to 'pkgmgr --help'"
else
echo "$OUTPUT"
echo
echo "[test-container] ERROR: $IMAGE failed to run 'pkgmgr --help'"
exit 1
fi
done
IMAGE="package-manager-test-$distro"
echo
echo "============================================================"
echo ">>> All containers passed the sanity check"
echo "============================================================"
echo "------------------------------------------------------------"
echo ">>> Testing container: $IMAGE"
echo "------------------------------------------------------------"
echo "[test-container] Inspect image metadata:"
docker image inspect "$IMAGE" | sed -n '1,40p'
echo "[test-container] Running: docker run --rm --entrypoint pkgmgr $IMAGE --help"
echo
# Run the command and capture the output
if OUTPUT=$(docker run --rm \
-e PKGMGR_DEV=1 \
-v pkgmgr_nix_store_${distro}:/nix \
-v "$(pwd):/src" \
-v "pkgmgr_nix_cache_${distro}:/root/.cache/nix" \
"$IMAGE" 2>&1); then
echo "$OUTPUT"
echo
echo "[test-container] SUCCESS: $IMAGE responded to 'pkgmgr --help'"
else
echo "$OUTPUT"
echo
echo "[test-container] ERROR: $IMAGE failed to run 'pkgmgr --help'"
exit 1
fi

View File

@@ -1,65 +1,60 @@
#!/usr/bin/env bash
set -euo pipefail
echo ">>> Running E2E tests in all distros: $DISTROS"
echo "============================================================"
echo ">>> Running E2E tests: $distro"
echo "============================================================"
for distro in $DISTROS; do
echo "============================================================"
echo ">>> Running E2E tests: $distro"
echo "============================================================"
docker run --rm \
-v "$(pwd):/src" \
-v "pkgmgr_nix_store_${distro}:/nix" \
-v "pkgmgr_nix_cache_${distro}:/root/.cache/nix" \
-e PKGMGR_DEV=1 \
-e TEST_PATTERN="${TEST_PATTERN}" \
--workdir /src \
"package-manager-test-${distro}" \
bash -lc '
set -euo pipefail
docker run --rm \
-v "$(pwd):/src" \
-v "pkgmgr_nix_store_${distro}:/nix" \
-v "pkgmgr_nix_cache_${distro}:/root/.cache/nix" \
-e PKGMGR_DEV=1 \
-e TEST_PATTERN="${TEST_PATTERN}" \
--workdir /src \
--entrypoint bash \
"package-manager-test-${distro}" \
-c '
set -euo pipefail
# Load distro info
if [ -f /etc/os-release ]; then
. /etc/os-release
fi
# Load distro info
if [ -f /etc/os-release ]; then
. /etc/os-release
fi
echo "Running tests inside distro: ${ID:-unknown}"
echo "Running tests inside distro: ${ID:-unknown}"
# Load Nix environment if available
if [ -f "/nix/var/nix/profiles/default/etc/profile.d/nix-daemon.sh" ]; then
. "/nix/var/nix/profiles/default/etc/profile.d/nix-daemon.sh"
fi
# Load Nix environment if available
if [ -f "/nix/var/nix/profiles/default/etc/profile.d/nix-daemon.sh" ]; then
. "/nix/var/nix/profiles/default/etc/profile.d/nix-daemon.sh"
fi
if [ -f "$HOME/.nix-profile/etc/profile.d/nix.sh" ]; then
. "$HOME/.nix-profile/etc/profile.d/nix.sh"
fi
if [ -f "$HOME/.nix-profile/etc/profile.d/nix.sh" ]; then
. "$HOME/.nix-profile/etc/profile.d/nix.sh"
fi
PATH="/nix/var/nix/profiles/default/bin:$HOME/.nix-profile/bin:$PATH"
PATH="/nix/var/nix/profiles/default/bin:$HOME/.nix-profile/bin:$PATH"
command -v nix >/dev/null || {
echo "ERROR: nix not found."
exit 1
}
command -v nix >/dev/null || {
echo "ERROR: nix not found."
exit 1
}
# Mark the mounted repository as safe to avoid Git ownership errors.
# Newer Git (e.g. on Ubuntu) complains about the gitdir (/src/.git),
# older versions about the worktree (/src). Nix turns "." into the
# flake input "git+file:///src", which then uses Git under the hood.
if command -v git >/dev/null 2>&1; then
# Worktree path
git config --global --add safe.directory /src || true
# 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
fi
# Mark the mounted repository as safe to avoid Git ownership errors.
# Newer Git (e.g. on Ubuntu) complains about the gitdir (/src/.git),
# older versions about the worktree (/src). Nix turns "." into the
# flake input "git+file:///src", which then uses Git under the hood.
if command -v git >/dev/null 2>&1; then
# Worktree path
git config --global --add safe.directory /src || true
# 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
fi
# Run the E2E tests inside the Nix development shell
nix develop .#default --no-write-lock-file -c \
python3 -m unittest discover \
-s /src/tests/e2e \
-p "$TEST_PATTERN"
'
done
# Run the E2E tests inside the Nix development shell
nix develop .#default --no-write-lock-file -c \
python3 -m unittest discover \
-s /src/tests/e2e \
-p "$TEST_PATTERN"
'

View File

@@ -1,8 +1,6 @@
#!/usr/bin/env bash
set -euo pipefail
: "${distro:=arch}"
echo "============================================================"
echo ">>> Running INTEGRATION tests in ${distro} container"
echo "============================================================"
@@ -14,9 +12,8 @@ docker run --rm \
--workdir /src \
-e PKGMGR_DEV=1 \
-e TEST_PATTERN="${TEST_PATTERN}" \
--entrypoint bash \
"package-manager-test-${distro}" \
-c '
bash -lc '
set -e;
git config --global --add safe.directory /src || true;
nix develop .#default --no-write-lock-file -c \

View File

@@ -1,8 +1,6 @@
#!/usr/bin/env bash
set -euo pipefail
: "${distro:=arch}"
echo "============================================================"
echo ">>> Running UNIT tests in ${distro} container"
echo "============================================================"
@@ -14,9 +12,8 @@ docker run --rm \
--workdir /src \
-e PKGMGR_DEV=1 \
-e TEST_PATTERN="${TEST_PATTERN}" \
--entrypoint bash \
"package-manager-test-${distro}" \
-c '
bash -lc '
set -e;
git config --global --add safe.directory /src || true;
nix develop .#default --no-write-lock-file -c \

View File

@@ -139,22 +139,27 @@ class NixFlakeInstaller(BaseInstaller):
for output, allow_failure in outputs:
cmd = f"nix profile install {ctx.repo_dir}#{output}"
print(f"[INFO] Running: {cmd}")
ret = os.system(cmd)
try:
run_command(
cmd,
cwd=ctx.repo_dir,
preview=ctx.preview,
allow_failure=allow_failure,
)
# 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
if exit_code == 0:
print(f"Nix flake output '{output}' successfully installed.")
except SystemExit as e:
print(f"[Error] Failed to install Nix flake output '{output}': {e}")
if not allow_failure:
# Mandatory output failed → fatal for the pipeline.
raise
# Optional output failed → log and continue.
print(
"[Warning] Continuing despite failure to install "
f"optional output '{output}'."
)
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}'."
)

View File

@@ -0,0 +1,26 @@
from __future__ import annotations
"""
High-level mirror actions.
Public API:
- list_mirrors
- diff_mirrors
- merge_mirrors
- setup_mirrors
"""
from .types import Repository, MirrorMap
from .list_cmd import list_mirrors
from .diff_cmd import diff_mirrors
from .merge_cmd import merge_mirrors
from .setup_cmd import setup_mirrors
__all__ = [
"Repository",
"MirrorMap",
"list_mirrors",
"diff_mirrors",
"merge_mirrors",
"setup_mirrors",
]

View File

@@ -0,0 +1,31 @@
from __future__ import annotations
from typing import List
from pkgmgr.core.repository.dir import get_repo_dir
from pkgmgr.core.repository.identifier import get_repo_identifier
from .io import load_config_mirrors, read_mirrors_file
from .types import MirrorMap, RepoMirrorContext, Repository
def build_context(
repo: Repository,
repositories_base_dir: str,
all_repos: List[Repository],
) -> RepoMirrorContext:
"""
Build a RepoMirrorContext for a single repository.
"""
identifier = get_repo_identifier(repo, all_repos)
repo_dir = get_repo_dir(repositories_base_dir, repo)
config_mirrors: MirrorMap = load_config_mirrors(repo)
file_mirrors: MirrorMap = read_mirrors_file(repo_dir)
return RepoMirrorContext(
identifier=identifier,
repo_dir=repo_dir,
config_mirrors=config_mirrors,
file_mirrors=file_mirrors,
)

View File

@@ -0,0 +1,60 @@
from __future__ import annotations
from typing import List
from .context import build_context
from .printing import print_header
from .types import Repository
def diff_mirrors(
selected_repos: List[Repository],
repositories_base_dir: str,
all_repos: List[Repository],
) -> None:
"""
Show differences between config mirrors and MIRRORS file.
- Mirrors present only in config are reported as "ONLY IN CONFIG".
- Mirrors present only in MIRRORS file are reported as "ONLY IN FILE".
- Mirrors with same name but different URLs are reported as "URL MISMATCH".
"""
for repo in selected_repos:
ctx = build_context(repo, repositories_base_dir, all_repos)
print_header("[MIRROR DIFF]", ctx)
config_m = ctx.config_mirrors
file_m = ctx.file_mirrors
if not config_m and not file_m:
print(" No mirrors configured in config or MIRRORS file.")
print()
continue
# Mirrors only in config
for name, url in sorted(config_m.items()):
if name not in file_m:
print(f" [ONLY IN CONFIG] {name}: {url}")
# Mirrors only in MIRRORS file
for name, url in sorted(file_m.items()):
if name not in config_m:
print(f" [ONLY IN FILE] {name}: {url}")
# Mirrors with same name but different URLs
shared = set(config_m) & set(file_m)
for name in sorted(shared):
url_cfg = config_m.get(name)
url_file = file_m.get(name)
if url_cfg != url_file:
print(
f" [URL MISMATCH] {name}:\n"
f" config: {url_cfg}\n"
f" file: {url_file}"
)
if config_m and file_m and config_m == file_m:
print(" [OK] Mirrors in config and MIRRORS file are in sync.")
print()

View File

@@ -0,0 +1,179 @@
from __future__ import annotations
import os
from typing import List, Optional, Set
from pkgmgr.core.command.run import run_command
from pkgmgr.core.git import GitError, run_git
from .types import MirrorMap, RepoMirrorContext, Repository
def build_default_ssh_url(repo: Repository) -> Optional[str]:
"""
Build a simple SSH URL from repo config if no explicit mirror is defined.
Example: git@github.com:account/repository.git
"""
provider = repo.get("provider")
account = repo.get("account")
name = repo.get("repository")
port = repo.get("port")
if not provider or not account or not name:
return None
provider = str(provider)
account = str(account)
name = str(name)
if port:
return f"ssh://git@{provider}:{port}/{account}/{name}.git"
# GitHub-style shorthand
return f"git@{provider}:{account}/{name}.git"
def determine_primary_remote_url(
repo: Repository,
resolved_mirrors: MirrorMap,
) -> Optional[str]:
"""
Determine the primary remote URL in a consistent way:
1. resolved_mirrors["origin"]
2. any resolved mirror (first by name)
3. default SSH URL from provider/account/repository
"""
if "origin" in resolved_mirrors:
return resolved_mirrors["origin"]
if resolved_mirrors:
first_name = sorted(resolved_mirrors.keys())[0]
return resolved_mirrors[first_name]
return build_default_ssh_url(repo)
def _safe_git_output(args: List[str], cwd: str) -> Optional[str]:
"""
Run a Git command via run_git and return its stdout, or None on failure.
"""
try:
return run_git(args, cwd=cwd)
except GitError:
return None
def current_origin_url(repo_dir: str) -> Optional[str]:
"""
Return the current URL for remote 'origin', or None if not present.
"""
output = _safe_git_output(["remote", "get-url", "origin"], cwd=repo_dir)
if not output:
return None
url = output.strip()
return url or None
def has_origin_remote(repo_dir: str) -> bool:
"""
Check whether a remote called 'origin' exists in the repository.
"""
output = _safe_git_output(["remote"], cwd=repo_dir)
if not output:
return False
names = output.split()
return "origin" in names
def _ensure_push_urls_for_origin(
repo_dir: str,
mirrors: MirrorMap,
preview: bool,
) -> None:
"""
Ensure that all mirror URLs are present as push URLs on 'origin'.
"""
desired: Set[str] = {url for url in mirrors.values() if url}
if not desired:
return
existing_output = _safe_git_output(
["remote", "get-url", "--push", "--all", "origin"],
cwd=repo_dir,
)
existing = set(existing_output.splitlines()) if existing_output else set()
missing = sorted(desired - existing)
for url in missing:
cmd = f"git remote set-url --add --push origin {url}"
if preview:
print(f"[PREVIEW] Would run in {repo_dir!r}: {cmd}")
else:
print(f"[INFO] Adding push URL to 'origin': {url}")
run_command(cmd, cwd=repo_dir, preview=False)
def ensure_origin_remote(
repo: Repository,
ctx: RepoMirrorContext,
preview: bool,
) -> None:
"""
Ensure that a usable 'origin' remote exists and has all push URLs.
"""
repo_dir = ctx.repo_dir
resolved_mirrors = ctx.resolved_mirrors
if not os.path.isdir(os.path.join(repo_dir, ".git")):
print(f"[WARN] {repo_dir} is not a Git repository (no .git directory).")
return
url = determine_primary_remote_url(repo, resolved_mirrors)
if not has_origin_remote(repo_dir):
if not url:
print(
"[WARN] Could not determine URL for 'origin' remote. "
"Please configure mirrors or provider/account/repository."
)
return
cmd = f"git remote add origin {url}"
if preview:
print(f"[PREVIEW] Would run in {repo_dir!r}: {cmd}")
else:
print(f"[INFO] Adding 'origin' remote in {repo_dir}: {url}")
run_command(cmd, cwd=repo_dir, preview=False)
else:
current = current_origin_url(repo_dir)
if current == url or not url:
print(
f"[INFO] 'origin' already points to "
f"{current or '<unknown>'} (no change needed)."
)
else:
# We do not auto-change origin here, only log the mismatch.
print(
"[INFO] 'origin' exists with URL "
f"{current or '<unknown>'}; not changing to {url}."
)
# Ensure all mirrors are present as push URLs
_ensure_push_urls_for_origin(repo_dir, resolved_mirrors, preview)
def is_remote_reachable(url: str, cwd: Optional[str] = None) -> bool:
"""
Check whether a remote repository is reachable via `git ls-remote`.
This does NOT modify anything; it only probes the remote.
"""
workdir = cwd or os.getcwd()
try:
# --exit-code → non-zero exit code if the remote does not exist
run_git(["ls-remote", "--exit-code", url], cwd=workdir)
return True
except GitError:
return False

View File

@@ -0,0 +1,98 @@
from __future__ import annotations
import os
from urllib.parse import urlparse
from typing import List, Mapping
from .types import MirrorMap, Repository
def load_config_mirrors(repo: Repository) -> MirrorMap:
mirrors = repo.get("mirrors") or {}
result: MirrorMap = {}
if isinstance(mirrors, dict):
for name, url in mirrors.items():
if url:
result[str(name)] = str(url)
return result
if isinstance(mirrors, list):
for entry in mirrors:
if isinstance(entry, dict):
name = entry.get("name")
url = entry.get("url")
if name and url:
result[str(name)] = str(url)
return result
def read_mirrors_file(repo_dir: str, filename: str = "MIRRORS") -> MirrorMap:
"""
Supports:
NAME URL
URL → auto name = hostname
"""
path = os.path.join(repo_dir, filename)
mirrors: MirrorMap = {}
if not os.path.exists(path):
return mirrors
try:
with open(path, "r", encoding="utf-8") as fh:
for line in fh:
stripped = line.strip()
if not stripped or stripped.startswith("#"):
continue
parts = stripped.split(None, 1)
# Case 1: "name url"
if len(parts) == 2:
name, url = parts
# Case 2: "url" → auto-generate name
elif len(parts) == 1:
url = parts[0]
parsed = urlparse(url)
host = (parsed.netloc or "").split(":")[0]
base = host or "mirror"
name = base
i = 2
while name in mirrors:
name = f"{base}{i}"
i += 1
else:
continue
mirrors[name] = url
except OSError as exc:
print(f"[WARN] Could not read MIRRORS file at {path}: {exc}")
return mirrors
def write_mirrors_file(
repo_dir: str,
mirrors: Mapping[str, str],
filename: str = "MIRRORS",
preview: bool = False,
) -> None:
path = os.path.join(repo_dir, filename)
lines = [f"{name} {url}" for name, url in sorted(mirrors.items())]
content = "\n".join(lines) + ("\n" if lines else "")
if preview:
print(f"[PREVIEW] Would write MIRRORS file at {path}:")
print(content or "(empty)")
return
try:
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "w", encoding="utf-8") as fh:
fh.write(content)
print(f"[INFO] Wrote MIRRORS file at {path}")
except OSError as exc:
print(f"[ERROR] Failed to write MIRRORS file at {path}: {exc}")

View File

@@ -0,0 +1,46 @@
from __future__ import annotations
from typing import List
from .context import build_context
from .printing import print_header, print_named_mirrors
from .types import Repository
def list_mirrors(
selected_repos: List[Repository],
repositories_base_dir: str,
all_repos: List[Repository],
source: str = "all",
) -> None:
"""
List mirrors for the selected repositories.
source:
- "config" → only mirrors from configuration
- "file" → only mirrors from MIRRORS file
- "resolved" → merged view (config + file, file wins)
- "all" → show config + file + resolved
"""
for repo in selected_repos:
ctx = build_context(repo, repositories_base_dir, all_repos)
resolved_m = ctx.resolved_mirrors
print_header("[MIRROR]", ctx)
if source in ("config", "all"):
print_named_mirrors("config mirrors", ctx.config_mirrors)
if source == "config":
print()
continue # next repo
if source in ("file", "all"):
print_named_mirrors("MIRRORS file", ctx.file_mirrors)
if source == "file":
print()
continue # next repo
if source in ("resolved", "all"):
print_named_mirrors("resolved mirrors", resolved_m)
print()

View File

@@ -0,0 +1,162 @@
from __future__ import annotations
import os
from typing import Dict, List, Tuple, Optional
import yaml
from pkgmgr.core.config.save import save_user_config
from .context import build_context
from .io import write_mirrors_file
from .types import MirrorMap, Repository
# -----------------------------------------------------------------------------
# Helpers
# -----------------------------------------------------------------------------
def _repo_key(repo: Repository) -> Tuple[str, str, str]:
"""
Normalised key for identifying a repository in config files.
"""
return (
str(repo.get("provider", "")),
str(repo.get("account", "")),
str(repo.get("repository", "")),
)
def _load_user_config(path: str) -> Dict[str, object]:
"""
Load a user config YAML file as dict.
Non-dicts yield {}.
"""
if not os.path.exists(path):
return {}
try:
with open(path, "r", encoding="utf-8") as f:
data = yaml.safe_load(f) or {}
return data if isinstance(data, dict) else {}
except Exception:
return {}
# -----------------------------------------------------------------------------
# Main merge command
# -----------------------------------------------------------------------------
def merge_mirrors(
selected_repos: List[Repository],
repositories_base_dir: str,
all_repos: List[Repository],
source: str,
target: str,
preview: bool = False,
user_config_path: Optional[str] = None,
) -> None:
"""
Merge mirrors between config and MIRRORS file.
Rules:
- source, target ∈ {"config", "file"}.
- merged = (target_mirrors overridden by source_mirrors)
- If target == "file" → write MIRRORS file.
- If target == "config":
* update the user config YAML directly
* write it using save_user_config()
The merge strategy is:
dst + src (src wins on same name)
"""
# Load user config once if we intend to write to it.
user_cfg: Optional[Dict[str, object]] = None
user_cfg_path_expanded: Optional[str] = None
if target == "config" and user_config_path and not preview:
user_cfg_path_expanded = os.path.expanduser(user_config_path)
user_cfg = _load_user_config(user_cfg_path_expanded)
if not isinstance(user_cfg.get("repositories"), list):
user_cfg["repositories"] = []
for repo in selected_repos:
ctx = build_context(repo, repositories_base_dir, all_repos)
print("============================================================")
print(f"[MIRROR MERGE] Repository: {ctx.identifier}")
print(f"[MIRROR MERGE] Directory: {ctx.repo_dir}")
print(f"[MIRROR MERGE] {source}{target}")
print("============================================================")
# Pick the correct source/target maps
if source == "config":
src = ctx.config_mirrors
dst = ctx.file_mirrors
else: # source == "file"
src = ctx.file_mirrors
dst = ctx.config_mirrors
# Merge (src overrides dst)
merged: MirrorMap = dict(dst)
merged.update(src)
# ---------------------------------------------------------
# WRITE TO FILE
# ---------------------------------------------------------
if target == "file":
write_mirrors_file(ctx.repo_dir, merged, preview=preview)
print()
continue
# ---------------------------------------------------------
# WRITE TO CONFIG
# ---------------------------------------------------------
if target == "config":
# If preview or no config path → show intended output
if preview or not user_cfg:
print("[INFO] The following mirrors would be written to config:")
if not merged:
print(" (no mirrors)")
else:
for name, url in sorted(merged.items()):
print(f" - {name}: {url}")
print(" (Config not modified due to preview or missing path.)")
print()
continue
repos = user_cfg.get("repositories")
target_key = _repo_key(repo)
existing_repo: Optional[Repository] = None
# Find existing repo entry
for entry in repos:
if isinstance(entry, dict) and _repo_key(entry) == target_key:
existing_repo = entry
break
# Create entry if missing
if existing_repo is None:
existing_repo = {
"provider": repo.get("provider"),
"account": repo.get("account"),
"repository": repo.get("repository"),
}
repos.append(existing_repo)
# Write or delete mirrors
if merged:
existing_repo["mirrors"] = dict(merged)
else:
existing_repo.pop("mirrors", None)
print(" [OK] Updated repo['mirrors'] in user config.")
print()
# -------------------------------------------------------------
# SAVE CONFIG (once at the end)
# -------------------------------------------------------------
if user_cfg is not None and user_cfg_path_expanded is not None and not preview:
save_user_config(user_cfg, user_cfg_path_expanded)
print(f"[OK] Saved updated config: {user_cfg_path_expanded}")

View File

@@ -0,0 +1,35 @@
from __future__ import annotations
from .types import MirrorMap, RepoMirrorContext
def print_header(
title_prefix: str,
ctx: RepoMirrorContext,
) -> None:
"""
Print a standard header for mirror-related output.
title_prefix examples:
- "[MIRROR]"
- "[MIRROR DIFF]"
- "[MIRROR MERGE]"
- "[MIRROR SETUP:LOCAL]"
- "[MIRROR SETUP:REMOTE]"
"""
print("============================================================")
print(f"{title_prefix} Repository: {ctx.identifier}")
print(f"{title_prefix} Directory: {ctx.repo_dir}")
print("============================================================")
def print_named_mirrors(label: str, mirrors: MirrorMap) -> None:
"""
Print a labeled mirror block (e.g. '[config mirrors]').
"""
print(f" [{label}]")
if mirrors:
for name, url in sorted(mirrors.items()):
print(f" - {name}: {url}")
else:
print(" (none)")

View File

@@ -0,0 +1,165 @@
from __future__ import annotations
from typing import List, Tuple
from pkgmgr.core.git import run_git, GitError
from .context import build_context
from .git_remote import determine_primary_remote_url, ensure_origin_remote
from .types import Repository
def _setup_local_mirrors_for_repo(
repo: Repository,
repositories_base_dir: str,
all_repos: List[Repository],
preview: bool,
) -> None:
"""
Ensure local Git state is sane (currently: 'origin' remote).
"""
ctx = build_context(repo, repositories_base_dir, all_repos)
print("------------------------------------------------------------")
print(f"[MIRROR SETUP:LOCAL] {ctx.identifier}")
print(f"[MIRROR SETUP:LOCAL] dir: {ctx.repo_dir}")
print("------------------------------------------------------------")
ensure_origin_remote(repo, ctx, preview=preview)
print()
def _probe_mirror(url: str, repo_dir: str) -> Tuple[bool, str]:
"""
Probe a remote mirror by running `git ls-remote <url>`.
Returns:
(True, "") on success,
(False, error_message) on failure.
Wichtig:
- Wir werten ausschließlich den Exit-Code aus.
- STDERR kann Hinweise/Warnings enthalten und ist NICHT automatisch ein Fehler.
"""
try:
# Wir ignorieren stdout komplett; wichtig ist nur, dass der Befehl ohne
# GitError (also Exit-Code 0) durchläuft.
run_git(["ls-remote", url], cwd=repo_dir)
return True, ""
except GitError as exc:
return False, str(exc)
def _setup_remote_mirrors_for_repo(
repo: Repository,
repositories_base_dir: str,
all_repos: List[Repository],
preview: bool,
) -> None:
"""
Remote-side setup / validation.
Aktuell werden nur **nicht-destruktive Checks** gemacht:
- Für jeden Mirror (aus config + MIRRORS-Datei, file gewinnt):
* `git ls-remote <url>` wird ausgeführt.
* Bei Exit-Code 0 → [OK]
* Bei Fehler → [WARN] + Details aus der GitError-Exception
Es werden **keine** Provider-APIs aufgerufen und keine Repos angelegt.
"""
ctx = build_context(repo, repositories_base_dir, all_repos)
resolved_m = ctx.resolved_mirrors
print("------------------------------------------------------------")
print(f"[MIRROR SETUP:REMOTE] {ctx.identifier}")
print(f"[MIRROR SETUP:REMOTE] dir: {ctx.repo_dir}")
print("------------------------------------------------------------")
if not resolved_m:
# Optional: Fallback auf eine heuristisch bestimmte URL, falls wir
# irgendwann "automatisch anlegen" implementieren wollen.
primary_url = determine_primary_remote_url(repo, resolved_m)
if not primary_url:
print(
"[INFO] No mirrors configured (config or MIRRORS file), and no "
"primary URL could be derived from provider/account/repository."
)
print()
return
ok, error_message = _probe_mirror(primary_url, ctx.repo_dir)
if ok:
print(f"[OK] Remote mirror (primary) is reachable: {primary_url}")
else:
print("[WARN] Primary remote URL is NOT reachable:")
print(f" {primary_url}")
if error_message:
print(" Details:")
for line in error_message.splitlines():
print(f" {line}")
print()
print(
"[INFO] Remote checks are non-destructive and only use `git ls-remote` "
"to probe mirror URLs."
)
print()
return
# Normaler Fall: wir haben benannte Mirrors aus config/MIRRORS
for name, url in sorted(resolved_m.items()):
ok, error_message = _probe_mirror(url, ctx.repo_dir)
if ok:
print(f"[OK] Remote mirror '{name}' is reachable: {url}")
else:
print(f"[WARN] Remote mirror '{name}' is NOT reachable:")
print(f" {url}")
if error_message:
print(" Details:")
for line in error_message.splitlines():
print(f" {line}")
print()
print(
"[INFO] Remote checks are non-destructive and only use `git ls-remote` "
"to probe mirror URLs."
)
print()
def setup_mirrors(
selected_repos: List[Repository],
repositories_base_dir: str,
all_repos: List[Repository],
preview: bool = False,
local: bool = True,
remote: bool = True,
) -> None:
"""
Setup mirrors for the selected repositories.
local:
- Configure local Git remotes (currently: ensure 'origin' is present and
points to a reasonable URL).
remote:
- Non-destructive remote checks using `git ls-remote` for each mirror URL.
Es werden keine Repositories auf dem Provider angelegt.
"""
for repo in selected_repos:
if local:
_setup_local_mirrors_for_repo(
repo,
repositories_base_dir=repositories_base_dir,
all_repos=all_repos,
preview=preview,
)
if remote:
_setup_remote_mirrors_for_repo(
repo,
repositories_base_dir=repositories_base_dir,
all_repos=all_repos,
preview=preview,
)

View File

@@ -0,0 +1,32 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Dict
Repository = Dict[str, Any]
MirrorMap = Dict[str, str]
@dataclass(frozen=True)
class RepoMirrorContext:
"""
Bundle mirror-related information for a single repository.
"""
identifier: str
repo_dir: str
config_mirrors: MirrorMap
file_mirrors: MirrorMap
@property
def resolved_mirrors(self) -> MirrorMap:
"""
Combined mirrors from config and MIRRORS file.
Strategy:
- Start from config mirrors
- Overlay MIRRORS file (file wins on same name)
"""
merged: MirrorMap = dict(self.config_mirrors)
merged.update(self.file_mirrors)
return merged

View File

@@ -6,6 +6,7 @@ from .version import handle_version
from .make import handle_make
from .changelog import handle_changelog
from .branch import handle_branch
from .mirror import handle_mirror_command
__all__ = [
"handle_repos_command",
@@ -16,4 +17,5 @@ __all__ = [
"handle_make",
"handle_changelog",
"handle_branch",
"handle_mirror_command",
]

View File

@@ -0,0 +1,118 @@
from __future__ import annotations
import sys
from typing import Any, Dict, List
from pkgmgr.actions.mirror import (
diff_mirrors,
list_mirrors,
merge_mirrors,
setup_mirrors,
)
from pkgmgr.cli.context import CLIContext
Repository = Dict[str, Any]
def handle_mirror_command(
args,
ctx: CLIContext,
selected: List[Repository],
) -> None:
"""
Entry point for 'pkgmgr mirror' subcommands.
Subcommands:
- mirror list → list configured mirrors
- mirror diff → compare config vs MIRRORS file
- mirror merge → merge mirrors between config and MIRRORS file
- mirror setup → configure local Git + remote placeholders
"""
if not selected:
print("[INFO] No repositories selected for 'mirror' command.")
sys.exit(1)
subcommand = getattr(args, "subcommand", None)
# ------------------------------------------------------------
# mirror list
# ------------------------------------------------------------
if subcommand == "list":
source = getattr(args, "source", "all")
list_mirrors(
selected_repos=selected,
repositories_base_dir=ctx.repositories_base_dir,
all_repos=ctx.all_repositories,
source=source,
)
return
# ------------------------------------------------------------
# mirror diff
# ------------------------------------------------------------
if subcommand == "diff":
diff_mirrors(
selected_repos=selected,
repositories_base_dir=ctx.repositories_base_dir,
all_repos=ctx.all_repositories,
)
return
# ------------------------------------------------------------
# mirror merge
# ------------------------------------------------------------
if subcommand == "merge":
source = getattr(args, "source", None)
target = getattr(args, "target", None)
preview = getattr(args, "preview", False)
if source == target:
print(
"[ERROR] For 'mirror merge', source and target "
"must differ (one of: config, file)."
)
sys.exit(2)
# Config file path can be passed explicitly via --config-path.
# If not given, fall back to the global context (if available).
explicit_config_path = getattr(args, "config_path", None)
user_config_path = explicit_config_path or getattr(
ctx, "user_config_path", None
)
merge_mirrors(
selected_repos=selected,
repositories_base_dir=ctx.repositories_base_dir,
all_repos=ctx.all_repositories,
source=source,
target=target,
preview=preview,
user_config_path=user_config_path,
)
return
# ------------------------------------------------------------
# mirror setup
# ------------------------------------------------------------
if subcommand == "setup":
local = getattr(args, "local", False)
remote = getattr(args, "remote", False)
preview = getattr(args, "preview", False)
# If neither flag is set → default to both.
if not local and not remote:
local = True
remote = True
setup_mirrors(
selected_repos=selected,
repositories_base_dir=ctx.repositories_base_dir,
all_repos=ctx.all_repositories,
preview=preview,
local=local,
remote=remote,
)
return
print(f"[ERROR] Unknown mirror subcommand: {subcommand}")
sys.exit(2)

View File

@@ -21,9 +21,9 @@ from pkgmgr.cli.commands import (
handle_make,
handle_changelog,
handle_branch,
handle_mirror_command,
)
def _has_explicit_selection(args) -> bool:
"""
Return True if the user explicitly selected repositories via
@@ -108,6 +108,7 @@ def dispatch_command(args, ctx: CLIContext) -> None:
"explore",
"terminal",
"code",
"mirror",
]
if getattr(args, "command", None) in commands_with_selection:
@@ -174,5 +175,9 @@ def dispatch_command(args, ctx: CLIContext) -> None:
handle_branch(args, ctx)
return
if args.command == "mirror":
handle_mirror_command(args, ctx, selected)
return
print(f"Unknown command: {args.command}")
sys.exit(2)

View File

@@ -1,505 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations
import argparse
from pkgmgr.cli.proxy import register_proxy_commands
class SortedSubParsersAction(argparse._SubParsersAction):
"""
Subparsers action that keeps choices sorted alphabetically.
"""
def add_parser(self, name, **kwargs):
parser = super().add_parser(name, **kwargs)
# Sort choices alphabetically by dest (subcommand name)
self._choices_actions.sort(key=lambda a: a.dest)
return parser
def add_identifier_arguments(subparser: argparse.ArgumentParser) -> None:
"""
Common identifier / selection arguments for many subcommands.
Selection modes (mutual intent, not hard-enforced):
- identifiers (positional): select by alias / provider/account/repo
- --all: select all repositories
- --category / --string / --tag: filter-based selection on top
of the full repository set
"""
subparser.add_argument(
"identifiers",
nargs="*",
help=(
"Identifier(s) for repositories. "
"Default: Repository of current folder."
),
)
subparser.add_argument(
"--all",
action="store_true",
default=False,
help=(
"Apply the subcommand to all repositories in the config. "
"Some subcommands ask for confirmation. If you want to give this "
"confirmation for all repositories, pipe 'yes'. E.g: "
"yes | pkgmgr {subcommand} --all"
),
)
subparser.add_argument(
"--category",
nargs="+",
default=[],
help=(
"Filter repositories by category patterns derived from config "
"filenames or repo metadata (use filename without .yml/.yaml, "
"or /regex/ to use a regular expression)."
),
)
subparser.add_argument(
"--string",
default="",
help=(
"Filter repositories whose identifier / name / path contains this "
"substring (case-insensitive). Use /regex/ for regular expressions."
),
)
subparser.add_argument(
"--tag",
action="append",
default=[],
help=(
"Filter repositories by tag. Matches tags from the repository "
"collector and category tags. Use /regex/ for regular expressions."
),
)
subparser.add_argument(
"--preview",
action="store_true",
help="Preview changes without executing commands",
)
subparser.add_argument(
"--list",
action="store_true",
help="List affected repositories (with preview or status)",
)
subparser.add_argument(
"-a",
"--args",
nargs=argparse.REMAINDER,
dest="extra_args",
help="Additional parameters to be attached.",
default=[],
)
def add_install_update_arguments(subparser: argparse.ArgumentParser) -> None:
"""
Common arguments for install/update commands.
"""
add_identifier_arguments(subparser)
subparser.add_argument(
"-q",
"--quiet",
action="store_true",
help="Suppress warnings and info messages",
)
subparser.add_argument(
"--no-verification",
action="store_true",
default=False,
help="Disable verification via commit/gpg",
)
subparser.add_argument(
"--dependencies",
action="store_true",
help="Also pull and update dependencies",
)
subparser.add_argument(
"--clone-mode",
choices=["ssh", "https", "shallow"],
default="ssh",
help=(
"Specify the clone mode: ssh, https, or shallow "
"(HTTPS shallow clone; default: ssh)"
),
)
def create_parser(description_text: str) -> argparse.ArgumentParser:
"""
Create the top-level argument parser for pkgmgr.
"""
parser = argparse.ArgumentParser(
description=description_text,
formatter_class=argparse.RawTextHelpFormatter,
)
subparsers = parser.add_subparsers(
dest="command",
help="Subcommands",
action=SortedSubParsersAction,
)
# ------------------------------------------------------------
# install / update / deinstall / delete
# ------------------------------------------------------------
install_parser = subparsers.add_parser(
"install",
help="Setup repository/repositories alias links to executables",
)
add_install_update_arguments(install_parser)
update_parser = subparsers.add_parser(
"update",
help="Update (pull + install) repository/repositories",
)
add_install_update_arguments(update_parser)
update_parser.add_argument(
"--system",
action="store_true",
help="Include system update commands",
)
deinstall_parser = subparsers.add_parser(
"deinstall",
help="Remove alias links to repository/repositories",
)
add_identifier_arguments(deinstall_parser)
delete_parser = subparsers.add_parser(
"delete",
help="Delete repository/repositories alias links to executables",
)
add_identifier_arguments(delete_parser)
# ------------------------------------------------------------
# create
# ------------------------------------------------------------
create_cmd_parser = subparsers.add_parser(
"create",
help=(
"Create new repository entries: add them to the config if not "
"already present, initialize the local repository, and push "
"remotely if --remote is set."
),
)
add_identifier_arguments(create_cmd_parser)
create_cmd_parser.add_argument(
"--remote",
action="store_true",
help="If set, add the remote and push the initial commit.",
)
# ------------------------------------------------------------
# status
# ------------------------------------------------------------
status_parser = subparsers.add_parser(
"status",
help="Show status for repository/repositories or system",
)
add_identifier_arguments(status_parser)
status_parser.add_argument(
"--system",
action="store_true",
help="Show system status",
)
# ------------------------------------------------------------
# config
# ------------------------------------------------------------
config_parser = subparsers.add_parser(
"config",
help="Manage configuration",
)
config_subparsers = config_parser.add_subparsers(
dest="subcommand",
help="Config subcommands",
required=True,
)
config_show = config_subparsers.add_parser(
"show",
help="Show configuration",
)
add_identifier_arguments(config_show)
config_subparsers.add_parser(
"add",
help="Interactively add a new repository entry",
)
config_subparsers.add_parser(
"edit",
help="Edit configuration file with nano",
)
config_subparsers.add_parser(
"init",
help="Initialize user configuration by scanning the base directory",
)
config_delete = config_subparsers.add_parser(
"delete",
help="Delete repository entry from user config",
)
add_identifier_arguments(config_delete)
config_ignore = config_subparsers.add_parser(
"ignore",
help="Set ignore flag for repository entries in user config",
)
add_identifier_arguments(config_ignore)
config_ignore.add_argument(
"--set",
choices=["true", "false"],
required=True,
help="Set ignore to true or false",
)
config_subparsers.add_parser(
"update",
help=(
"Update default config files in ~/.config/pkgmgr/ from the "
"installed pkgmgr package (does not touch config.yaml)."
),
)
# ------------------------------------------------------------
# path / explore / terminal / code / shell
# ------------------------------------------------------------
path_parser = subparsers.add_parser(
"path",
help="Print the path(s) of repository/repositories",
)
add_identifier_arguments(path_parser)
explore_parser = subparsers.add_parser(
"explore",
help="Open repository in Nautilus file manager",
)
add_identifier_arguments(explore_parser)
terminal_parser = subparsers.add_parser(
"terminal",
help="Open repository in a new GNOME Terminal tab",
)
add_identifier_arguments(terminal_parser)
code_parser = subparsers.add_parser(
"code",
help="Open repository workspace with VS Code",
)
add_identifier_arguments(code_parser)
shell_parser = subparsers.add_parser(
"shell",
help="Execute a shell command in each repository",
)
add_identifier_arguments(shell_parser)
shell_parser.add_argument(
"-c",
"--command",
nargs=argparse.REMAINDER,
dest="shell_command",
help=(
"The shell command (and its arguments) to execute in each "
"repository"
),
default=[],
)
# ------------------------------------------------------------
# branch
# ------------------------------------------------------------
branch_parser = subparsers.add_parser(
"branch",
help="Branch-related utilities (e.g. open/close feature branches)",
)
branch_subparsers = branch_parser.add_subparsers(
dest="subcommand",
help="Branch subcommands",
required=True,
)
branch_open = branch_subparsers.add_parser(
"open",
help="Create and push a new branch on top of a base branch",
)
branch_open.add_argument(
"name",
nargs="?",
help=(
"Name of the new branch (optional; will be asked interactively "
"if omitted)"
),
)
branch_open.add_argument(
"--base",
default="main",
help="Base branch to create the new branch from (default: main)",
)
branch_close = branch_subparsers.add_parser(
"close",
help="Merge a feature branch into base and delete it",
)
branch_close.add_argument(
"name",
nargs="?",
help=(
"Name of the branch to close (optional; current branch is used "
"if omitted)"
),
)
branch_close.add_argument(
"--base",
default="main",
help=(
"Base branch to merge into (default: main; falls back to master "
"internally if main does not exist)"
),
)
# ------------------------------------------------------------
# release
# ------------------------------------------------------------
release_parser = subparsers.add_parser(
"release",
help=(
"Create a release for repository/ies by incrementing version "
"and updating the changelog."
),
)
release_parser.add_argument(
"release_type",
choices=["major", "minor", "patch"],
help="Type of version increment for the release (major, minor, patch).",
)
release_parser.add_argument(
"-m",
"--message",
default=None,
help=(
"Optional release message to add to the changelog and tag."
),
)
# Generic selection / preview / list / extra_args
add_identifier_arguments(release_parser)
# Close current branch after successful release
release_parser.add_argument(
"--close",
action="store_true",
help=(
"Close the current branch after a successful release in each "
"repository, if it is not main/master."
),
)
# Force: skip preview+confirmation and run release directly
release_parser.add_argument(
"-f",
"--force",
action="store_true",
help=(
"Skip the interactive preview+confirmation step and run the "
"release directly."
),
)
# ------------------------------------------------------------
# version
# ------------------------------------------------------------
version_parser = subparsers.add_parser(
"version",
help=(
"Show version information for repository/ies "
"(git tags, pyproject.toml, flake.nix, PKGBUILD, debian, spec, "
"Ansible Galaxy)."
),
)
add_identifier_arguments(version_parser)
# ------------------------------------------------------------
# changelog
# ------------------------------------------------------------
changelog_parser = subparsers.add_parser(
"changelog",
help=(
"Show changelog derived from Git history. "
"By default, shows the changes between the last two SemVer tags."
),
)
changelog_parser.add_argument(
"range",
nargs="?",
default="",
help=(
"Optional tag or range (e.g. v1.2.3 or v1.2.0..v1.2.3). "
"If omitted, the changelog between the last two SemVer "
"tags is shown."
),
)
add_identifier_arguments(changelog_parser)
# ------------------------------------------------------------
# list
# ------------------------------------------------------------
list_parser = subparsers.add_parser(
"list",
help="List all repositories with details and status",
)
# dieselbe Selektionslogik wie bei install/update/etc.:
add_identifier_arguments(list_parser)
list_parser.add_argument(
"--status",
type=str,
default="",
help=(
"Filter repositories by status (case insensitive). "
"Use /regex/ for regular expressions."
),
)
list_parser.add_argument(
"--description",
action="store_true",
help=(
"Show an additional detailed section per repository "
"(description, homepage, tags, categories, paths)."
),
)
# ------------------------------------------------------------
# make
# ------------------------------------------------------------
make_parser = subparsers.add_parser(
"make",
help="Executes make commands",
)
add_identifier_arguments(make_parser)
make_subparsers = make_parser.add_subparsers(
dest="subcommand",
help="Make subcommands",
required=True,
)
make_install = make_subparsers.add_parser(
"install",
help="Executes the make install command",
)
add_identifier_arguments(make_install)
make_deinstall = make_subparsers.add_parser(
"deinstall",
help="Executes the make deinstall command",
)
add_identifier_arguments(make_deinstall)
# ------------------------------------------------------------
# Proxy commands (git, docker, docker compose, ...)
# ------------------------------------------------------------
register_proxy_commands(subparsers)
return parser

View File

@@ -0,0 +1,68 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations
import argparse
from pkgmgr.cli.proxy import register_proxy_commands
from .common import SortedSubParsersAction
from .install_update import add_install_update_subparsers
from .config_cmd import add_config_subparsers
from .navigation_cmd import add_navigation_subparsers
from .branch_cmd import add_branch_subparsers
from .release_cmd import add_release_subparser
from .version_cmd import add_version_subparser
from .changelog_cmd import add_changelog_subparser
from .list_cmd import add_list_subparser
from .make_cmd import add_make_subparsers
from .mirror_cmd import add_mirror_subparsers
def create_parser(description_text: str) -> argparse.ArgumentParser:
"""
Create the top-level argument parser for pkgmgr.
"""
parser = argparse.ArgumentParser(
description=description_text,
formatter_class=argparse.RawTextHelpFormatter,
)
subparsers = parser.add_subparsers(
dest="command",
help="Subcommands",
action=SortedSubParsersAction,
)
# Core repo operations
add_install_update_subparsers(subparsers)
add_config_subparsers(subparsers)
# Navigation / tooling around repos
add_navigation_subparsers(subparsers)
# Branch & release workflow
add_branch_subparsers(subparsers)
add_release_subparser(subparsers)
# Info commands
add_version_subparser(subparsers)
add_changelog_subparser(subparsers)
add_list_subparser(subparsers)
# Make wrapper
add_make_subparsers(subparsers)
# Mirror management
add_mirror_subparsers(subparsers)
# Proxy commands (git, docker, docker compose, ...)
register_proxy_commands(subparsers)
return parser
__all__ = [
"create_parser",
"SortedSubParsersAction",
]

View File

@@ -0,0 +1,62 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations
import argparse
def add_branch_subparsers(
subparsers: argparse._SubParsersAction,
) -> None:
"""
Register branch command and its subcommands.
"""
branch_parser = subparsers.add_parser(
"branch",
help="Branch-related utilities (e.g. open/close feature branches)",
)
branch_subparsers = branch_parser.add_subparsers(
dest="subcommand",
help="Branch subcommands",
required=True,
)
branch_open = branch_subparsers.add_parser(
"open",
help="Create and push a new branch on top of a base branch",
)
branch_open.add_argument(
"name",
nargs="?",
help=(
"Name of the new branch (optional; will be asked interactively "
"if omitted)"
),
)
branch_open.add_argument(
"--base",
default="main",
help="Base branch to create the new branch from (default: main)",
)
branch_close = branch_subparsers.add_parser(
"close",
help="Merge a feature branch into base and delete it",
)
branch_close.add_argument(
"name",
nargs="?",
help=(
"Name of the branch to close (optional; current branch is used "
"if omitted)"
),
)
branch_close.add_argument(
"--base",
default="main",
help=(
"Base branch to merge into (default: main; falls back to master "
"internally if main does not exist)"
),
)

View File

@@ -0,0 +1,34 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations
import argparse
from .common import add_identifier_arguments
def add_changelog_subparser(
subparsers: argparse._SubParsersAction,
) -> None:
"""
Register the changelog command.
"""
changelog_parser = subparsers.add_parser(
"changelog",
help=(
"Show changelog derived from Git history. "
"By default, shows the changes between the last two SemVer tags."
),
)
changelog_parser.add_argument(
"range",
nargs="?",
default="",
help=(
"Optional tag or range (e.g. v1.2.3 or v1.2.0..v1.2.3). "
"If omitted, the changelog between the last two SemVer "
"tags is shown."
),
)
add_identifier_arguments(changelog_parser)

View File

@@ -0,0 +1,127 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations
import argparse
class SortedSubParsersAction(argparse._SubParsersAction):
"""
Subparsers action that keeps choices sorted alphabetically.
"""
def add_parser(self, name, **kwargs):
parser = super().add_parser(name, **kwargs)
# Sort choices alphabetically by dest (subcommand name)
self._choices_actions.sort(key=lambda a: a.dest)
return parser
def add_identifier_arguments(subparser: argparse.ArgumentParser) -> None:
"""
Common identifier / selection arguments for many subcommands.
Selection modes (mutual intent, not hard-enforced):
- identifiers (positional): select by alias / provider/account/repo
- --all: select all repositories
- --category / --string / --tag: filter-based selection on top
of the full repository set
"""
subparser.add_argument(
"identifiers",
nargs="*",
help=(
"Identifier(s) for repositories. "
"Default: Repository of current folder."
),
)
subparser.add_argument(
"--all",
action="store_true",
default=False,
help=(
"Apply the subcommand to all repositories in the config. "
"Some subcommands ask for confirmation. If you want to give this "
"confirmation for all repositories, pipe 'yes'. E.g: "
"yes | pkgmgr {subcommand} --all"
),
)
subparser.add_argument(
"--category",
nargs="+",
default=[],
help=(
"Filter repositories by category patterns derived from config "
"filenames or repo metadata (use filename without .yml/.yaml, "
"or /regex/ to use a regular expression)."
),
)
subparser.add_argument(
"--string",
default="",
help=(
"Filter repositories whose identifier / name / path contains this "
"substring (case-insensitive). Use /regex/ for regular expressions."
),
)
subparser.add_argument(
"--tag",
action="append",
default=[],
help=(
"Filter repositories by tag. Matches tags from the repository "
"collector and category tags. Use /regex/ for regular expressions."
),
)
subparser.add_argument(
"--preview",
action="store_true",
help="Preview changes without executing commands",
)
subparser.add_argument(
"--list",
action="store_true",
help="List affected repositories (with preview or status)",
)
subparser.add_argument(
"-a",
"--args",
nargs=argparse.REMAINDER,
dest="extra_args",
help="Additional parameters to be attached.",
default=[],
)
def add_install_update_arguments(subparser: argparse.ArgumentParser) -> None:
"""
Common arguments for install/update commands.
"""
add_identifier_arguments(subparser)
subparser.add_argument(
"-q",
"--quiet",
action="store_true",
help="Suppress warnings and info messages",
)
subparser.add_argument(
"--no-verification",
action="store_true",
default=False,
help="Disable verification via commit/gpg",
)
subparser.add_argument(
"--dependencies",
action="store_true",
help="Also pull and update dependencies",
)
subparser.add_argument(
"--clone-mode",
choices=["ssh", "https", "shallow"],
default="ssh",
help=(
"Specify the clone mode: ssh, https, or shallow "
"(HTTPS shallow clone; default: ssh)"
),
)

View File

@@ -0,0 +1,72 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations
import argparse
from .common import add_identifier_arguments
def add_config_subparsers(
subparsers: argparse._SubParsersAction,
) -> None:
"""
Register config command and its subcommands.
"""
config_parser = subparsers.add_parser(
"config",
help="Manage configuration",
)
config_subparsers = config_parser.add_subparsers(
dest="subcommand",
help="Config subcommands",
required=True,
)
config_show = config_subparsers.add_parser(
"show",
help="Show configuration",
)
add_identifier_arguments(config_show)
config_subparsers.add_parser(
"add",
help="Interactively add a new repository entry",
)
config_subparsers.add_parser(
"edit",
help="Edit configuration file with nano",
)
config_subparsers.add_parser(
"init",
help="Initialize user configuration by scanning the base directory",
)
config_delete = config_subparsers.add_parser(
"delete",
help="Delete repository entry from user config",
)
add_identifier_arguments(config_delete)
config_ignore = config_subparsers.add_parser(
"ignore",
help="Set ignore flag for repository entries in user config",
)
add_identifier_arguments(config_ignore)
config_ignore.add_argument(
"--set",
choices=["true", "false"],
required=True,
help="Set ignore to true or false",
)
config_subparsers.add_parser(
"update",
help=(
"Update default config files in ~/.config/pkgmgr/ from the "
"installed pkgmgr package (does not touch config.yaml)."
),
)

View File

@@ -0,0 +1,44 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations
import argparse
from .common import add_install_update_arguments, add_identifier_arguments
def add_install_update_subparsers(
subparsers: argparse._SubParsersAction,
) -> None:
"""
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)
update_parser = subparsers.add_parser(
"update",
help="Update (pull + install) repository/repositories",
)
add_install_update_arguments(update_parser)
update_parser.add_argument(
"--system",
action="store_true",
help="Include system update commands",
)
deinstall_parser = subparsers.add_parser(
"deinstall",
help="Remove alias links to repository/repositories",
)
add_identifier_arguments(deinstall_parser)
delete_parser = subparsers.add_parser(
"delete",
help="Delete repository/repositories alias links to executables",
)
add_identifier_arguments(delete_parser)

View File

@@ -0,0 +1,38 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations
import argparse
from .common import add_identifier_arguments
def add_list_subparser(
subparsers: argparse._SubParsersAction,
) -> None:
"""
Register the list command.
"""
list_parser = subparsers.add_parser(
"list",
help="List all repositories with details and status",
)
add_identifier_arguments(list_parser)
list_parser.add_argument(
"--status",
type=str,
default="",
help=(
"Filter repositories by status (case insensitive). "
"Use /regex/ for regular expressions."
),
)
list_parser.add_argument(
"--description",
action="store_true",
help=(
"Show an additional detailed section per repository "
"(description, homepage, tags, categories, paths)."
),
)

View File

@@ -0,0 +1,38 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations
import argparse
from .common import add_identifier_arguments
def add_make_subparsers(
subparsers: argparse._SubParsersAction,
) -> None:
"""
Register make command and its subcommands.
"""
make_parser = subparsers.add_parser(
"make",
help="Executes make commands",
)
add_identifier_arguments(make_parser)
make_subparsers = make_parser.add_subparsers(
dest="subcommand",
help="Make subcommands",
required=True,
)
make_install = make_subparsers.add_parser(
"install",
help="Executes the make install command",
)
add_identifier_arguments(make_install)
make_deinstall = make_subparsers.add_parser(
"deinstall",
help="Executes the make deinstall command",
)
add_identifier_arguments(make_deinstall)

View File

@@ -0,0 +1,110 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations
import argparse
from .common import add_identifier_arguments
def add_mirror_subparsers(
subparsers: argparse._SubParsersAction,
) -> None:
"""
Register mirror command and its subcommands (list, diff, merge, setup).
"""
mirror_parser = subparsers.add_parser(
"mirror",
help="Mirror-related utilities (list, diff, merge, setup)",
)
mirror_subparsers = mirror_parser.add_subparsers(
dest="subcommand",
help="Mirror subcommands",
required=True,
)
# ------------------------------------------------------------------
# mirror list
# ------------------------------------------------------------------
mirror_list = mirror_subparsers.add_parser(
"list",
help="List configured mirrors for repositories",
)
add_identifier_arguments(mirror_list)
mirror_list.add_argument(
"--source",
choices=["all", "config", "file", "resolved"],
default="all",
help="Which mirror source to show.",
)
# ------------------------------------------------------------------
# mirror diff
# ------------------------------------------------------------------
mirror_diff = mirror_subparsers.add_parser(
"diff",
help="Show differences between config mirrors and MIRRORS file",
)
add_identifier_arguments(mirror_diff)
# ------------------------------------------------------------------
# mirror merge {config,file} {config,file}
# ------------------------------------------------------------------
mirror_merge = mirror_subparsers.add_parser(
"merge",
help=(
"Merge mirrors between config and MIRRORS file "
"(example: pkgmgr mirror merge config file --all)"
),
)
# First define merge direction positionals, then selection args.
mirror_merge.add_argument(
"source",
choices=["config", "file"],
help="Source of mirrors.",
)
mirror_merge.add_argument(
"target",
choices=["config", "file"],
help="Target of mirrors.",
)
# Selection / filter / preview arguments
add_identifier_arguments(mirror_merge)
mirror_merge.add_argument(
"--config-path",
help=(
"Path to the user config file to update. "
"If omitted, the global config path is used."
),
)
# Note: --preview, --all, --category, --tag, --list, etc. are provided
# by add_identifier_arguments().
# ------------------------------------------------------------------
# mirror setup
# ------------------------------------------------------------------
mirror_setup = mirror_subparsers.add_parser(
"setup",
help=(
"Setup mirror configuration for repositories.\n"
" --local → configure local Git (remotes, pushurls)\n"
" --remote → create remote repositories if missing\n"
"Default: both local and remote."
),
)
add_identifier_arguments(mirror_setup)
mirror_setup.add_argument(
"--local",
action="store_true",
help="Only configure the local Git repository.",
)
mirror_setup.add_argument(
"--remote",
action="store_true",
help="Only operate on remote repositories.",
)
# Note: --preview also comes from add_identifier_arguments().

View File

@@ -0,0 +1,56 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations
import argparse
from .common import add_identifier_arguments
def add_navigation_subparsers(
subparsers: argparse._SubParsersAction,
) -> None:
"""
Register path / explore / terminal / code / shell commands.
"""
path_parser = subparsers.add_parser(
"path",
help="Print the path(s) of repository/repositories",
)
add_identifier_arguments(path_parser)
explore_parser = subparsers.add_parser(
"explore",
help="Open repository in Nautilus file manager",
)
add_identifier_arguments(explore_parser)
terminal_parser = subparsers.add_parser(
"terminal",
help="Open repository in a new GNOME Terminal tab",
)
add_identifier_arguments(terminal_parser)
code_parser = subparsers.add_parser(
"code",
help="Open repository workspace with VS Code",
)
add_identifier_arguments(code_parser)
shell_parser = subparsers.add_parser(
"shell",
help="Execute a shell command in each repository",
)
add_identifier_arguments(shell_parser)
shell_parser.add_argument(
"-c",
"--command",
nargs=argparse.REMAINDER,
dest="shell_command",
help=(
"The shell command (and its arguments) to execute in each "
"repository"
),
default=[],
)

View File

@@ -0,0 +1,57 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations
import argparse
from .common import add_identifier_arguments
def add_release_subparser(
subparsers: argparse._SubParsersAction,
) -> None:
"""
Register the release command.
"""
release_parser = subparsers.add_parser(
"release",
help=(
"Create a release for repository/ies by incrementing version "
"and updating the changelog."
),
)
release_parser.add_argument(
"release_type",
choices=["major", "minor", "patch"],
help="Type of version increment for the release (major, minor, patch).",
)
release_parser.add_argument(
"-m",
"--message",
default=None,
help=(
"Optional release message to add to the changelog and tag."
),
)
# Generic selection / preview / list / extra_args
add_identifier_arguments(release_parser)
# Close current branch after successful release
release_parser.add_argument(
"--close",
action="store_true",
help=(
"Close the current branch after a successful release in each "
"repository, if it is not main/master."
),
)
# Force: skip preview+confirmation and run release directly
release_parser.add_argument(
"-f",
"--force",
action="store_true",
help=(
"Skip the interactive preview+confirmation step and run the "
"release directly."
),
)

View File

@@ -0,0 +1,25 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations
import argparse
from .common import add_identifier_arguments
def add_version_subparser(
subparsers: argparse._SubParsersAction,
) -> None:
"""
Register the version command.
"""
version_parser = subparsers.add_parser(
"version",
help=(
"Show version information for repository/ies "
"(git tags, pyproject.toml, flake.nix, PKGBUILD, debian, spec, "
"Ansible Galaxy)."
),
)
add_identifier_arguments(version_parser)

View File

@@ -0,0 +1,143 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
E2E integration tests for the `pkgmgr mirror` command family.
This test class covers:
- pkgmgr mirror --help
- pkgmgr mirror list --preview --all
- pkgmgr mirror diff --preview --all
- pkgmgr mirror merge config file --preview --all
- pkgmgr mirror setup --preview --all
All of these subcommands are fully wired at CLI level and do not
require mocks. With --preview, merge and setup do not perform
destructive actions, making them safe for CI execution.
"""
from __future__ import annotations
import io
import runpy
import sys
import unittest
from contextlib import redirect_stdout, redirect_stderr
class TestIntegrationMirrorCommands(unittest.TestCase):
"""
E2E tests for `pkgmgr mirror` commands.
"""
# ------------------------------------------------------------
# Helper
# ------------------------------------------------------------
def _run_pkgmgr(self, args: list[str]) -> str:
"""
Execute pkgmgr with the given arguments and return captured stdout+stderr.
- Treat SystemExit(0) or SystemExit(None) as success.
- Convert non-zero exit codes into AssertionError.
"""
original_argv = list(sys.argv)
buffer = io.StringIO()
cmd_repr = "pkgmgr " + " ".join(args)
try:
sys.argv = ["pkgmgr"] + args
try:
with redirect_stdout(buffer), redirect_stderr(buffer):
runpy.run_module("main", run_name="__main__")
except SystemExit as exc:
code = exc.code if isinstance(exc.code, int) else None
if code not in (0, None):
raise AssertionError(
f"{cmd_repr!r} failed with exit code {exc.code}. "
"Scroll up to inspect the pkgmgr output."
) from exc
return buffer.getvalue()
finally:
sys.argv = original_argv
# ------------------------------------------------------------
# Tests
# ------------------------------------------------------------
def test_mirror_help(self) -> None:
"""
Ensure `pkgmgr mirror --help` runs successfully
and prints a usage message for the mirror command.
"""
output = self._run_pkgmgr(["mirror", "--help"])
self.assertIn("usage:", output)
self.assertIn("pkgmgr mirror", output)
def test_mirror_list_preview_all(self) -> None:
"""
`pkgmgr mirror list --preview --all` should run without error
and produce some output for the selected repositories.
"""
output = self._run_pkgmgr(["mirror", "list", "--preview", "--all"])
# Do not assert specific wording; just ensure something was printed.
self.assertTrue(
output.strip(),
msg="Expected `pkgmgr mirror list --preview --all` to produce output.",
)
def test_mirror_diff_preview_all(self) -> None:
"""
`pkgmgr mirror diff --preview --all` should run without error
and produce some diagnostic output (diff header, etc.).
"""
output = self._run_pkgmgr(["mirror", "diff", "--preview", "--all"])
self.assertTrue(
output.strip(),
msg="Expected `pkgmgr mirror diff --preview --all` to produce output.",
)
def test_mirror_merge_config_to_file_preview_all(self) -> None:
"""
`pkgmgr mirror merge config file --preview --all` should run without error.
In preview mode this does not change either config or MIRRORS files;
it only prints what would be merged.
"""
output = self._run_pkgmgr(
[
"mirror",
"merge",
"config",
"file",
"--preview",
"--all",
]
)
self.assertTrue(
output.strip(),
msg=(
"Expected `pkgmgr mirror merge config file --preview --all` "
"to produce output."
),
)
def test_mirror_setup_preview_all(self) -> None:
"""
`pkgmgr mirror setup --preview --all` should run without error.
In preview mode only the intended Git operations and remote
suggestions are printed; no real changes are made.
"""
output = self._run_pkgmgr(["mirror", "setup", "--preview", "--all"])
self.assertTrue(
output.strip(),
msg="Expected `pkgmgr mirror setup --preview --all` to produce output.",
)
if __name__ == "__main__":
unittest.main()

View File

@@ -1,140 +1,206 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import os
import unittest
from unittest import mock
from unittest.mock import MagicMock, patch
"""
Unit tests for NixFlakeInstaller using unittest (no pytest).
Covers:
- Successful installation (exit_code == 0)
- Mandatory failure → SystemExit with correct code
- Optional failure (pkgmgr default) → no raise, but warning
- supports() behavior incl. PKGMGR_DISABLE_NIX_FLAKE_INSTALLER
"""
import io
import os
import shutil
import tempfile
import unittest
from contextlib import redirect_stdout
from unittest.mock import patch
from pkgmgr.actions.install.context import RepoContext
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):
self.identifier = identifier
self.repo_dir = repo_dir
self.preview = preview
class TestNixFlakeInstaller(unittest.TestCase):
def setUp(self) -> None:
self.repo = {"repository": "package-manager"}
# Important: identifier "pkgmgr" triggers both "pkgmgr" and "default"
self.ctx = RepoContext(
repo=self.repo,
identifier="pkgmgr",
repo_dir="/tmp/repo",
repositories_base_dir="/tmp",
bin_dir="/bin",
all_repos=[self.repo],
no_verification=False,
preview=False,
quiet=False,
clone_mode="ssh",
update_dependencies=False,
)
self.installer = NixFlakeInstaller()
# Create a temporary repository directory with a flake.nix file
self._tmpdir = tempfile.mkdtemp(prefix="nix_flake_test_")
self.repo_dir = self._tmpdir
flake_path = os.path.join(self.repo_dir, "flake.nix")
with open(flake_path, "w", encoding="utf-8") as f:
f.write("{}\n")
@patch("pkgmgr.actions.install.installers.nix_flake.os.path.exists")
@patch("pkgmgr.actions.install.installers.nix_flake.shutil.which")
def test_supports_true_when_nix_and_flake_exist(
self,
mock_which: MagicMock,
mock_exists: MagicMock,
) -> None:
mock_which.return_value = "/usr/bin/nix"
mock_exists.return_value = True
# Ensure the disable env var is not set by default
os.environ.pop("PKGMGR_DISABLE_NIX_FLAKE_INSTALLER", None)
with patch.dict(os.environ, {"PKGMGR_DISABLE_NIX_FLAKE_INSTALLER": ""}, clear=False):
self.assertTrue(self.installer.supports(self.ctx))
def tearDown(self) -> None:
# Cleanup temporary directory
if os.path.isdir(self._tmpdir):
shutil.rmtree(self._tmpdir, ignore_errors=True)
mock_which.assert_called_once_with("nix")
mock_exists.assert_called_once_with(
os.path.join(self.ctx.repo_dir, self.installer.FLAKE_FILE)
)
def _enable_nix_in_module(self, which_patch):
"""Ensure shutil.which('nix') in nix_flake module returns a path."""
which_patch.return_value = "/usr/bin/nix"
@patch("pkgmgr.actions.install.installers.nix_flake.os.path.exists")
@patch("pkgmgr.actions.install.installers.nix_flake.shutil.which")
def test_supports_false_when_nix_missing(
self,
mock_which: MagicMock,
mock_exists: MagicMock,
) -> None:
mock_which.return_value = None
mock_exists.return_value = True # flake exists but nix is missing
with patch.dict(os.environ, {"PKGMGR_DISABLE_NIX_FLAKE_INSTALLER": ""}, clear=False):
self.assertFalse(self.installer.supports(self.ctx))
@patch("pkgmgr.actions.install.installers.nix_flake.os.path.exists")
@patch("pkgmgr.actions.install.installers.nix_flake.shutil.which")
def test_supports_false_when_disabled_via_env(
self,
mock_which: MagicMock,
mock_exists: MagicMock,
) -> None:
mock_which.return_value = "/usr/bin/nix"
mock_exists.return_value = True
with patch.dict(
os.environ,
{"PKGMGR_DISABLE_NIX_FLAKE_INSTALLER": "1"},
clear=False,
):
self.assertFalse(self.installer.supports(self.ctx))
@patch("pkgmgr.actions.install.installers.nix_flake.NixFlakeInstaller.supports")
@patch("pkgmgr.actions.install.installers.nix_flake.run_command")
def test_run_removes_old_profile_and_installs_outputs(
self,
mock_run_command: MagicMock,
mock_supports: MagicMock,
) -> None:
def test_nix_flake_run_success(self):
"""
run() should:
- remove the old profile
- install both 'pkgmgr' and 'default' outputs for identifier 'pkgmgr'
- call commands in the correct order
When os.system returns a successful exit code, the installer
should report success and not raise.
"""
mock_supports.return_value = True
ctx = DummyCtx(identifier="some-lib", repo_dir=self.repo_dir)
commands: list[str] = []
installer = NixFlakeInstaller()
def side_effect(cmd: str, cwd: str | None = None, preview: bool = False, **_: object) -> None:
commands.append(cmd)
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):
self._enable_nix_in_module(which_mock)
mock_run_command.side_effect = side_effect
# Simulate os.system returning success (exit code 0)
system_mock.return_value = 0
with patch.dict(os.environ, {"PKGMGR_DISABLE_NIX_FLAKE_INSTALLER": ""}, clear=False):
self.installer.run(self.ctx)
# Sanity: supports() must be True
self.assertTrue(installer.supports(ctx))
remove_cmd = f"nix profile remove {self.installer.PROFILE_NAME} || true"
install_pkgmgr_cmd = f"nix profile install {self.ctx.repo_dir}#pkgmgr"
install_default_cmd = f"nix profile install {self.ctx.repo_dir}#default"
installer.run(ctx)
self.assertIn(remove_cmd, commands)
self.assertIn(install_pkgmgr_cmd, commands)
self.assertIn(install_default_cmd, commands)
out = buf.getvalue()
self.assertIn("[INFO] Running: nix profile install", out)
self.assertIn("Nix flake output 'default' successfully installed.", out)
self.assertEqual(commands[0], remove_cmd)
@patch("pkgmgr.actions.install.installers.nix_flake.shutil.which")
@patch("pkgmgr.actions.install.installers.nix_flake.run_command")
def test_ensure_old_profile_removed_ignores_systemexit(
self,
mock_run_command: MagicMock,
mock_which: MagicMock,
) -> None:
mock_which.return_value = "/usr/bin/nix"
def side_effect(cmd: str, cwd: str | None = None, preview: bool = False, **_: object) -> None:
raise SystemExit(1)
mock_run_command.side_effect = side_effect
self.installer._ensure_old_profile_removed(self.ctx)
remove_cmd = f"nix profile remove {self.installer.PROFILE_NAME} || true"
mock_run_command.assert_called_with(
remove_cmd,
cwd=self.ctx.repo_dir,
preview=self.ctx.preview,
# Ensure the nix command was actually invoked
system_mock.assert_called_with(
f"nix profile install {self.repo_dir}#default"
)
def test_nix_flake_run_mandatory_failure_raises(self):
"""
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).
"""
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):
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
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)
def test_nix_flake_run_optional_failure_does_not_raise(self):
"""
For the package-manager repository, the 'default' output is optional.
Failure to install it must not raise, but should log a warning instead.
"""
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):
self._enable_nix_in_module(which_mock)
self.assertTrue(installer.supports(ctx))
# Optional failure must NOT raise
installer.run(ctx)
out = buf.getvalue()
# Both outputs should have been mentioned
self.assertIn(
"attempting to install profile outputs: pkgmgr, default", out
)
# First output ("pkgmgr") succeeded
self.assertIn(
"Nix flake 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,
)
# 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],
)
def test_nix_flake_supports_respects_disable_env(self):
"""
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)
installer = NixFlakeInstaller()
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))
if __name__ == "__main__":
unittest.main()

View File

@@ -0,0 +1,110 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations
import unittest
from pkgmgr.actions.mirror.git_remote import (
build_default_ssh_url,
determine_primary_remote_url,
)
from pkgmgr.actions.mirror.types import MirrorMap, Repository
class TestMirrorGitRemote(unittest.TestCase):
"""
Unit tests for SSH URL and primary remote selection logic.
"""
def test_build_default_ssh_url_without_port(self) -> None:
repo: Repository = {
"provider": "github.com",
"account": "kevinveenbirkenbach",
"repository": "package-manager",
}
url = build_default_ssh_url(repo)
self.assertEqual(
url,
"git@github.com:kevinveenbirkenbach/package-manager.git",
)
def test_build_default_ssh_url_with_port(self) -> None:
repo: Repository = {
"provider": "code.cymais.cloud",
"account": "kevinveenbirkenbach",
"repository": "pkgmgr",
"port": 2201,
}
url = build_default_ssh_url(repo)
self.assertEqual(
url,
"ssh://git@code.cymais.cloud:2201/kevinveenbirkenbach/pkgmgr.git",
)
def test_build_default_ssh_url_missing_fields_returns_none(self) -> None:
repo: Repository = {
"provider": "github.com",
"account": "kevinveenbirkenbach",
# "repository" fehlt absichtlich
}
url = build_default_ssh_url(repo)
self.assertIsNone(url)
def test_determine_primary_remote_url_prefers_origin_in_resolved_mirrors(
self,
) -> None:
repo: Repository = {
"provider": "github.com",
"account": "kevinveenbirkenbach",
"repository": "package-manager",
}
mirrors: MirrorMap = {
"origin": "git@github.com:kevinveenbirkenbach/package-manager.git",
"backup": "ssh://git@git.veen.world:2201/kevinveenbirkenbach/pkgmgr.git",
}
url = determine_primary_remote_url(repo, mirrors)
self.assertEqual(
url,
"git@github.com:kevinveenbirkenbach/package-manager.git",
)
def test_determine_primary_remote_url_uses_any_mirror_if_no_origin(self) -> None:
repo: Repository = {
"provider": "github.com",
"account": "kevinveenbirkenbach",
"repository": "package-manager",
}
mirrors: MirrorMap = {
"backup": "ssh://git@git.veen.world:2201/kevinveenbirkenbach/pkgmgr.git",
"mirror2": "ssh://git@code.cymais.cloud:2201/kevinveenbirkenbach/pkgmgr.git",
}
url = determine_primary_remote_url(repo, mirrors)
# Alphabetisch sortiert: backup, mirror2 → backup gewinnt
self.assertEqual(
url,
"ssh://git@git.veen.world:2201/kevinveenbirkenbach/pkgmgr.git",
)
def test_determine_primary_remote_url_falls_back_to_default_ssh(self) -> None:
repo: Repository = {
"provider": "github.com",
"account": "kevinveenbirkenbach",
"repository": "package-manager",
}
mirrors: MirrorMap = {}
url = determine_primary_remote_url(repo, mirrors)
self.assertEqual(
url,
"git@github.com:kevinveenbirkenbach/package-manager.git",
)
if __name__ == "__main__":
unittest.main()

View File

@@ -0,0 +1,135 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations
import os
import tempfile
import unittest
from pkgmgr.actions.mirror.io import (
load_config_mirrors,
read_mirrors_file,
)
class TestMirrorIO(unittest.TestCase):
"""
Unit tests for pkgmgr.actions.mirror.io helpers.
"""
# ------------------------------------------------------------------
# load_config_mirrors
# ------------------------------------------------------------------
def test_load_config_mirrors_from_dict(self) -> None:
repo = {
"mirrors": {
"origin": "ssh://git@example.com/account/repo.git",
"backup": "ssh://git@backup/account/repo.git",
"empty": "",
"none": None,
}
}
mirrors = load_config_mirrors(repo)
self.assertEqual(
mirrors,
{
"origin": "ssh://git@example.com/account/repo.git",
"backup": "ssh://git@backup/account/repo.git",
},
)
def test_load_config_mirrors_from_list(self) -> None:
repo = {
"mirrors": [
{"name": "origin", "url": "ssh://git@example.com/account/repo.git"},
{"name": "backup", "url": "ssh://git@backup/account/repo.git"},
{"name": "", "url": "ssh://git@invalid/ignored.git"},
{"name": "missing-url"},
"not-a-dict",
]
}
mirrors = load_config_mirrors(repo)
self.assertEqual(
mirrors,
{
"origin": "ssh://git@example.com/account/repo.git",
"backup": "ssh://git@backup/account/repo.git",
},
)
def test_load_config_mirrors_empty_when_missing(self) -> None:
repo = {}
mirrors = load_config_mirrors(repo)
self.assertEqual(mirrors, {})
# ------------------------------------------------------------------
# read_mirrors_file
# ------------------------------------------------------------------
def test_read_mirrors_file_with_named_and_url_only_entries(self) -> None:
"""
Ensure that the MIRRORS file format is parsed correctly:
- 'name url' → exact name
- 'url' → auto name derived from netloc (host[:port]),
with numeric suffix if duplicated.
"""
with tempfile.TemporaryDirectory() as tmpdir:
mirrors_path = os.path.join(tmpdir, "MIRRORS")
content = "\n".join(
[
"# comment",
"",
"origin ssh://git@example.com/account/repo.git",
"https://github.com/kevinveenbirkenbach/package-manager",
"https://github.com/kevinveenbirkenbach/another-repo",
"ssh://git@git.veen.world:2201/kevinveenbirkenbach/pkgmgr.git",
]
)
with open(mirrors_path, "w", encoding="utf-8") as fh:
fh.write(content + "\n")
mirrors = read_mirrors_file(tmpdir)
# 'origin' is preserved as given
self.assertIn("origin", mirrors)
self.assertEqual(
mirrors["origin"],
"ssh://git@example.com/account/repo.git",
)
# Two GitHub URLs → auto names: github.com, github.com2
github_urls = {
mirrors.get("github.com"),
mirrors.get("github.com2"),
}
self.assertIn(
"https://github.com/kevinveenbirkenbach/package-manager",
github_urls,
)
self.assertIn(
"https://github.com/kevinveenbirkenbach/another-repo",
github_urls,
)
# SSH-URL mit User-Teil → netloc ist "git@git.veen.world:2201"
# → host = "git@git.veen.world"
self.assertIn("git@git.veen.world", mirrors)
self.assertEqual(
mirrors["git@git.veen.world"],
"ssh://git@git.veen.world:2201/kevinveenbirkenbach/pkgmgr.git",
)
def test_read_mirrors_file_missing_returns_empty(self) -> None:
with tempfile.TemporaryDirectory() as tmpdir:
mirrors = read_mirrors_file(tmpdir) # no MIRRORS file
self.assertEqual(mirrors, {})
if __name__ == "__main__":
unittest.main()

View File

@@ -0,0 +1,59 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations
import unittest
from unittest.mock import patch
from pkgmgr.actions.mirror.setup_cmd import _probe_mirror
from pkgmgr.core.git import GitError
class TestMirrorSetupCmd(unittest.TestCase):
"""
Unit tests for the non-destructive remote probing logic in setup_cmd.
"""
@patch("pkgmgr.actions.mirror.setup_cmd.run_git")
def test_probe_mirror_success_returns_true_and_empty_message(
self,
mock_run_git,
) -> None:
"""
If run_git returns successfully, _probe_mirror must report (True, "").
"""
mock_run_git.return_value = "dummy-output"
ok, message = _probe_mirror(
"ssh://git@code.cymais.cloud:2201/kevinveenbirkenbach/pkgmgr.git",
"/tmp/some-repo",
)
self.assertTrue(ok)
self.assertEqual(message, "")
mock_run_git.assert_called_once()
@patch("pkgmgr.actions.mirror.setup_cmd.run_git")
def test_probe_mirror_failure_returns_false_and_error_message(
self,
mock_run_git,
) -> None:
"""
If run_git raises GitError, _probe_mirror must report (False, <message>),
and not re-raise the exception.
"""
mock_run_git.side_effect = GitError("Git command failed (simulated)")
ok, message = _probe_mirror(
"ssh://git@code.cymais.cloud:2201/kevinveenbirkenbach/pkgmgr.git",
"/tmp/some-repo",
)
self.assertFalse(ok)
self.assertIn("Git command failed", message)
mock_run_git.assert_called_once()
if __name__ == "__main__":
unittest.main()