Compare commits

...

9 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
17 changed files with 821 additions and 261 deletions

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,3 +1,3 @@
https://github.com/kevinveenbirkenbach/package-manager
https://git.veen.world/kevinveenbirkenbach/package-manager
https://code.infinito.nexus/kevinveenbirkenbach/package-manager
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

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

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

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

@@ -1,7 +1,7 @@
from __future__ import annotations
import os
from typing import List, Optional
from typing import List, Optional, Set
from pkgmgr.core.command.run import run_command
from pkgmgr.core.git import GitError, run_git
@@ -87,18 +87,41 @@ def has_origin_remote(repo_dir: str) -> bool:
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.
Priority for choosing URL:
1. resolved_mirrors["origin"]
2. any resolved mirror (first by name)
3. default SSH URL derived from provider/account/repository
Ensure that a usable 'origin' remote exists and has all push URLs.
"""
repo_dir = ctx.repo_dir
resolved_mirrors = ctx.resolved_mirrors
@@ -109,33 +132,48 @@ def ensure_origin_remote(
url = determine_primary_remote_url(repo, resolved_mirrors)
if not url:
print(
"[WARN] Could not determine URL for 'origin' remote. "
"Please configure mirrors or provider/account/repository."
)
return
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)
return
current = current_origin_url(repo_dir)
if current == url:
print(f"[INFO] 'origin' already points to {url} (no change needed).")
return
cmd = f"git remote set-url origin {url}"
if preview:
print(f"[PREVIEW] Would run in {repo_dir!r}: {cmd}")
else:
print(
f"[INFO] Updating 'origin' remote in {repo_dir} "
f"from {current or '<unknown>'} to {url}"
)
run_command(cmd, cwd=repo_dir, preview=False)
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

@@ -1,61 +1,38 @@
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:
"""
Load mirrors from the repository configuration entry.
Supported shapes:
repo["mirrors"] = {
"origin": "ssh://git@example.com/...",
"backup": "ssh://git@backup/...",
}
or
repo["mirrors"] = [
{"name": "origin", "url": "ssh://git@example.com/..."},
{"name": "backup", "url": "ssh://git@backup/..."},
]
"""
mirrors = repo.get("mirrors") or {}
result: MirrorMap = {}
if isinstance(mirrors, dict):
for name, url in mirrors.items():
if not url:
continue
result[str(name)] = str(url)
if url:
result[str(name)] = str(url)
return result
if isinstance(mirrors, list):
for entry in mirrors:
if not isinstance(entry, dict):
continue
name = entry.get("name")
url = entry.get("url")
if not name or not url:
continue
result[str(name)] = str(url)
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:
"""
Read mirrors from the MIRRORS file in the repository directory.
Simple text format:
# comment
origin ssh://git@example.com/account/repo.git
backup ssh://git@backup/account/repo.git
Supports:
NAME URL
URL → auto name = hostname
"""
path = os.path.join(repo_dir, filename)
mirrors: MirrorMap = {}
@@ -71,10 +48,24 @@ def read_mirrors_file(repo_dir: str, filename: str = "MIRRORS") -> MirrorMap:
continue
parts = stripped.split(None, 1)
if len(parts) != 2:
# Ignore malformed lines silently
# 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
name, url = parts
mirrors[name] = url
except OSError as exc:
print(f"[WARN] Could not read MIRRORS file at {path}: {exc}")
@@ -88,22 +79,14 @@ def write_mirrors_file(
filename: str = "MIRRORS",
preview: bool = False,
) -> None:
"""
Write mirrors to MIRRORS file.
Existing file is overwritten. In preview mode we only print what would
be written.
"""
path = os.path.join(repo_dir, filename)
lines: List[str] = [f"{name} {url}" for name, url in sorted(mirrors.items())]
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}:")
if content:
print(content.rstrip())
else:
print("(empty)")
print(content or "(empty)")
return
try:

View File

@@ -1,6 +1,8 @@
from __future__ import annotations
from typing import List
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
@@ -13,6 +15,9 @@ def _setup_local_mirrors_for_repo(
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("------------------------------------------------------------")
@@ -24,6 +29,27 @@ def _setup_local_mirrors_for_repo(
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,
@@ -31,45 +57,75 @@ def _setup_remote_mirrors_for_repo(
preview: bool,
) -> None:
"""
Placeholder for remote-side setup.
Remote-side setup / validation.
This is intentionally conservative:
- We *do not* call any provider APIs automatically here.
- Instead, we show what should exist and which URL should be created.
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
primary_url = determine_primary_remote_url(repo, resolved_m)
print("------------------------------------------------------------")
print(f"[MIRROR SETUP:REMOTE] {ctx.identifier}")
print(f"[MIRROR SETUP:REMOTE] dir: {ctx.repo_dir}")
print("------------------------------------------------------------")
if not primary_url:
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(
"[WARN] Could not determine primary remote URL for this repository.\n"
" Please ensure provider/account/repository and/or mirrors "
"are set in your config."
"[INFO] Remote checks are non-destructive and only use `git ls-remote` "
"to probe mirror URLs."
)
print()
return
if preview:
print(
"[PREVIEW] Would ensure that a remote repository exists for:\n"
f" {primary_url}\n"
" (Provider-specific API calls not implemented yet.)"
)
else:
print(
"[INFO] Remote-setup logic is not implemented yet.\n"
" Please create the remote repository manually if needed:\n"
f" {primary_url}\n"
)
# 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(
@@ -88,8 +144,8 @@ def setup_mirrors(
points to a reasonable URL).
remote:
- Placeholder that prints what should exist on the remote side.
Actual API calls to providers are not implemented yet.
- 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:

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