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
This commit is contained in:
Kevin Veen-Birkenbach
2025-12-11 17:49:31 +01:00
parent 39b16b87a8
commit f3c5460e48
8 changed files with 485 additions and 104 deletions

View File

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

View File

@@ -1,7 +1,7 @@
from __future__ import annotations from __future__ import annotations
import os import os
from typing import List, Optional from typing import List, Optional, Set
from pkgmgr.core.command.run import run_command from pkgmgr.core.command.run import run_command
from pkgmgr.core.git import GitError, run_git from pkgmgr.core.git import GitError, run_git
@@ -87,18 +87,41 @@ def has_origin_remote(repo_dir: str) -> bool:
return "origin" in names 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( def ensure_origin_remote(
repo: Repository, repo: Repository,
ctx: RepoMirrorContext, ctx: RepoMirrorContext,
preview: bool, preview: bool,
) -> None: ) -> None:
""" """
Ensure that a usable 'origin' remote exists. Ensure that a usable 'origin' remote exists and has all push URLs.
Priority for choosing URL:
1. resolved_mirrors["origin"]
2. any resolved mirror (first by name)
3. default SSH URL derived from provider/account/repository
""" """
repo_dir = ctx.repo_dir repo_dir = ctx.repo_dir
resolved_mirrors = ctx.resolved_mirrors resolved_mirrors = ctx.resolved_mirrors
@@ -109,6 +132,7 @@ def ensure_origin_remote(
url = determine_primary_remote_url(repo, resolved_mirrors) url = determine_primary_remote_url(repo, resolved_mirrors)
if not has_origin_remote(repo_dir):
if not url: if not url:
print( print(
"[WARN] Could not determine URL for 'origin' remote. " "[WARN] Could not determine URL for 'origin' remote. "
@@ -116,26 +140,40 @@ def ensure_origin_remote(
) )
return return
if not has_origin_remote(repo_dir):
cmd = f"git remote add origin {url}" cmd = f"git remote add origin {url}"
if preview: if preview:
print(f"[PREVIEW] Would run in {repo_dir!r}: {cmd}") print(f"[PREVIEW] Would run in {repo_dir!r}: {cmd}")
else: else:
print(f"[INFO] Adding 'origin' remote in {repo_dir}: {url}") print(f"[INFO] Adding 'origin' remote in {repo_dir}: {url}")
run_command(cmd, cwd=repo_dir, preview=False) 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: else:
current = current_origin_url(repo_dir)
if current == url or not url:
print( print(
f"[INFO] Updating 'origin' remote in {repo_dir} " f"[INFO] 'origin' already points to "
f"from {current or '<unknown>'} to {url}" f"{current or '<unknown>'} (no change needed)."
) )
run_command(cmd, cwd=repo_dir, preview=False) 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,47 +1,28 @@
from __future__ import annotations from __future__ import annotations
import os import os
from urllib.parse import urlparse
from typing import List, Mapping from typing import List, Mapping
from .types import MirrorMap, Repository from .types import MirrorMap, Repository
def load_config_mirrors(repo: Repository) -> MirrorMap: 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 {} mirrors = repo.get("mirrors") or {}
result: MirrorMap = {} result: MirrorMap = {}
if isinstance(mirrors, dict): if isinstance(mirrors, dict):
for name, url in mirrors.items(): for name, url in mirrors.items():
if not url: if url:
continue
result[str(name)] = str(url) result[str(name)] = str(url)
return result return result
if isinstance(mirrors, list): if isinstance(mirrors, list):
for entry in mirrors: for entry in mirrors:
if not isinstance(entry, dict): if isinstance(entry, dict):
continue
name = entry.get("name") name = entry.get("name")
url = entry.get("url") url = entry.get("url")
if not name or not url: if name and url:
continue
result[str(name)] = str(url) result[str(name)] = str(url)
return result return result
@@ -49,13 +30,9 @@ def load_config_mirrors(repo: Repository) -> MirrorMap:
def read_mirrors_file(repo_dir: str, filename: str = "MIRRORS") -> MirrorMap: def read_mirrors_file(repo_dir: str, filename: str = "MIRRORS") -> MirrorMap:
""" """
Read mirrors from the MIRRORS file in the repository directory. Supports:
NAME URL
Simple text format: URL → auto name = hostname
# comment
origin ssh://git@example.com/account/repo.git
backup ssh://git@backup/account/repo.git
""" """
path = os.path.join(repo_dir, filename) path = os.path.join(repo_dir, filename)
mirrors: MirrorMap = {} mirrors: MirrorMap = {}
@@ -71,10 +48,24 @@ def read_mirrors_file(repo_dir: str, filename: str = "MIRRORS") -> MirrorMap:
continue continue
parts = stripped.split(None, 1) parts = stripped.split(None, 1)
if len(parts) != 2:
# Ignore malformed lines silently # Case 1: "name url"
continue if len(parts) == 2:
name, url = parts name, url = parts
# Case 2: "url" → auto-generate name
elif len(parts) == 1:
url = parts[0]
parsed = urlparse(url)
host = (parsed.netloc or "").split(":")[0]
base = host or "mirror"
name = base
i = 2
while name in mirrors:
name = f"{base}{i}"
i += 1
else:
continue
mirrors[name] = url mirrors[name] = url
except OSError as exc: except OSError as exc:
print(f"[WARN] Could not read MIRRORS file at {path}: {exc}") print(f"[WARN] Could not read MIRRORS file at {path}: {exc}")
@@ -88,22 +79,14 @@ def write_mirrors_file(
filename: str = "MIRRORS", filename: str = "MIRRORS",
preview: bool = False, preview: bool = False,
) -> None: ) -> 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) 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 "") content = "\n".join(lines) + ("\n" if lines else "")
if preview: if preview:
print(f"[PREVIEW] Would write MIRRORS file at {path}:") print(f"[PREVIEW] Would write MIRRORS file at {path}:")
if content: print(content or "(empty)")
print(content.rstrip())
else:
print("(empty)")
return return
try: try:

