From 9802293871dc634bdb7437b92a1b744f780f9ffb Mon Sep 17 00:00:00 2001 From: Kevin Veen-Birkenbach Date: Sat, 20 Dec 2025 14:26:55 +0100 Subject: [PATCH] ***feat(mirror): add remote repository visibility support*** * Add mirror visibility subcommand and provision --public flag * Implement core visibility API with provider support (GitHub, Gitea) * Extend provider interface and EnsureStatus * Add unit, integration and e2e tests for visibility handling https://chatgpt.com/share/6946a44e-4f48-800f-8124-9c0b9b2b6b04 --- src/pkgmgr/actions/mirror/__init__.py | 2 + src/pkgmgr/actions/mirror/setup_cmd.py | 84 ++++- src/pkgmgr/actions/mirror/visibility_cmd.py | 134 ++++++++ src/pkgmgr/cli/commands/mirror.py | 23 ++ src/pkgmgr/cli/parser/mirror_cmd.py | 19 +- .../core/remote_provisioning/__init__.py | 3 +- .../remote_provisioning/providers/base.py | 20 +- .../remote_provisioning/providers/gitea.py | 33 ++ .../remote_provisioning/providers/github.py | 33 ++ src/pkgmgr/core/remote_provisioning/types.py | 11 +- .../core/remote_provisioning/visibility.py | 118 +++++++ tests/e2e/test_mirror_visibility_smoke.py | 127 +++++++ .../test_visibility_integration.py | 322 ++++++++++++++++++ .../actions/mirror/test_visibility_cmd.py | 177 ++++++++++ .../core/remote_provisioning/__init__.py | 0 .../remote_provisioning/test_visibility.py | 227 ++++++++++++ 16 files changed, 1323 insertions(+), 10 deletions(-) create mode 100644 src/pkgmgr/actions/mirror/visibility_cmd.py create mode 100644 src/pkgmgr/core/remote_provisioning/visibility.py create mode 100644 tests/e2e/test_mirror_visibility_smoke.py create mode 100644 tests/integration/test_visibility_integration.py create mode 100644 tests/unit/pkgmgr/actions/mirror/test_visibility_cmd.py create mode 100644 tests/unit/pkgmgr/core/remote_provisioning/__init__.py create mode 100644 tests/unit/pkgmgr/core/remote_provisioning/test_visibility.py diff --git a/src/pkgmgr/actions/mirror/__init__.py b/src/pkgmgr/actions/mirror/__init__.py index 25b8ab1..d77f419 100644 --- a/src/pkgmgr/actions/mirror/__init__.py +++ b/src/pkgmgr/actions/mirror/__init__.py @@ -14,6 +14,7 @@ from .list_cmd import list_mirrors from .diff_cmd import diff_mirrors from .merge_cmd import merge_mirrors from .setup_cmd import setup_mirrors +from .visibility_cmd import set_mirror_visibility __all__ = [ "Repository", @@ -22,4 +23,5 @@ __all__ = [ "diff_mirrors", "merge_mirrors", "setup_mirrors", + "set_mirror_visibility", ] diff --git a/src/pkgmgr/actions/mirror/setup_cmd.py b/src/pkgmgr/actions/mirror/setup_cmd.py index adc641e..fe22014 100644 --- a/src/pkgmgr/actions/mirror/setup_cmd.py +++ b/src/pkgmgr/actions/mirror/setup_cmd.py @@ -3,11 +3,14 @@ from __future__ import annotations from typing import List from pkgmgr.core.git.queries import probe_remote_reachable_detail +from pkgmgr.core.remote_provisioning import ProviderHint, RepoSpec, set_repo_visibility +from pkgmgr.core.remote_provisioning.visibility import VisibilityOptions from .context import build_context from .git_remote import determine_primary_remote_url, ensure_origin_remote from .remote_provision import ensure_remote_repository_for_url from .types import Repository +from .url_utils import normalize_provider_host, parse_repo_from_git_url def _is_git_remote_url(url: str) -> bool: @@ -25,6 +28,45 @@ def _is_git_remote_url(url: str) -> bool: return False +def _provider_hint_from_host(host: str) -> str | None: + h = (host or "").lower() + if h == "github.com": + return "github" + return "gitea" if h else None + + +def _apply_visibility_for_url( + *, + url: str, + private: bool, + description: str, + preview: bool, +) -> None: + 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(f"[WARN] Could not parse repo from URL: {url}") + return + + spec = RepoSpec( + host=host, + owner=owner, + name=name, + private=private, + description=description, + ) + + provider_kind = _provider_hint_from_host(host) + res = set_repo_visibility( + spec, + private=private, + provider_hint=ProviderHint(kind=provider_kind), + options=VisibilityOptions(preview=preview), + ) + print(f"[REMOTE VISIBILITY] {res.status.upper()}: {res.message}") + + 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. @@ -67,6 +109,7 @@ def _setup_remote_mirrors_for_repo( all_repos: List[Repository], preview: bool, ensure_remote: bool, + ensure_visibility: str | None, ) -> None: ctx = build_context(repo, repositories_base_dir, all_repos) @@ -79,6 +122,22 @@ def _setup_remote_mirrors_for_repo( k: v for k, v in ctx.resolved_mirrors.items() if _is_git_remote_url(v) } + def _desired_private_default() -> bool: + # default behavior: repo['private'] (or True) + if ensure_visibility == "public": + return False + if ensure_visibility == "private": + return True + return bool(repo.get("private", True)) + + def _should_enforce_visibility() -> bool: + return ensure_visibility in ("public", "private") + + def _visibility_private_value() -> bool: + return ensure_visibility == "private" + + description = str(repo.get("description", "")) + # If there are no git mirrors, fall back to primary (git) URL. if not git_mirrors: primary = determine_primary_remote_url(repo, ctx) @@ -91,10 +150,18 @@ def _setup_remote_mirrors_for_repo( 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", "")), + private_default=_desired_private_default(), + description=description, preview=preview, ) + # IMPORTANT: enforce visibility only if requested + if _should_enforce_visibility(): + _apply_visibility_for_url( + url=primary, + private=_visibility_private_value(), + description=description, + preview=preview, + ) print() _print_probe_result(None, primary, cwd=ctx.repo_dir) @@ -107,10 +174,17 @@ def _setup_remote_mirrors_for_repo( 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", "")), + private_default=_desired_private_default(), + description=description, preview=preview, ) + if _should_enforce_visibility(): + _apply_visibility_for_url( + url=url, + private=_visibility_private_value(), + description=description, + preview=preview, + ) print() # Probe ALL git mirrors @@ -128,6 +202,7 @@ def setup_mirrors( local: bool = True, remote: bool = True, ensure_remote: bool = False, + ensure_visibility: str | None = None, ) -> None: for repo in selected_repos: if local: @@ -145,4 +220,5 @@ def setup_mirrors( all_repos, preview, ensure_remote, + ensure_visibility, ) diff --git a/src/pkgmgr/actions/mirror/visibility_cmd.py b/src/pkgmgr/actions/mirror/visibility_cmd.py new file mode 100644 index 0000000..6d16e62 --- /dev/null +++ b/src/pkgmgr/actions/mirror/visibility_cmd.py @@ -0,0 +1,134 @@ +from __future__ import annotations + +from typing import List + +from pkgmgr.core.remote_provisioning import ProviderHint, RepoSpec, set_repo_visibility +from pkgmgr.core.remote_provisioning.visibility import VisibilityOptions + +from .context import build_context +from .git_remote import determine_primary_remote_url +from .types import Repository +from .url_utils import normalize_provider_host, parse_repo_from_git_url + + +def _is_git_remote_url(url: str) -> bool: + # Keep same semantics as setup_cmd.py / git_remote.py + u = (url or "").strip() + if not u: + return False + if u.startswith("git@"): + return True + if u.startswith("ssh://"): + return True + if (u.startswith("https://") or u.startswith("http://")) and u.endswith(".git"): + return True + return False + + +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 _apply_visibility_for_url( + *, + url: str, + private: bool, + description: str, + preview: bool, +) -> None: + 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(f"[WARN] Could not parse repo from URL: {url}") + return + + spec = RepoSpec( + host=host, + owner=owner, + name=name, + private=private, + description=description, + ) + + provider_kind = _provider_hint_from_host(host) + res = set_repo_visibility( + spec, + private=private, + provider_hint=ProviderHint(kind=provider_kind), + options=VisibilityOptions(preview=preview), + ) + print(f"[REMOTE VISIBILITY] {res.status.upper()}: {res.message}") + + +def set_mirror_visibility( + selected_repos: List[Repository], + repositories_base_dir: str, + all_repos: List[Repository], + *, + visibility: str, + preview: bool = False, +) -> None: + """ + Set remote repository visibility for all git mirrors of each selected repo. + + visibility: + - "private" + - "public" + """ + v = (visibility or "").strip().lower() + if v not in ("private", "public"): + raise ValueError("visibility must be 'private' or 'public'") + + desired_private = v == "private" + + for repo in selected_repos: + ctx = build_context(repo, repositories_base_dir, all_repos) + + print("------------------------------------------------------------") + print(f"[MIRROR VISIBILITY] {ctx.identifier}") + print(f"[MIRROR VISIBILITY] dir: {ctx.repo_dir}") + print(f"[MIRROR VISIBILITY] target: {v}") + print("------------------------------------------------------------") + + git_mirrors = { + name: url + for name, url in ctx.resolved_mirrors.items() + if url and _is_git_remote_url(url) + } + + # 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 found (and no primary git URL). Nothing to do." + ) + print() + continue + + print(f"[MIRROR VISIBILITY] applying to primary: {primary}") + _apply_visibility_for_url( + url=primary, + private=desired_private, + description=str(repo.get("description", "")), + preview=preview, + ) + print() + continue + + # Apply to ALL git mirrors + for name, url in git_mirrors.items(): + print(f"[MIRROR VISIBILITY] applying to mirror {name!r}: {url}") + _apply_visibility_for_url( + url=url, + private=desired_private, + description=str(repo.get("description", "")), + preview=preview, + ) + + print() diff --git a/src/pkgmgr/cli/commands/mirror.py b/src/pkgmgr/cli/commands/mirror.py index 907ec08..e341e17 100644 --- a/src/pkgmgr/cli/commands/mirror.py +++ b/src/pkgmgr/cli/commands/mirror.py @@ -8,6 +8,7 @@ from pkgmgr.actions.mirror import ( diff_mirrors, list_mirrors, merge_mirrors, + set_mirror_visibility, setup_mirrors, ) from pkgmgr.cli.context import CLIContext @@ -30,6 +31,7 @@ def handle_mirror_command( - mirror setup - mirror check - mirror provision + - mirror visibility """ if not selected: print("[INFO] No repositories selected for 'mirror' command.") @@ -92,6 +94,7 @@ def handle_mirror_command( local=True, remote=False, ensure_remote=False, + ensure_visibility=None, ) return @@ -105,11 +108,14 @@ def handle_mirror_command( local=False, remote=True, ensure_remote=False, + ensure_visibility=None, ) return if subcommand == "provision": preview = getattr(args, "preview", False) + public = bool(getattr(args, "public", False)) + setup_mirrors( selected_repos=selected, repositories_base_dir=ctx.repositories_base_dir, @@ -118,6 +124,23 @@ def handle_mirror_command( local=False, remote=True, ensure_remote=True, + ensure_visibility="public" if public else None, + ) + return + + if subcommand == "visibility": + preview = getattr(args, "preview", False) + visibility = getattr(args, "visibility", None) + if visibility not in ("private", "public"): + print("[ERROR] mirror visibility expects 'private' or 'public'.") + sys.exit(2) + + set_mirror_visibility( + selected_repos=selected, + repositories_base_dir=ctx.repositories_base_dir, + all_repos=ctx.all_repositories, + visibility=visibility, + preview=preview, ) return diff --git a/src/pkgmgr/cli/parser/mirror_cmd.py b/src/pkgmgr/cli/parser/mirror_cmd.py index b467570..0a6e860 100644 --- a/src/pkgmgr/cli/parser/mirror_cmd.py +++ b/src/pkgmgr/cli/parser/mirror_cmd.py @@ -1,4 +1,3 @@ -# src/pkgmgr/cli/parser/mirror_cmd.py #!/usr/bin/env python3 # -*- coding: utf-8 -*- @@ -12,7 +11,7 @@ from .common import add_identifier_arguments def add_mirror_subparsers(subparsers: argparse._SubParsersAction) -> None: mirror_parser = subparsers.add_parser( "mirror", - help="Mirror-related utilities (list, diff, merge, setup, check, provision)", + help="Mirror-related utilities (list, diff, merge, setup, check, provision, visibility)", ) mirror_subparsers = mirror_parser.add_subparsers( dest="subcommand", @@ -68,4 +67,20 @@ def add_mirror_subparsers(subparsers: argparse._SubParsersAction) -> None: "provision", help="Provision remote repositories via provider APIs (create missing repos).", ) + mirror_provision.add_argument( + "--public", + action="store_true", + help="After ensuring repos exist, enforce public visibility on the remote provider.", + ) add_identifier_arguments(mirror_provision) + + mirror_visibility = mirror_subparsers.add_parser( + "visibility", + help="Set visibility (public/private) for all remote git mirrors via provider APIs.", + ) + mirror_visibility.add_argument( + "visibility", + choices=["private", "public"], + help="Target visibility for all git mirrors.", + ) + add_identifier_arguments(mirror_visibility) diff --git a/src/pkgmgr/core/remote_provisioning/__init__.py b/src/pkgmgr/core/remote_provisioning/__init__.py index ca07e4c..8501004 100644 --- a/src/pkgmgr/core/remote_provisioning/__init__.py +++ b/src/pkgmgr/core/remote_provisioning/__init__.py @@ -1,12 +1,13 @@ -# src/pkgmgr/core/remote_provisioning/__init__.py """Remote repository provisioning (ensure remote repo exists).""" from .ensure import ensure_remote_repo from .registry import ProviderRegistry from .types import EnsureResult, ProviderHint, RepoSpec +from .visibility import set_repo_visibility __all__ = [ "ensure_remote_repo", + "set_repo_visibility", "RepoSpec", "EnsureResult", "ProviderHint", diff --git a/src/pkgmgr/core/remote_provisioning/providers/base.py b/src/pkgmgr/core/remote_provisioning/providers/base.py index a92854f..96da7ed 100644 --- a/src/pkgmgr/core/remote_provisioning/providers/base.py +++ b/src/pkgmgr/core/remote_provisioning/providers/base.py @@ -1,4 +1,3 @@ -# src/pkgmgr/core/remote_provisioning/providers/base.py from __future__ import annotations from abc import ABC, abstractmethod @@ -23,7 +22,26 @@ class RemoteProvider(ABC): def create_repo(self, token: str, spec: RepoSpec) -> EnsureResult: """Create a repository (owner may be user or org).""" + @abstractmethod + def get_repo_private(self, token: str, spec: RepoSpec) -> bool | None: + """ + Return current repo privacy, or None if repo not found / inaccessible. + + IMPORTANT: + - Must NOT create repositories. + - Should return None on 404 (not found) or when the repo cannot be accessed. + """ + + @abstractmethod + def set_repo_private(self, token: str, spec: RepoSpec, *, private: bool) -> None: + """ + Update repo privacy (PATCH). Must NOT create repositories. + + Implementations should raise HttpError on API failure. + """ + def ensure_repo(self, token: str, spec: RepoSpec) -> EnsureResult: + """Ensure repository exists (create if missing).""" if self.repo_exists(token, spec): return EnsureResult(status="exists", message="Repository exists.") return self.create_repo(token, spec) diff --git a/src/pkgmgr/core/remote_provisioning/providers/gitea.py b/src/pkgmgr/core/remote_provisioning/providers/gitea.py index cbf7c4b..1c89b39 100644 --- a/src/pkgmgr/core/remote_provisioning/providers/gitea.py +++ b/src/pkgmgr/core/remote_provisioning/providers/gitea.py @@ -52,6 +52,39 @@ class GiteaProvider(RemoteProvider): return False raise + def get_repo_private(self, token: str, spec: RepoSpec) -> bool | None: + base = self._api_base(spec.host) + url = f"{base}/api/v1/repos/{spec.owner}/{spec.name}" + try: + resp = self._http.request_json("GET", url, headers=self._headers(token)) + except HttpError as exc: + if exc.status == 404: + return None + raise + + if not (200 <= resp.status < 300): + return None + data = resp.json or {} + return bool(data.get("private", False)) + + def set_repo_private(self, token: str, spec: RepoSpec, *, private: bool) -> None: + base = self._api_base(spec.host) + url = f"{base}/api/v1/repos/{spec.owner}/{spec.name}" + payload: Dict[str, Any] = {"private": bool(private)} + + resp = self._http.request_json( + "PATCH", + url, + headers=self._headers(token), + payload=payload, + ) + if not (200 <= resp.status < 300): + raise HttpError( + status=resp.status, + message="Failed to update repository.", + body=resp.text, + ) + def create_repo(self, token: str, spec: RepoSpec) -> EnsureResult: base = self._api_base(spec.host) diff --git a/src/pkgmgr/core/remote_provisioning/providers/github.py b/src/pkgmgr/core/remote_provisioning/providers/github.py index 74352f7..832b39d 100644 --- a/src/pkgmgr/core/remote_provisioning/providers/github.py +++ b/src/pkgmgr/core/remote_provisioning/providers/github.py @@ -54,6 +54,39 @@ class GitHubProvider(RemoteProvider): return False raise + def get_repo_private(self, token: str, spec: RepoSpec) -> bool | None: + api = self._api_base(spec.host) + url = f"{api}/repos/{spec.owner}/{spec.name}" + try: + resp = self._http.request_json("GET", url, headers=self._headers(token)) + except HttpError as exc: + if exc.status == 404: + return None + raise + + if not (200 <= resp.status < 300): + return None + data = resp.json or {} + return bool(data.get("private", False)) + + def set_repo_private(self, token: str, spec: RepoSpec, *, private: bool) -> None: + api = self._api_base(spec.host) + url = f"{api}/repos/{spec.owner}/{spec.name}" + payload: Dict[str, Any] = {"private": bool(private)} + + resp = self._http.request_json( + "PATCH", + url, + headers=self._headers(token), + payload=payload, + ) + if not (200 <= resp.status < 300): + raise HttpError( + status=resp.status, + message="Failed to update repository.", + body=resp.text, + ) + def create_repo(self, token: str, spec: RepoSpec) -> EnsureResult: api = self._api_base(spec.host) diff --git a/src/pkgmgr/core/remote_provisioning/types.py b/src/pkgmgr/core/remote_provisioning/types.py index 2f74824..46b5afd 100644 --- a/src/pkgmgr/core/remote_provisioning/types.py +++ b/src/pkgmgr/core/remote_provisioning/types.py @@ -1,10 +1,17 @@ -# src/pkgmgr/core/remote_provisioning/types.py from __future__ import annotations from dataclasses import dataclass from typing import Literal, Optional -EnsureStatus = Literal["exists", "created", "skipped", "failed"] +EnsureStatus = Literal[ + "exists", + "created", + "updated", + "noop", + "notfound", + "skipped", + "failed", +] @dataclass(frozen=True) diff --git a/src/pkgmgr/core/remote_provisioning/visibility.py b/src/pkgmgr/core/remote_provisioning/visibility.py new file mode 100644 index 0000000..3dc60b1 --- /dev/null +++ b/src/pkgmgr/core/remote_provisioning/visibility.py @@ -0,0 +1,118 @@ +# src/pkgmgr/core/remote_provisioning/visibility.py +from __future__ import annotations + +from dataclasses import dataclass +from typing import Optional + +from pkgmgr.core.credentials.resolver import ResolutionOptions, TokenResolver + +from .http.errors import HttpError +from .registry import ProviderRegistry +from .types import ( + AuthError, + EnsureResult, + NetworkError, + PermissionError, + ProviderHint, + RepoSpec, + UnsupportedProviderError, +) + + +@dataclass(frozen=True) +class VisibilityOptions: + """Options controlling remote visibility updates.""" + + preview: bool = False + interactive: bool = True + allow_prompt: bool = True + save_prompt_token_to_keyring: bool = True + + +def _raise_mapped_http_error(exc: HttpError, host: str) -> None: + """Map HttpError into domain-specific error types.""" + if exc.status == 0: + raise NetworkError(f"Network error while talking to {host}: {exc}") from exc + if exc.status == 401: + raise AuthError(f"Authentication failed for {host} (401).") from exc + if exc.status == 403: + raise PermissionError(f"Permission denied for {host} (403).") from exc + + raise NetworkError( + f"HTTP error from {host}: status={exc.status}, message={exc}, body={exc.body}" + ) from exc + + +def set_repo_visibility( + spec: RepoSpec, + *, + private: bool, + provider_hint: Optional[ProviderHint] = None, + options: Optional[VisibilityOptions] = None, + registry: Optional[ProviderRegistry] = None, + token_resolver: Optional[TokenResolver] = None, +) -> EnsureResult: + """ + Set repository visibility (public/private) WITHOUT creating repositories. + + Behavior: + - If repo does not exist -> status=notfound + - If already desired -> status=noop + - If changed -> status=updated + - Respects preview mode -> status=skipped + - Maps HTTP errors to domain-specific errors + """ + opts = options or VisibilityOptions() + reg = registry or ProviderRegistry.default() + resolver = token_resolver or TokenResolver() + + provider = reg.resolve(spec.host) + if provider_hint and provider_hint.kind: + forced = provider_hint.kind.strip().lower() + forced_provider = next( + (p for p in reg.providers if getattr(p, "kind", "").lower() == forced), + None, + ) + if forced_provider is not None: + provider = forced_provider + + if provider is None: + raise UnsupportedProviderError(f"No provider matched host: {spec.host}") + + token_opts = ResolutionOptions( + interactive=opts.interactive, + allow_prompt=opts.allow_prompt, + save_prompt_token_to_keyring=opts.save_prompt_token_to_keyring, + ) + token = resolver.get_token( + provider_kind=getattr(provider, "kind", "unknown"), + host=spec.host, + owner=spec.owner, + options=token_opts, + ) + + if opts.preview: + return EnsureResult( + status="skipped", + message="Preview mode: no remote changes performed.", + ) + + try: + current_private = provider.get_repo_private(token.token, spec) + if current_private is None: + return EnsureResult(status="notfound", message="Repository not found.") + + if bool(current_private) == bool(private): + return EnsureResult( + status="noop", + message=f"Repository already {'private' if private else 'public'}.", + ) + + provider.set_repo_private(token.token, spec, private=private) + return EnsureResult( + status="updated", + message=f"Visibility updated to {'private' if private else 'public'}.", + ) + except HttpError as exc: + _raise_mapped_http_error(exc, host=spec.host) + return EnsureResult(status="failed", message="Unreachable error mapping.") diff --git a/tests/e2e/test_mirror_visibility_smoke.py b/tests/e2e/test_mirror_visibility_smoke.py new file mode 100644 index 0000000..c3c7ea0 --- /dev/null +++ b/tests/e2e/test_mirror_visibility_smoke.py @@ -0,0 +1,127 @@ +# tests/e2e/test_mirror_visibility_smoke.py +from __future__ import annotations + +import os +import shutil +import subprocess +import sys +import unittest +from pathlib import Path + + +class TestMirrorVisibilityE2ESmoke(unittest.TestCase): + """ + E2E smoke tests for the new mirror visibility feature. + + We intentionally DO NOT execute provider APIs or require tokens. + The tests only verify that: + - CLI exposes the new subcommands / flags via --help + - Python public API surface is wired and importable + + IMPORTANT: + - `python -m pkgmgr.cli` is NOT valid unless pkgmgr/cli/__main__.py exists. + - In this repo, `from pkgmgr.cli import main` is the stable entrypoint. + """ + + @staticmethod + def _project_root() -> Path: + # tests/e2e/... -> project root is parents[2] + return Path(__file__).resolve().parents[2] + + def _run(self, args: list[str]) -> subprocess.CompletedProcess[str]: + env = os.environ.copy() + env.setdefault("PYTHONUNBUFFERED", "1") + + return subprocess.run( + args, + cwd=str(self._project_root()), + env=env, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + check=False, + ) + + def _run_pkgmgr(self, pkgmgr_args: list[str]) -> subprocess.CompletedProcess[str]: + """ + Run the pkgmgr CLI in a way that works both: + - when the console script `pkgmgr` is available on PATH + - when only source imports are available + + We prefer the console script if present because it's closest to real E2E. + """ + exe = shutil.which("pkgmgr") + if exe: + return self._run([exe, *pkgmgr_args]) + + # Fallback to a Python-level entrypoint that exists in your repo: + # The stacktrace showed: from pkgmgr.cli import main + # We call it with argv simulation. + code = r""" +import sys +from pkgmgr.cli import main + +sys.argv = ["pkgmgr"] + sys.argv[1:] +main() +""" + return self._run([sys.executable, "-c", code, *pkgmgr_args]) + + def test_cli_help_lists_visibility_and_provision_public(self) -> None: + # `pkgmgr mirror --help` should mention "visibility" + p = self._run_pkgmgr(["mirror", "--help"]) + self.assertEqual( + p.returncode, + 0, + msg=f"Expected exit code 0, got {p.returncode}\n\nOutput:\n{p.stdout}", + ) + out_lower = p.stdout.lower() + self.assertIn("visibility", out_lower) + self.assertIn("provision", out_lower) + + # `pkgmgr mirror provision --help` should show `--public` + p = self._run_pkgmgr(["mirror", "provision", "--help"]) + self.assertEqual( + p.returncode, + 0, + msg=f"Expected exit code 0, got {p.returncode}\n\nOutput:\n{p.stdout}", + ) + self.assertIn("--public", p.stdout) + + # `pkgmgr mirror visibility --help` should show choices {private, public} + p = self._run_pkgmgr(["mirror", "visibility", "--help"]) + self.assertEqual( + p.returncode, + 0, + msg=f"Expected exit code 0, got {p.returncode}\n\nOutput:\n{p.stdout}", + ) + out_lower = p.stdout.lower() + self.assertIn("private", out_lower) + self.assertIn("public", out_lower) + + def test_python_api_surface_is_exposed(self) -> None: + # Ensure public exports exist and setup_mirrors has ensure_visibility in signature. + code = r""" +import inspect + +from pkgmgr.actions import mirror as mirror_actions +from pkgmgr.core import remote_provisioning as rp + +assert hasattr(mirror_actions, "set_mirror_visibility"), "set_mirror_visibility missing in pkgmgr.actions.mirror" +assert hasattr(rp, "set_repo_visibility"), "set_repo_visibility missing in pkgmgr.core.remote_provisioning" + +sig = inspect.signature(mirror_actions.setup_mirrors) +assert "ensure_visibility" in sig.parameters, "setup_mirrors missing ensure_visibility parameter" + +print("OK") +""" + p = self._run([sys.executable, "-c", code]) + self.assertEqual( + p.returncode, + 0, + msg=f"Expected exit code 0, got {p.returncode}\n\nOutput:\n{p.stdout}", + ) + self.assertIn("OK", p.stdout) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/integration/test_visibility_integration.py b/tests/integration/test_visibility_integration.py new file mode 100644 index 0000000..db8c50a --- /dev/null +++ b/tests/integration/test_visibility_integration.py @@ -0,0 +1,322 @@ +# tests/integration/test_visibility_integration.py +from __future__ import annotations + +import io +import os +import tempfile +import types +import unittest +from contextlib import redirect_stdout +from typing import Any, Dict, List, Optional, Tuple +from unittest.mock import patch + +from pkgmgr.actions.mirror.setup_cmd import setup_mirrors +from pkgmgr.actions.mirror.visibility_cmd import set_mirror_visibility +from pkgmgr.core.remote_provisioning.types import RepoSpec + + +Repository = Dict[str, Any] + + +class _FakeRegistry: + """ + Minimal ProviderRegistry-like object for tests. + + - has .providers for provider-hint selection + - has .resolve(host) to pick a provider + """ + + def __init__(self, provider: Any) -> None: + self.providers = [provider] + self._provider = provider + + def resolve(self, host: str) -> Any: + return self._provider + + +class FakeProvider: + """ + Fake remote provider implementing the visibility API surface. + + Key feature: tolerant host matching, because normalize_provider_host()/URL parsing + may drop ports or schemes. + """ + + kind = "gitea" + + def __init__(self) -> None: + # maps (host, owner, name) -> private(bool) + self.privacy: Dict[Tuple[str, str, str], bool] = {} + self.calls: List[Tuple[str, Any]] = [] + + def can_handle(self, host: str) -> bool: + return True + + def _candidate_hosts(self, host: str) -> List[str]: + """ + Be tolerant against host normalization differences: + - may contain scheme (https://...) + - may contain port (host:2201) + """ + h = (host or "").strip() + if not h: + return [h] + + candidates = [h] + + # strip scheme if present + if h.startswith("http://"): + candidates.append(h[len("http://") :]) + if h.startswith("https://"): + candidates.append(h[len("https://") :]) + + # strip port if present (host:port) + for c in list(candidates): + if ":" in c: + candidates.append(c.split(":", 1)[0]) + + # de-dup + out: List[str] = [] + for c in candidates: + if c not in out: + out.append(c) + return out + + def repo_exists(self, token: str, spec: RepoSpec) -> bool: + self.calls.append(("repo_exists", (token, spec))) + for h in self._candidate_hosts(spec.host): + if (h, spec.owner, spec.name) in self.privacy: + return True + return False + + def create_repo(self, token: str, spec: RepoSpec): + self.calls.append(("create_repo", (token, spec))) + # store under the provided host (as-is) + self.privacy[(spec.host, spec.owner, spec.name)] = bool(spec.private) + return types.SimpleNamespace(status="created", message="created", url=None) + + def get_repo_private(self, token: str, spec: RepoSpec) -> Optional[bool]: + self.calls.append(("get_repo_private", (token, spec))) + for h in self._candidate_hosts(spec.host): + key = (h, spec.owner, spec.name) + if key in self.privacy: + return self.privacy[key] + return None + + def set_repo_private(self, token: str, spec: RepoSpec, *, private: bool) -> None: + self.calls.append(("set_repo_private", (token, spec, private))) + # update whichever key exists; else create on spec.host + for h in self._candidate_hosts(spec.host): + key = (h, spec.owner, spec.name) + if key in self.privacy: + self.privacy[key] = bool(private) + return + self.privacy[(spec.host, spec.owner, spec.name)] = bool(private) + + +def _mk_ctx(*, identifier: str, repo_dir: str, mirrors: Dict[str, str]) -> Any: + return types.SimpleNamespace( + identifier=identifier, + repo_dir=repo_dir, + resolved_mirrors=mirrors, + ) + + +class TestMirrorVisibilityIntegration(unittest.TestCase): + """ + Integration tests for: + - pkgmgr.actions.mirror.visibility_cmd.set_mirror_visibility + - pkgmgr.actions.mirror.setup_cmd.setup_mirrors (ensure_visibility semantics) + """ + + def setUp(self) -> None: + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + + def _repo_dir(self, name: str) -> str: + d = os.path.join(self.tmp.name, name) + os.makedirs(d, exist_ok=True) + return d + + @patch("pkgmgr.core.credentials.resolver.TokenResolver.get_token") + @patch("pkgmgr.core.remote_provisioning.visibility.ProviderRegistry.default") + @patch("pkgmgr.actions.mirror.visibility_cmd.build_context") + def test_mirror_visibility_applies_to_all_git_mirrors_updated_and_noop( + self, + m_build_context, + m_registry_default, + m_get_token, + ) -> None: + """ + Scenario: + - repo has two git mirrors + - one mirror needs update -> UPDATED + - second mirror already desired -> NOOP + """ + provider = FakeProvider() + registry = _FakeRegistry(provider) + m_registry_default.return_value = registry + + # Avoid interactive token prompt + m_get_token.return_value = types.SimpleNamespace(token="test-token") + + # Seed provider state: + # - repo1 currently private=True + # - We'll set visibility to public -> should UPDATE + provider.privacy[("git.veen.world", "me", "repo1")] = True + + repo = {"id": "repo1", "description": "Repo 1"} + repo_dir = self._repo_dir("repo1") + + m_build_context.return_value = _mk_ctx( + identifier="repo1", + repo_dir=repo_dir, + mirrors={ + "origin": "ssh://git.veen.world:2201/me/repo1.git", + "backup": "https://git.veen.world:2201/me/repo1.git", + }, + ) + + buf = io.StringIO() + with redirect_stdout(buf): + set_mirror_visibility( + selected_repos=[repo], + repositories_base_dir=self.tmp.name, + all_repos=[repo], + visibility="public", + preview=False, + ) + out = buf.getvalue() + + # We apply to BOTH git mirrors. + self.assertIn("[MIRROR VISIBILITY] applying to mirror 'origin':", out) + self.assertIn("[MIRROR VISIBILITY] applying to mirror 'backup':", out) + + # After first update, second call will see it already public (NOOP). + self.assertIn("[REMOTE VISIBILITY] UPDATED:", out) + self.assertIn("[REMOTE VISIBILITY] NOOP:", out) + + # Final state must be public (private=False) + self.assertFalse(provider.privacy[("git.veen.world", "me", "repo1")]) + + @patch("pkgmgr.core.credentials.resolver.TokenResolver.get_token") + @patch("pkgmgr.core.remote_provisioning.visibility.ProviderRegistry.default") + @patch("pkgmgr.actions.mirror.visibility_cmd.build_context") + @patch("pkgmgr.actions.mirror.visibility_cmd.determine_primary_remote_url") + def test_mirror_visibility_fallback_to_primary_when_no_git_mirrors( + self, + m_determine_primary, + m_build_context, + m_registry_default, + m_get_token, + ) -> None: + """ + Scenario: + - no git mirrors in MIRRORS config + - we fall back to primary URL and apply visibility there + """ + provider = FakeProvider() + registry = _FakeRegistry(provider) + m_registry_default.return_value = registry + m_get_token.return_value = types.SimpleNamespace(token="test-token") + + # Seed state: currently public (private=False), target private -> UPDATED + provider.privacy[("git.veen.world", "me", "repo2")] = False + + repo = {"id": "repo2", "description": "Repo 2"} + repo_dir = self._repo_dir("repo2") + + m_build_context.return_value = _mk_ctx( + identifier="repo2", + repo_dir=repo_dir, + mirrors={ + # non-git mirror entries + "pypi": "https://pypi.org/project/example/", + }, + ) + m_determine_primary.return_value = "ssh://git.veen.world:2201/me/repo2.git" + + buf = io.StringIO() + with redirect_stdout(buf): + set_mirror_visibility( + selected_repos=[repo], + repositories_base_dir=self.tmp.name, + all_repos=[repo], + visibility="private", + preview=False, + ) + out = buf.getvalue() + + self.assertIn("[MIRROR VISIBILITY] applying to primary:", out) + self.assertIn("[REMOTE VISIBILITY] UPDATED:", out) + self.assertTrue(provider.privacy[("git.veen.world", "me", "repo2")]) + + @patch("pkgmgr.actions.mirror.setup_cmd.probe_remote_reachable_detail") + @patch("pkgmgr.actions.mirror.setup_cmd.ensure_remote_repository_for_url") + @patch("pkgmgr.core.credentials.resolver.TokenResolver.get_token") + @patch("pkgmgr.core.remote_provisioning.visibility.ProviderRegistry.default") + @patch("pkgmgr.actions.mirror.setup_cmd.build_context") + def test_setup_mirrors_provision_public_enforces_visibility_and_private_default( + self, + m_build_context, + m_registry_default, + m_get_token, + m_ensure_remote_for_url, + m_probe, + ) -> None: + """ + Covers the "mirror provision --public" semantics: + - setup_mirrors(remote=True, ensure_remote=True, ensure_visibility="public") + - ensure_remote_repository_for_url is called with private_default=False + - then set_repo_visibility is applied (UPDATED/NOOP depending on current state) + - git probing is mocked (no subprocess) + """ + provider = FakeProvider() + registry = _FakeRegistry(provider) + m_registry_default.return_value = registry + m_get_token.return_value = types.SimpleNamespace(token="test-token") + + # Make git probing always OK (no subprocess calls) + m_probe.return_value = (True, "") + + # Seed provider: repo4 currently private=True, target public -> UPDATED + provider.privacy[("git.veen.world", "me", "repo4")] = True + + repo = {"id": "repo4", "description": "Repo 4", "private": True} + repo_dir = self._repo_dir("repo4") + + m_build_context.return_value = _mk_ctx( + identifier="repo4", + repo_dir=repo_dir, + mirrors={ + "origin": "ssh://git.veen.world:2201/me/repo4.git", + }, + ) + + buf = io.StringIO() + with redirect_stdout(buf): + setup_mirrors( + selected_repos=[repo], + repositories_base_dir=self.tmp.name, + all_repos=[repo], + preview=False, + local=False, + remote=True, + ensure_remote=True, + ensure_visibility="public", + ) + out = buf.getvalue() + + # ensure_remote_repository_for_url called and private_default overridden to False + self.assertTrue(m_ensure_remote_for_url.called) + _, kwargs = m_ensure_remote_for_url.call_args + self.assertIn("private_default", kwargs) + self.assertFalse(kwargs["private_default"]) + + # Visibility should be enforced + self.assertIn("[REMOTE VISIBILITY] UPDATED:", out) + self.assertFalse(provider.privacy[("git.veen.world", "me", "repo4")]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/pkgmgr/actions/mirror/test_visibility_cmd.py b/tests/unit/pkgmgr/actions/mirror/test_visibility_cmd.py new file mode 100644 index 0000000..e76e352 --- /dev/null +++ b/tests/unit/pkgmgr/actions/mirror/test_visibility_cmd.py @@ -0,0 +1,177 @@ +# tests/unit/pkgmgr/actions/mirror/test_visibility_cmd.py +from __future__ import annotations + +import io +import unittest +from contextlib import redirect_stdout +from unittest.mock import patch, MagicMock + +from pkgmgr.actions.mirror.visibility_cmd import set_mirror_visibility + + +class TestMirrorVisibilityCmd(unittest.TestCase): + def test_invalid_visibility_raises_value_error(self) -> None: + with self.assertRaises(ValueError): + set_mirror_visibility( + selected_repos=[{"id": "x"}], + repositories_base_dir="/tmp", + all_repos=[], + visibility="nope", + ) + + @patch("pkgmgr.actions.mirror.visibility_cmd.build_context") + @patch("pkgmgr.actions.mirror.visibility_cmd.determine_primary_remote_url") + def test_no_git_mirrors_and_no_primary_prints_nothing_to_do( + self, + mock_determine_primary: MagicMock, + mock_build_ctx: MagicMock, + ) -> None: + ctx = MagicMock() + ctx.identifier = "repo1" + ctx.repo_dir = "/tmp/repo1" + ctx.resolved_mirrors = {"pypi": "https://pypi.org/project/x/"} # non-git + mock_build_ctx.return_value = ctx + mock_determine_primary.return_value = None + + buf = io.StringIO() + with redirect_stdout(buf): + set_mirror_visibility( + selected_repos=[{"id": "repo1", "description": "desc"}], + repositories_base_dir="/tmp", + all_repos=[], + visibility="public", + preview=True, + ) + + out = buf.getvalue() + self.assertIn("[MIRROR VISIBILITY] repo1", out) + self.assertIn("Nothing to do.", out) + + @patch("pkgmgr.actions.mirror.visibility_cmd.build_context") + @patch("pkgmgr.actions.mirror.visibility_cmd.determine_primary_remote_url") + @patch("pkgmgr.actions.mirror.visibility_cmd.normalize_provider_host") + @patch("pkgmgr.actions.mirror.visibility_cmd.parse_repo_from_git_url") + @patch("pkgmgr.actions.mirror.visibility_cmd.set_repo_visibility") + def test_applies_to_primary_when_no_git_mirrors( + self, + mock_set_repo_visibility: MagicMock, + mock_parse: MagicMock, + mock_norm: MagicMock, + mock_determine_primary: MagicMock, + mock_build_ctx: MagicMock, + ) -> None: + ctx = MagicMock() + ctx.identifier = "repo1" + ctx.repo_dir = "/tmp/repo1" + ctx.resolved_mirrors = {} # no mirrors + mock_build_ctx.return_value = ctx + + primary = "ssh://git.veen.world:2201/me/repo1.git" + mock_determine_primary.return_value = primary + + mock_parse.return_value = ("git.veen.world:2201", "me", "repo1") + mock_norm.return_value = "git.veen.world:2201" + + mock_set_repo_visibility.return_value = MagicMock( + status="skipped", message="Preview" + ) + + buf = io.StringIO() + with redirect_stdout(buf): + set_mirror_visibility( + selected_repos=[{"id": "repo1", "description": "desc"}], + repositories_base_dir="/tmp", + all_repos=[], + visibility="private", + preview=True, + ) + + mock_set_repo_visibility.assert_called_once() + _, kwargs = mock_set_repo_visibility.call_args + self.assertEqual( + kwargs["private"], True + ) # visibility=private => desired_private=True + out = buf.getvalue() + self.assertIn("applying to primary", out) + + @patch("pkgmgr.actions.mirror.visibility_cmd.build_context") + @patch("pkgmgr.actions.mirror.visibility_cmd.normalize_provider_host") + @patch("pkgmgr.actions.mirror.visibility_cmd.parse_repo_from_git_url") + @patch("pkgmgr.actions.mirror.visibility_cmd.set_repo_visibility") + def test_applies_to_all_git_mirrors( + self, + mock_set_repo_visibility: MagicMock, + mock_parse: MagicMock, + mock_norm: MagicMock, + mock_build_ctx: MagicMock, + ) -> None: + ctx = MagicMock() + ctx.identifier = "repo1" + ctx.repo_dir = "/tmp/repo1" + ctx.resolved_mirrors = { + "origin": "ssh://git.veen.world:2201/me/repo1.git", + "backup": "git@git.veen.world:me/repo1.git", + "notgit": "https://pypi.org/project/x/", + } + mock_build_ctx.return_value = ctx + + # For both URLs, parsing returns same repo + mock_parse.return_value = ("git.veen.world", "me", "repo1") + mock_norm.return_value = "git.veen.world" + + mock_set_repo_visibility.return_value = MagicMock( + status="noop", message="Already public" + ) + + buf = io.StringIO() + with redirect_stdout(buf): + set_mirror_visibility( + selected_repos=[{"id": "repo1", "description": "desc"}], + repositories_base_dir="/tmp", + all_repos=[], + visibility="public", + preview=False, + ) + + # Should be called for origin + backup (2), but not for notgit + self.assertEqual(mock_set_repo_visibility.call_count, 2) + + # Each call should request desired private=False for "public" + for call in mock_set_repo_visibility.call_args_list: + _, kwargs = call + self.assertEqual(kwargs["private"], False) + + out = buf.getvalue() + self.assertIn("applying to mirror 'origin'", out) + self.assertIn("applying to mirror 'backup'", out) + + @patch("pkgmgr.actions.mirror.visibility_cmd.build_context") + @patch("pkgmgr.actions.mirror.visibility_cmd.determine_primary_remote_url") + def test_primary_not_git_prints_nothing_to_do( + self, + mock_determine_primary: MagicMock, + mock_build_ctx: MagicMock, + ) -> None: + ctx = MagicMock() + ctx.identifier = "repo1" + ctx.repo_dir = "/tmp/repo1" + ctx.resolved_mirrors = {} + mock_build_ctx.return_value = ctx + + mock_determine_primary.return_value = "https://example.com/not-a-git-url" + + buf = io.StringIO() + with redirect_stdout(buf): + set_mirror_visibility( + selected_repos=[{"id": "repo1"}], + repositories_base_dir="/tmp", + all_repos=[], + visibility="public", + ) + + out = buf.getvalue() + self.assertIn("Nothing to do.", out) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/pkgmgr/core/remote_provisioning/__init__.py b/tests/unit/pkgmgr/core/remote_provisioning/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/pkgmgr/core/remote_provisioning/test_visibility.py b/tests/unit/pkgmgr/core/remote_provisioning/test_visibility.py new file mode 100644 index 0000000..064af5c --- /dev/null +++ b/tests/unit/pkgmgr/core/remote_provisioning/test_visibility.py @@ -0,0 +1,227 @@ +# tests/unit/pkgmgr/core/remote_provisioning/test_visibility.py +from __future__ import annotations + +import unittest +from unittest.mock import MagicMock + +from pkgmgr.core.remote_provisioning.types import ( + AuthError, + NetworkError, + PermissionError, + ProviderHint, + RepoSpec, + UnsupportedProviderError, +) +from pkgmgr.core.remote_provisioning.visibility import ( + VisibilityOptions, + set_repo_visibility, +) +from pkgmgr.core.remote_provisioning.http.errors import HttpError + + +class TestSetRepoVisibility(unittest.TestCase): + def _mk_provider(self, *, kind: str = "gitea") -> MagicMock: + p = MagicMock() + p.kind = kind + return p + + def _mk_registry( + self, provider: MagicMock | None, providers: list[MagicMock] | None = None + ) -> MagicMock: + reg = MagicMock() + reg.resolve.return_value = provider + reg.providers = ( + providers + if providers is not None + else ([provider] if provider is not None else []) + ) + return reg + + def _mk_token_resolver(self, token: str = "TOKEN") -> MagicMock: + resolver = MagicMock() + tok = MagicMock() + tok.token = token + resolver.get_token.return_value = tok + return resolver + + def test_preview_returns_skipped_and_does_not_call_provider(self) -> None: + provider = self._mk_provider() + reg = self._mk_registry(provider) + resolver = self._mk_token_resolver() + + spec = RepoSpec(host="git.veen.world", owner="me", name="repo", private=True) + + res = set_repo_visibility( + spec, + private=False, + options=VisibilityOptions(preview=True), + registry=reg, + token_resolver=resolver, + ) + + self.assertEqual(res.status, "skipped") + provider.get_repo_private.assert_not_called() + provider.set_repo_private.assert_not_called() + + def test_unsupported_provider_raises(self) -> None: + reg = self._mk_registry(provider=None, providers=[]) + + spec = RepoSpec(host="unknown.host", owner="me", name="repo", private=True) + + with self.assertRaises(UnsupportedProviderError): + set_repo_visibility( + spec, + private=True, + registry=reg, + token_resolver=self._mk_token_resolver(), + ) + + def test_notfound_when_provider_returns_none(self) -> None: + provider = self._mk_provider() + provider.get_repo_private.return_value = None + + reg = self._mk_registry(provider) + resolver = self._mk_token_resolver() + + spec = RepoSpec(host="git.veen.world", owner="me", name="repo", private=True) + + res = set_repo_visibility( + spec, + private=True, + registry=reg, + token_resolver=resolver, + ) + + self.assertEqual(res.status, "notfound") + provider.set_repo_private.assert_not_called() + + def test_noop_when_already_desired(self) -> None: + provider = self._mk_provider() + provider.get_repo_private.return_value = True + + reg = self._mk_registry(provider) + resolver = self._mk_token_resolver() + + spec = RepoSpec(host="git.veen.world", owner="me", name="repo", private=True) + + res = set_repo_visibility( + spec, + private=True, + registry=reg, + token_resolver=resolver, + ) + + self.assertEqual(res.status, "noop") + provider.set_repo_private.assert_not_called() + + def test_updated_when_needs_change(self) -> None: + provider = self._mk_provider() + provider.get_repo_private.return_value = True + + reg = self._mk_registry(provider) + resolver = self._mk_token_resolver() + + spec = RepoSpec(host="git.veen.world", owner="me", name="repo", private=True) + + res = set_repo_visibility( + spec, + private=False, + registry=reg, + token_resolver=resolver, + ) + + self.assertEqual(res.status, "updated") + provider.set_repo_private.assert_called_once() + args, kwargs = provider.set_repo_private.call_args + self.assertEqual(kwargs.get("private"), False) + + def test_provider_hint_overrides_registry_resolution(self) -> None: + # registry.resolve returns gitea provider, but hint forces github provider + gitea = self._mk_provider(kind="gitea") + github = self._mk_provider(kind="github") + github.get_repo_private.return_value = True + + reg = self._mk_registry(gitea, providers=[gitea, github]) + resolver = self._mk_token_resolver() + + spec = RepoSpec(host="github.com", owner="me", name="repo", private=True) + + res = set_repo_visibility( + spec, + private=False, + provider_hint=ProviderHint(kind="github"), + registry=reg, + token_resolver=resolver, + ) + + self.assertEqual(res.status, "updated") + github.get_repo_private.assert_called_once() + gitea.get_repo_private.assert_not_called() + + def test_http_error_401_maps_to_auth_error(self) -> None: + provider = self._mk_provider() + provider.get_repo_private.side_effect = HttpError( + status=401, message="nope", body="" + ) + + reg = self._mk_registry(provider) + resolver = self._mk_token_resolver() + + spec = RepoSpec(host="git.veen.world", owner="me", name="repo", private=True) + + with self.assertRaises(AuthError): + set_repo_visibility( + spec, private=True, registry=reg, token_resolver=resolver + ) + + def test_http_error_403_maps_to_permission_error(self) -> None: + provider = self._mk_provider() + provider.get_repo_private.side_effect = HttpError( + status=403, message="nope", body="" + ) + + reg = self._mk_registry(provider) + resolver = self._mk_token_resolver() + + spec = RepoSpec(host="git.veen.world", owner="me", name="repo", private=True) + + with self.assertRaises(PermissionError): + set_repo_visibility( + spec, private=True, registry=reg, token_resolver=resolver + ) + + def test_http_error_status_0_maps_to_network_error(self) -> None: + provider = self._mk_provider() + provider.get_repo_private.side_effect = HttpError( + status=0, message="connection failed", body="" + ) + + reg = self._mk_registry(provider) + resolver = self._mk_token_resolver() + + spec = RepoSpec(host="git.veen.world", owner="me", name="repo", private=True) + + with self.assertRaises(NetworkError): + set_repo_visibility( + spec, private=True, registry=reg, token_resolver=resolver + ) + + def test_http_error_other_maps_to_network_error(self) -> None: + provider = self._mk_provider() + provider.get_repo_private.side_effect = HttpError( + status=500, message="boom", body="server error" + ) + + reg = self._mk_registry(provider) + resolver = self._mk_token_resolver() + + spec = RepoSpec(host="git.veen.world", owner="me", name="repo", private=True) + + with self.assertRaises(NetworkError): + set_repo_visibility( + spec, private=True, registry=reg, token_resolver=resolver + ) + + +if __name__ == "__main__": + unittest.main()