Compare commits
9 Commits
39b16b87a8
...
stable
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
fcf9d4b59b | ||
|
|
b483dbfaad | ||
|
|
9630917570 | ||
|
|
6a4432dd04 | ||
|
|
cfb91d825a | ||
|
|
a3b21f23fc | ||
|
|
e49dd85200 | ||
|
|
c9dec5ecd6 | ||
|
|
f3c5460e48 |
47
CHANGELOG.md
47
CHANGELOG.md
@@ -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 Debian’s 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.
|
||||
|
||||
6
MIRRORS
6
MIRRORS
@@ -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
|
||||
@@ -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 = ./.;
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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}'."
|
||||
)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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()
|
||||
|
||||
0
tests/unit/pkgmgr/actions/mirror/__init__.py
Normal file
0
tests/unit/pkgmgr/actions/mirror/__init__.py
Normal file
110
tests/unit/pkgmgr/actions/mirror/test_git_remote.py
Normal file
110
tests/unit/pkgmgr/actions/mirror/test_git_remote.py
Normal 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()
|
||||
135
tests/unit/pkgmgr/actions/mirror/test_io.py
Normal file
135
tests/unit/pkgmgr/actions/mirror/test_io.py
Normal 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()
|
||||
59
tests/unit/pkgmgr/actions/mirror/test_setup_cmd.py
Normal file
59
tests/unit/pkgmgr/actions/mirror/test_setup_cmd.py
Normal 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()
|
||||
Reference in New Issue
Block a user