View File

@@ -1,6 +1,8 @@
from __future__ import annotations 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 .context import build_context
from .git_remote import determine_primary_remote_url, ensure_origin_remote 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], all_repos: List[Repository],
preview: bool, preview: bool,
) -> None: ) -> None:
"""
Ensure local Git state is sane (currently: 'origin' remote).
"""
ctx = build_context(repo, repositories_base_dir, all_repos) ctx = build_context(repo, repositories_base_dir, all_repos)
print("------------------------------------------------------------") print("------------------------------------------------------------")
@@ -24,6 +29,27 @@ def _setup_local_mirrors_for_repo(
print() 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( def _setup_remote_mirrors_for_repo(
repo: Repository, repo: Repository,
repositories_base_dir: str, repositories_base_dir: str,
@@ -31,45 +57,75 @@ def _setup_remote_mirrors_for_repo(
preview: bool, preview: bool,
) -> None: ) -> None:
""" """
Placeholder for remote-side setup. Remote-side setup / validation.
This is intentionally conservative: Aktuell werden nur **nicht-destruktive Checks** gemacht:
- We *do not* call any provider APIs automatically here.
- Instead, we show what should exist and which URL should be created. - 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) ctx = build_context(repo, repositories_base_dir, all_repos)
resolved_m = ctx.resolved_mirrors resolved_m = ctx.resolved_mirrors
primary_url = determine_primary_remote_url(repo, resolved_m)
print("------------------------------------------------------------") print("------------------------------------------------------------")
print(f"[MIRROR SETUP:REMOTE] {ctx.identifier}") print(f"[MIRROR SETUP:REMOTE] {ctx.identifier}")
print(f"[MIRROR SETUP:REMOTE] dir: {ctx.repo_dir}") print(f"[MIRROR SETUP:REMOTE] dir: {ctx.repo_dir}")
print("------------------------------------------------------------") print("------------------------------------------------------------")
if not resolved_m:
# Optional: Fallback auf eine heuristisch bestimmte URL, falls wir
# irgendwann "automatisch anlegen" implementieren wollen.
primary_url = determine_primary_remote_url(repo, resolved_m)
if not primary_url: if not primary_url:
print( print(
"[WARN] Could not determine primary remote URL for this repository.\n" "[INFO] No mirrors configured (config or MIRRORS file), and no "
" Please ensure provider/account/repository and/or mirrors " "primary URL could be derived from provider/account/repository."
"are set in your config."
) )
print() print()
return return
if preview: ok, error_message = _probe_mirror(primary_url, ctx.repo_dir)
print( if ok:
"[PREVIEW] Would ensure that a remote repository exists for:\n" print(f"[OK] Remote mirror (primary) is reachable: {primary_url}")
f" {primary_url}\n"
" (Provider-specific API calls not implemented yet.)"
)
else: else:
print( print("[WARN] Primary remote URL is NOT reachable:")
"[INFO] Remote-setup logic is not implemented yet.\n" print(f" {primary_url}")
" Please create the remote repository manually if needed:\n" if error_message:
f" {primary_url}\n" print(" Details:")
) for line in error_message.splitlines():
print(f" {line}")
print() print()
print(
"[INFO] Remote checks are non-destructive and only use `git ls-remote` "
"to probe mirror URLs."
)
print()
return
# Normaler Fall: wir haben benannte Mirrors aus config/MIRRORS
for name, url in sorted(resolved_m.items()):
ok, error_message = _probe_mirror(url, ctx.repo_dir)
if ok:
print(f"[OK] Remote mirror '{name}' is reachable: {url}")
else:
print(f"[WARN] Remote mirror '{name}' is NOT reachable:")
print(f" {url}")
if error_message:
print(" Details:")
for line in error_message.splitlines():
print(f" {line}")
print()
print(
"[INFO] Remote checks are non-destructive and only use `git ls-remote` "
"to probe mirror URLs."
)
print()
def setup_mirrors( def setup_mirrors(
@@ -88,8 +144,8 @@ def setup_mirrors(
points to a reasonable URL). points to a reasonable URL).
remote: remote:
- Placeholder that prints what should exist on the remote side. - Non-destructive remote checks using `git ls-remote` for each mirror URL.
Actual API calls to providers are not implemented yet. Es werden keine Repositories auf dem Provider angelegt.
""" """
for repo in selected_repos: for repo in selected_repos:
if local: if local:

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