diff --git a/src/pkgmgr/actions/mirror/remote_provision.py b/src/pkgmgr/actions/mirror/remote_provision.py index 75fadf5..a7faf37 100644 --- a/src/pkgmgr/actions/mirror/remote_provision.py +++ b/src/pkgmgr/actions/mirror/remote_provision.py @@ -11,35 +11,37 @@ from .types import Repository from .url_utils import normalize_provider_host, parse_repo_from_git_url -def ensure_remote_repository( - repo: Repository, - repositories_base_dir: str, - all_repos: List[Repository], +def _provider_hint_from_host(host: str) -> str | None: + h = (host or "").lower() + if h == "github.com": + return "github" + # Best-effort default for self-hosted git domains + return "gitea" if h else None + + +def ensure_remote_repository_for_url( + *, + url: str, + private_default: bool, + description: str, preview: bool, ) -> None: - ctx = build_context(repo, repositories_base_dir, all_repos) - - primary_url = determine_primary_remote_url(repo, ctx) - if not primary_url: - print("[INFO] No primary URL found; skipping remote provisioning.") - return - - host_raw, owner, name = parse_repo_from_git_url(primary_url) + host_raw, owner, name = parse_repo_from_git_url(url) host = normalize_provider_host(host_raw) if not host or not owner or not name: - print("[WARN] Could not parse remote URL:", primary_url) + print(f"[WARN] Could not parse repo from URL: {url}") return spec = RepoSpec( host=host, owner=owner, name=name, - private=bool(repo.get("private", True)), - description=str(repo.get("description", "")), + private=private_default, + description=description, ) - provider_kind = str(repo.get("provider", "")).lower() or None + provider_kind = _provider_hint_from_host(host) try: result = ensure_remote_repo( @@ -56,4 +58,29 @@ def ensure_remote_repository( if result.url: print(f"[REMOTE ENSURE] URL: {result.url}") except Exception as exc: # noqa: BLE001 - print(f"[ERROR] Remote provisioning failed: {exc}") + print(f"[ERROR] Remote provisioning failed for {url!r}: {exc}") + + +def ensure_remote_repository( + repo: Repository, + repositories_base_dir: str, + all_repos: List[Repository], + preview: bool, +) -> None: + """ + Backwards-compatible wrapper: ensure the *primary* remote repository + derived from the primary URL. + """ + ctx = build_context(repo, repositories_base_dir, all_repos) + + primary_url = determine_primary_remote_url(repo, ctx) + if not primary_url: + print("[INFO] No primary URL found; skipping remote provisioning.") + return + + ensure_remote_repository_for_url( + url=primary_url, + private_default=bool(repo.get("private", True)), + description=str(repo.get("description", "")), + preview=preview, + ) diff --git a/src/pkgmgr/actions/mirror/setup_cmd.py b/src/pkgmgr/actions/mirror/setup_cmd.py index 8f3057f..adc641e 100644 --- a/src/pkgmgr/actions/mirror/setup_cmd.py +++ b/src/pkgmgr/actions/mirror/setup_cmd.py @@ -2,11 +2,11 @@ from __future__ import annotations from typing import List -from pkgmgr.core.git.queries import probe_remote_reachable +from pkgmgr.core.git.queries import probe_remote_reachable_detail from .context import build_context -from .git_remote import ensure_origin_remote, determine_primary_remote_url -from .remote_provision import ensure_remote_repository +from .git_remote import determine_primary_remote_url, ensure_origin_remote +from .remote_provision import ensure_remote_repository_for_url from .types import Repository @@ -25,6 +25,25 @@ def _is_git_remote_url(url: str) -> bool: return False +def _print_probe_result(name: str | None, url: str, *, cwd: str) -> None: + """ + Print probe result for a git remote URL, including a short failure reason. + """ + ok, reason = probe_remote_reachable_detail(url, cwd=cwd) + + prefix = f"{name}: " if name else "" + if ok: + print(f"[OK] {prefix}{url}") + return + + print(f"[WARN] {prefix}{url}") + if reason: + reason = reason.strip() + if len(reason) > 240: + reason = reason[:240].rstrip() + "…" + print(f" reason: {reason}") + + def _setup_local_mirrors_for_repo( repo: Repository, repositories_base_dir: str, @@ -56,35 +75,47 @@ def _setup_remote_mirrors_for_repo( print(f"[MIRROR SETUP:REMOTE] dir: {ctx.repo_dir}") print("------------------------------------------------------------") - if ensure_remote: - ensure_remote_repository( - repo, - repositories_base_dir, - all_repos, - preview, - ) - - # Probe only git URLs (do not try ls-remote against PyPI etc.) - # If there are no mirrors at all, probe the primary git URL. git_mirrors = { k: v for k, v in ctx.resolved_mirrors.items() if _is_git_remote_url(v) } + # If there are no git mirrors, fall back to primary (git) URL. if not git_mirrors: primary = determine_primary_remote_url(repo, ctx) if not primary or not _is_git_remote_url(primary): - print("[INFO] No git mirrors to probe.") + print("[INFO] No git mirrors to probe or provision.") print() return - ok = probe_remote_reachable(primary, cwd=ctx.repo_dir) - print("[OK]" if ok else "[WARN]", primary) + if ensure_remote: + print(f"[REMOTE ENSURE] ensuring primary: {primary}") + ensure_remote_repository_for_url( + url=primary, + private_default=bool(repo.get("private", True)), + description=str(repo.get("description", "")), + preview=preview, + ) + print() + + _print_probe_result(None, primary, cwd=ctx.repo_dir) print() return + # Provision ALL git mirrors (if requested) + if ensure_remote: + for name, url in git_mirrors.items(): + print(f"[REMOTE ENSURE] ensuring mirror {name!r}: {url}") + ensure_remote_repository_for_url( + url=url, + private_default=bool(repo.get("private", True)), + description=str(repo.get("description", "")), + preview=preview, + ) + print() + + # Probe ALL git mirrors for name, url in git_mirrors.items(): - ok = probe_remote_reachable(url, cwd=ctx.repo_dir) - print(f"[OK] {name}: {url}" if ok else f"[WARN] {name}: {url}") + _print_probe_result(name, url, cwd=ctx.repo_dir) print() diff --git a/src/pkgmgr/core/git/queries/__init__.py b/src/pkgmgr/core/git/queries/__init__.py index 96bb7d4..ad75585 100644 --- a/src/pkgmgr/core/git/queries/__init__.py +++ b/src/pkgmgr/core/git/queries/__init__.py @@ -20,7 +20,10 @@ from .get_tags_at_ref import GitTagsAtRefQueryError, get_tags_at_ref from .get_upstream_ref import get_upstream_ref from .list_remotes import list_remotes from .list_tags import list_tags -from .probe_remote_reachable import probe_remote_reachable +from .probe_remote_reachable import ( + probe_remote_reachable, + probe_remote_reachable_detail, +) from .resolve_base_branch import GitBaseBranchNotFoundError, resolve_base_branch __all__ = [ @@ -37,6 +40,7 @@ __all__ = [ "list_remotes", "get_remote_push_urls", "probe_remote_reachable", + "probe_remote_reachable_detail", "get_changelog", "GitChangelogQueryError", "get_tags_at_ref", diff --git a/src/pkgmgr/core/git/queries/probe_remote_reachable.py b/src/pkgmgr/core/git/queries/probe_remote_reachable.py index a6c87cf..350a0d2 100644 --- a/src/pkgmgr/core/git/queries/probe_remote_reachable.py +++ b/src/pkgmgr/core/git/queries/probe_remote_reachable.py @@ -1,21 +1,121 @@ from __future__ import annotations +from typing import Tuple + from ..errors import GitRunError from ..run import run -def probe_remote_reachable(url: str, cwd: str = ".") -> bool: +def _first_useful_line(text: str) -> str: + lines: list[str] = [] + for line in (text or "").splitlines(): + s = line.strip() + if s: + lines.append(s) + + if not lines: + return "" + + preferred_keywords = ( + "fatal:", + "permission denied", + "repository not found", + "could not read from remote repository", + "connection refused", + "connection timed out", + "no route to host", + "name or service not known", + "temporary failure in name resolution", + "host key verification failed", + "could not resolve hostname", + "authentication failed", + "publickey", + "the authenticity of host", + "known_hosts", + ) + for s in lines: + low = s.lower() + if any(k in low for k in preferred_keywords): + return s + + # Avoid returning a meaningless "error:" if possible + for s in lines: + if s.lower() not in ("error:", "error"): + return s + + return lines[0] + + +def _looks_like_real_transport_error(text: str) -> bool: """ - Check whether a remote URL is reachable. + True if stderr/stdout contains strong indicators that the remote is NOT usable. + """ + low = (text or "").lower() + indicators = ( + "repository not found", + "could not read from remote repository", + "permission denied", + "authentication failed", + "publickey", + "host key verification failed", + "could not resolve hostname", + "name or service not known", + "connection refused", + "connection timed out", + "no route to host", + ) + return any(i in low for i in indicators) - Equivalent to: - git ls-remote --exit-code - Returns: - True if reachable, False otherwise. +def _format_reason(exc: GitRunError, *, url: str) -> str: + stderr = getattr(exc, "stderr", "") or "" + stdout = getattr(exc, "stdout", "") or "" + rc = getattr(exc, "returncode", None) + + reason = ( + _first_useful_line(stderr) + or _first_useful_line(stdout) + or _first_useful_line(str(exc)) + ) + + if rc is not None: + reason = f"(exit {rc}) {reason}".strip() if reason else f"(exit {rc})" + + # If we still have nothing useful, provide a hint to debug SSH transport + if not reason or reason.lower() in ("(exit 2)", "(exit 128)"): + reason = ( + f"{reason} | hint: run " + f"GIT_SSH_COMMAND='ssh -vvv' git ls-remote --exit-code {url!r}" + ).strip() + + return reason.strip() + + +def probe_remote_reachable_detail(url: str, cwd: str = ".") -> Tuple[bool, str]: + """ + Probe whether a remote URL is reachable. + + Implementation detail: + - We run `git ls-remote --exit-code `. + - Git may return exit code 2 when the remote is reachable but no refs exist + (e.g. an empty repository). We treat that as reachable. """ try: run(["ls-remote", "--exit-code", url], cwd=cwd) - return True - except GitRunError: - return False + return True, "" + except GitRunError as exc: + rc = getattr(exc, "returncode", None) + stderr = getattr(exc, "stderr", "") or "" + stdout = getattr(exc, "stdout", "") or "" + + # Important: `git ls-remote --exit-code` uses exit code 2 when no refs match. + # For a completely empty repo, this can happen even though auth/transport is OK. + if rc == 2 and not _looks_like_real_transport_error(stderr + "\n" + stdout): + return True, "remote reachable, but no refs found yet (empty repository)" + + return False, _format_reason(exc, url=url) + + +def probe_remote_reachable(url: str, cwd: str = ".") -> bool: + ok, _ = probe_remote_reachable_detail(url, cwd=cwd) + return ok diff --git a/src/pkgmgr/core/git/run.py b/src/pkgmgr/core/git/run.py index 8100e5b..f60c486 100644 --- a/src/pkgmgr/core/git/run.py +++ b/src/pkgmgr/core/git/run.py @@ -42,16 +42,34 @@ def run( ) except subprocess.CalledProcessError as exc: stderr = exc.stderr or "" - if _is_not_repo_error(stderr): - raise GitNotRepositoryError( - f"Not a git repository: {cwd!r}\nCommand: {cmd_str}\nSTDERR:\n{stderr}" - ) from exc + stdout = exc.stdout or "" - raise GitRunError( + if _is_not_repo_error(stderr): + err = GitNotRepositoryError( + f"Not a git repository: {cwd!r}\nCommand: {cmd_str}\nSTDERR:\n{stderr}" + ) + # Attach details for callers who want to debug + err.cwd = cwd + err.cmd = cmd + err.cmd_str = cmd_str + err.returncode = exc.returncode + err.stdout = stdout + err.stderr = stderr + raise err from exc + + err = GitRunError( f"Git command failed in {cwd!r}: {cmd_str}\n" f"Exit code: {exc.returncode}\n" - f"STDOUT:\n{exc.stdout}\n" + f"STDOUT:\n{stdout}\n" f"STDERR:\n{stderr}" - ) from exc + ) + # Attach details for callers who want to debug + err.cwd = cwd + err.cmd = cmd + err.cmd_str = cmd_str + err.returncode = exc.returncode + err.stdout = stdout + err.stderr = stderr + raise err from exc return result.stdout.strip() diff --git a/tests/integration/test_mirror_commands.py b/tests/integration/test_mirror_commands.py index 241ade9..8265ed8 100644 --- a/tests/integration/test_mirror_commands.py +++ b/tests/integration/test_mirror_commands.py @@ -113,17 +113,12 @@ class TestIntegrationMirrorCommands(unittest.TestCase): ) ) - # Deterministic remote probing (new refactor: probe_remote_reachable) + # Deterministic remote probing (refactor: probe_remote_reachable_detail) + # Patch where it is USED (setup_cmd imported it directly). stack.enter_context( _p( - "pkgmgr.core.git.queries.probe_remote_reachable", - return_value=True, - ) - ) - stack.enter_context( - _p( - "pkgmgr.actions.mirror.setup_cmd.probe_remote_reachable", - return_value=True, + "pkgmgr.actions.mirror.setup_cmd.probe_remote_reachable_detail", + return_value=(True, ""), ) ) diff --git a/tests/integration/test_mirror_probe_detail_and_provision.py b/tests/integration/test_mirror_probe_detail_and_provision.py new file mode 100644 index 0000000..1e73b91 --- /dev/null +++ b/tests/integration/test_mirror_probe_detail_and_provision.py @@ -0,0 +1,220 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +""" +Integration test for mirror probing + provisioning after refactor. + +We test the CLI entrypoint `handle_mirror_command()` directly to avoid +depending on repo-selection / config parsing for `--all`. + +Covers: +- setup_cmd uses probe_remote_reachable_detail() +- check prints [OK]/[WARN] and 'reason:' lines for failures +- provision triggers ensure_remote_repo (preview-safe) for each git mirror +""" + +from __future__ import annotations + +import io +import tempfile +import unittest +from contextlib import redirect_stderr, redirect_stdout +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import MagicMock, PropertyMock, patch + +from pkgmgr.cli.commands.mirror import handle_mirror_command + + +class TestIntegrationMirrorProbeDetailAndProvision(unittest.TestCase): + def _make_ctx( + self, *, repositories_base_dir: str, all_repositories: list[dict] + ) -> MagicMock: + ctx = MagicMock() + ctx.repositories_base_dir = repositories_base_dir + ctx.all_repositories = all_repositories + # mirror merge may look at this; keep it present for safety + ctx.user_config_path = str(Path(repositories_base_dir) / "user.yml") + return ctx + + def _make_dummy_repo_ctx(self, *, repo_dir: str) -> MagicMock: + """ + This is the RepoMirrorContext-like object returned by build_context(). + """ + dummy = MagicMock() + dummy.identifier = "dummy-repo" + dummy.repo_dir = repo_dir + dummy.config_mirrors = {"origin": "git@github.com:alice/repo.git"} + dummy.file_mirrors = {"backup": "ssh://git@git.example:2201/alice/repo.git"} + type(dummy).resolved_mirrors = PropertyMock( + return_value={ + "origin": "git@github.com:alice/repo.git", + "backup": "ssh://git@git.example:2201/alice/repo.git", + } + ) + return dummy + + def _run_handle( + self, + *, + subcommand: str, + preview: bool, + selected: list[dict], + dummy_repo_dir: str, + probe_detail_side_effect, + ) -> str: + """ + Run handle_mirror_command() with patched side effects and capture output. + """ + args = SimpleNamespace(subcommand=subcommand, preview=preview) + + # Fake ensure_remote_repo result (preview safe) + def _fake_ensure_remote_repo(spec, provider_hint=None, options=None): + if options is not None and getattr(options, "preview", False) is not True: + raise AssertionError( + "ensure_remote_repo called without preview=True (should never happen in tests)." + ) + r = MagicMock() + r.status = "preview" + r.message = "Preview mode: no remote provisioning performed." + r.url = None + return r + + buf = io.StringIO() + ctx = self._make_ctx( + repositories_base_dir=str(Path(dummy_repo_dir).parent), + all_repositories=selected, + ) + dummy_repo_ctx = self._make_dummy_repo_ctx(repo_dir=dummy_repo_dir) + + with ( + patch( + "pkgmgr.actions.mirror.setup_cmd.build_context", + return_value=dummy_repo_ctx, + ), + patch( + "pkgmgr.actions.mirror.setup_cmd.ensure_origin_remote", + return_value=None, + ), + patch( + "pkgmgr.actions.mirror.git_remote.ensure_origin_remote", + return_value=None, + ), + patch( + "pkgmgr.actions.mirror.setup_cmd.probe_remote_reachable_detail", + side_effect=probe_detail_side_effect, + ), + patch( + "pkgmgr.actions.mirror.remote_provision.ensure_remote_repo", + side_effect=_fake_ensure_remote_repo, + ), + redirect_stdout(buf), + redirect_stderr(buf), + ): + handle_mirror_command(ctx, args, selected) + + return buf.getvalue() + + def test_mirror_check_preview_prints_warn_reason(self) -> None: + """ + 'mirror check --preview' should: + - probe both git mirrors + - print [OK] for origin + - print [WARN] for backup + reason line + """ + with tempfile.TemporaryDirectory() as tmp: + tmp_path = Path(tmp) + repo_dir = tmp_path / "dummy-repo" + repo_dir.mkdir(parents=True, exist_ok=True) + + selected = [ + {"provider": "github.com", "account": "alice", "repository": "repo"} + ] + + def probe_side_effect(url: str, cwd: str = "."): + if "github.com" in url: + # show "empty repo reachable" note; setup_cmd prints [OK] and does not print reason for ok + return ( + True, + "remote reachable, but no refs found yet (empty repository)", + ) + return False, "(exit 128) fatal: Could not read from remote repository." + + out = self._run_handle( + subcommand="check", + preview=True, + selected=selected, + dummy_repo_dir=str(repo_dir), + probe_detail_side_effect=probe_side_effect, + ) + + self.assertIn("[MIRROR SETUP:REMOTE]", out) + + # origin OK (even with a note returned; still OK) + self.assertIn("[OK] origin: git@github.com:alice/repo.git", out) + + # backup WARN prints reason line + self.assertIn( + "[WARN] backup: ssh://git@git.example:2201/alice/repo.git", out + ) + self.assertIn("reason:", out) + self.assertIn("Could not read from remote repository", out) + + def test_mirror_provision_preview_provisions_each_git_mirror(self) -> None: + """ + 'mirror provision --preview' should: + - print provisioning lines for each git mirror + - still probe and print [OK]/[WARN] + - call ensure_remote_repo only in preview mode (enforced by fake) + """ + with tempfile.TemporaryDirectory() as tmp: + tmp_path = Path(tmp) + repo_dir = tmp_path / "dummy-repo" + repo_dir.mkdir(parents=True, exist_ok=True) + + selected = [ + { + "provider": "github.com", + "account": "alice", + "repository": "repo", + "private": True, + "description": "desc", + } + ] + + def probe_side_effect(url: str, cwd: str = "."): + if "github.com" in url: + return True, "" + return False, "(exit 128) fatal: Could not read from remote repository." + + out = self._run_handle( + subcommand="provision", + preview=True, + selected=selected, + dummy_repo_dir=str(repo_dir), + probe_detail_side_effect=probe_side_effect, + ) + + # provisioning should attempt BOTH mirrors + self.assertIn( + "[REMOTE ENSURE] ensuring mirror 'origin': git@github.com:alice/repo.git", + out, + ) + self.assertIn( + "[REMOTE ENSURE] ensuring mirror 'backup': ssh://git@git.example:2201/alice/repo.git", + out, + ) + + # patched ensure_remote_repo prints PREVIEW status via remote_provision + self.assertIn("[REMOTE ENSURE]", out) + self.assertIn("PREVIEW", out.upper()) + + # probes after provisioning + self.assertIn("[OK] origin: git@github.com:alice/repo.git", out) + self.assertIn( + "[WARN] backup: ssh://git@git.example:2201/alice/repo.git", out + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/integration/test_repos_create_pypi_not_in_git_config.py b/tests/integration/test_repos_create_pypi_not_in_git_config.py index 9e4e5f5..f38edb2 100644 --- a/tests/integration/test_repos_create_pypi_not_in_git_config.py +++ b/tests/integration/test_repos_create_pypi_not_in_git_config.py @@ -40,8 +40,8 @@ class TestCreateRepoPypiNotInGitConfig(unittest.TestCase): with ( # Avoid any real network calls during mirror "remote probing" patch( - "pkgmgr.actions.mirror.setup_cmd.probe_remote_reachable", - return_value=True, + "pkgmgr.actions.mirror.setup_cmd.probe_remote_reachable_detail", + return_value=(True, ""), ), # Force templates to come from our temp directory patch( diff --git a/tests/unit/pkgmgr/actions/mirror/test_setup_cmd.py b/tests/unit/pkgmgr/actions/mirror/test_setup_cmd.py index bff6ab2..e930529 100644 --- a/tests/unit/pkgmgr/actions/mirror/test_setup_cmd.py +++ b/tests/unit/pkgmgr/actions/mirror/test_setup_cmd.py @@ -38,7 +38,6 @@ class TestMirrorSetupCmd(unittest.TestCase): ensure_remote=False, ) - # ensure_origin_remote(repo, ctx, preview) is called positionally in your code m_ensure.assert_called_once() args, kwargs = m_ensure.call_args @@ -50,13 +49,13 @@ class TestMirrorSetupCmd(unittest.TestCase): @patch("pkgmgr.actions.mirror.setup_cmd.build_context") @patch("pkgmgr.actions.mirror.setup_cmd.determine_primary_remote_url") - @patch("pkgmgr.actions.mirror.setup_cmd.probe_remote_reachable") + @patch("pkgmgr.actions.mirror.setup_cmd.probe_remote_reachable_detail") def test_setup_mirrors_remote_no_mirrors_probes_primary( - self, m_probe, m_primary, m_ctx + self, m_probe_detail, m_primary, m_ctx ) -> None: m_ctx.return_value = self._ctx(repo_dir="/tmp/repo", resolved={}) m_primary.return_value = "git@github.com:alice/repo.git" - m_probe.return_value = True + m_probe_detail.return_value = (True, "") repos = [{"provider": "github.com", "account": "alice", "repository": "repo"}] setup_mirrors( @@ -70,14 +69,14 @@ class TestMirrorSetupCmd(unittest.TestCase): ) m_primary.assert_called() - m_probe.assert_called_once_with( + m_probe_detail.assert_called_once_with( "git@github.com:alice/repo.git", cwd="/tmp/repo" ) @patch("pkgmgr.actions.mirror.setup_cmd.build_context") - @patch("pkgmgr.actions.mirror.setup_cmd.probe_remote_reachable") + @patch("pkgmgr.actions.mirror.setup_cmd.probe_remote_reachable_detail") def test_setup_mirrors_remote_with_mirrors_probes_each( - self, m_probe, m_ctx + self, m_probe_detail, m_ctx ) -> None: m_ctx.return_value = self._ctx( repo_dir="/tmp/repo", @@ -86,7 +85,7 @@ class TestMirrorSetupCmd(unittest.TestCase): "backup": "ssh://git@git.veen.world:2201/alice/repo.git", }, ) - m_probe.return_value = True + m_probe_detail.return_value = (True, "") repos = [{"provider": "github.com", "account": "alice", "repository": "repo"}] setup_mirrors( @@ -99,12 +98,105 @@ class TestMirrorSetupCmd(unittest.TestCase): ensure_remote=False, ) - self.assertEqual(m_probe.call_count, 2) - m_probe.assert_any_call("git@github.com:alice/repo.git", cwd="/tmp/repo") - m_probe.assert_any_call( + # Should probe BOTH git mirror URLs + self.assertEqual(m_probe_detail.call_count, 2) + m_probe_detail.assert_any_call("git@github.com:alice/repo.git", cwd="/tmp/repo") + m_probe_detail.assert_any_call( "ssh://git@git.veen.world:2201/alice/repo.git", cwd="/tmp/repo" ) + @patch("pkgmgr.actions.mirror.setup_cmd.build_context") + @patch("pkgmgr.actions.mirror.setup_cmd.probe_remote_reachable_detail") + @patch("pkgmgr.actions.mirror.setup_cmd.ensure_remote_repository_for_url") + def test_setup_mirrors_remote_with_mirrors_ensure_remote_provisions_each( + self, m_ensure_url, m_probe_detail, m_ctx + ) -> None: + m_ctx.return_value = self._ctx( + repo_dir="/tmp/repo", + resolved={ + "origin": "git@github.com:alice/repo.git", + "backup": "ssh://git@git.veen.world:2201/alice/repo.git", + }, + ) + m_probe_detail.return_value = (True, "") + + repos = [ + { + "provider": "github.com", + "account": "alice", + "repository": "repo", + "private": True, + "description": "desc", + } + ] + setup_mirrors( + selected_repos=repos, + repositories_base_dir="/tmp", + all_repos=repos, + preview=True, + local=False, + remote=True, + ensure_remote=True, + ) + + # Provision both mirrors + self.assertEqual(m_ensure_url.call_count, 2) + m_ensure_url.assert_any_call( + url="git@github.com:alice/repo.git", + private_default=True, + description="desc", + preview=True, + ) + m_ensure_url.assert_any_call( + url="ssh://git@git.veen.world:2201/alice/repo.git", + private_default=True, + description="desc", + preview=True, + ) + + # Still probes both + self.assertEqual(m_probe_detail.call_count, 2) + + @patch("pkgmgr.actions.mirror.setup_cmd.build_context") + @patch("pkgmgr.actions.mirror.setup_cmd.determine_primary_remote_url") + @patch("pkgmgr.actions.mirror.setup_cmd.ensure_remote_repository_for_url") + @patch("pkgmgr.actions.mirror.setup_cmd.probe_remote_reachable_detail") + def test_setup_mirrors_remote_no_mirrors_ensure_remote_provisions_primary( + self, m_probe_detail, m_ensure_url, m_primary, m_ctx + ) -> None: + m_ctx.return_value = self._ctx(repo_dir="/tmp/repo", resolved={}) + m_primary.return_value = "git@github.com:alice/repo.git" + m_probe_detail.return_value = (True, "") + + repos = [ + { + "provider": "github.com", + "account": "alice", + "repository": "repo", + "private": False, + "description": "desc", + } + ] + setup_mirrors( + selected_repos=repos, + repositories_base_dir="/tmp", + all_repos=repos, + preview=True, + local=False, + remote=True, + ensure_remote=True, + ) + + m_ensure_url.assert_called_once_with( + url="git@github.com:alice/repo.git", + private_default=False, + description="desc", + preview=True, + ) + m_probe_detail.assert_called_once_with( + "git@github.com:alice/repo.git", cwd="/tmp/repo" + ) + if __name__ == "__main__": unittest.main() diff --git a/tests/unit/pkgmgr/core/git/queries/test_probe_remote_reachable.py b/tests/unit/pkgmgr/core/git/queries/test_probe_remote_reachable.py new file mode 100644 index 0000000..36b077e --- /dev/null +++ b/tests/unit/pkgmgr/core/git/queries/test_probe_remote_reachable.py @@ -0,0 +1,145 @@ +from __future__ import annotations + +import importlib +import unittest +from unittest.mock import patch + +from pkgmgr.core.git.errors import GitRunError + +# IMPORTANT: +# Import the MODULE, not the function exported by pkgmgr.core.git.queries.__init__. +pr = importlib.import_module("pkgmgr.core.git.queries.probe_remote_reachable") + + +def _git_error( + *, + returncode: int, + stderr: str = "", + stdout: str = "", + message: str = "git failed", +) -> GitRunError: + """ + Create a GitRunError that mimics what pkgmgr.core.git.run attaches. + """ + exc = GitRunError(message) + exc.returncode = returncode + exc.stderr = stderr + exc.stdout = stdout + return exc + + +class TestProbeRemoteReachableHelpers(unittest.TestCase): + def test_first_useful_line_prefers_keyword_lines(self) -> None: + text = "\nerror:\n \nFATAL: Could not read from remote repository.\nmore\n" + self.assertEqual( + pr._first_useful_line(text), + "FATAL: Could not read from remote repository.", + ) + + def test_first_useful_line_skips_plain_error_if_possible(self) -> None: + text = "error:\nsome other info\n" + self.assertEqual(pr._first_useful_line(text), "some other info") + + def test_first_useful_line_returns_empty_for_empty(self) -> None: + self.assertEqual(pr._first_useful_line(" \n\n"), "") + + def test_looks_like_real_transport_error_true(self) -> None: + self.assertTrue( + pr._looks_like_real_transport_error( + "fatal: Could not read from remote repository." + ) + ) + + def test_looks_like_real_transport_error_false(self) -> None: + self.assertFalse(pr._looks_like_real_transport_error("some harmless output")) + + +class TestProbeRemoteReachableDetail(unittest.TestCase): + @patch.object(pr, "run", return_value="") + def test_detail_success_returns_true_empty_reason(self, m_run) -> None: + ok, reason = pr.probe_remote_reachable_detail( + "git@github.com:alice/repo.git", + cwd="/tmp", + ) + self.assertTrue(ok) + self.assertEqual(reason, "") + m_run.assert_called_once() + + @patch.object(pr, "run") + def test_detail_rc2_without_transport_indicators_treated_as_reachable( + self, m_run + ) -> None: + # rc=2 but no transport/auth indicators => treat as reachable (empty repo) + m_run.side_effect = _git_error( + returncode=2, + stderr="", + stdout="", + message="Git command failed (exit 2)", + ) + + ok, reason = pr.probe_remote_reachable_detail( + "git@github.com:alice/empty.git", + cwd="/tmp", + ) + self.assertTrue(ok) + self.assertIn("empty repository", reason.lower()) + + @patch.object(pr, "run") + def test_detail_rc2_with_transport_indicators_is_not_reachable(self, m_run) -> None: + # rc=2 but stderr indicates transport/auth problem => NOT reachable + m_run.side_effect = _git_error( + returncode=2, + stderr="ERROR: Repository not found.", + stdout="", + message="Git command failed (exit 2)", + ) + + ok, reason = pr.probe_remote_reachable_detail( + "git@github.com:alice/missing.git", + cwd="/tmp", + ) + self.assertFalse(ok) + self.assertIn("repository not found", reason.lower()) + + @patch.object(pr, "run") + def test_detail_rc128_reports_reason(self, m_run) -> None: + m_run.side_effect = _git_error( + returncode=128, + stderr="fatal: Could not read from remote repository.", + stdout="", + message="Git command failed (exit 128)", + ) + + ok, reason = pr.probe_remote_reachable_detail( + "ssh://git@host:2201/a/b.git", + cwd="/tmp", + ) + self.assertFalse(ok) + self.assertIn("(exit 128)", reason.lower()) + self.assertIn("could not read from remote repository", reason.lower()) + + @patch.object(pr, "run") + def test_detail_adds_hint_if_reason_is_generic(self, m_run) -> None: + # Generic failure: rc=128 but no stderr/stdout => should append hint + m_run.side_effect = _git_error( + returncode=128, + stderr="", + stdout="", + message="", + ) + + url = "git@github.com:alice/repo.git" + ok, reason = pr.probe_remote_reachable_detail(url, cwd="/tmp") + + self.assertFalse(ok) + self.assertIn("hint:", reason.lower()) + self.assertIn("git ls-remote --exit-code", reason.lower()) + + @patch.object(pr, "probe_remote_reachable_detail", return_value=(True, "")) + def test_probe_remote_reachable_delegates_to_detail(self, m_detail) -> None: + self.assertTrue(pr.probe_remote_reachable("x", cwd="/tmp")) + m_detail.assert_called_once_with("x", cwd="/tmp") + + +if __name__ == "__main__": + unittest.main()