generated from coulomb/repo-seed
REUSE-WP-0019-T01: forge host abstraction + URL migration inventory
Some checks failed
ci / validate-registry (push) Has been cancelled
Some checks failed
ci / validate-registry (push) Has been cancelled
reuse_surface/forge_host.py: parse/derive/rewrite raw index URLs across Gitea and Forgejo (handles both the legacy /raw/<branch>/... form and the canonical /raw/branch/<branch>/... form both forges serve without a 303 redirect). migrate_source_host() verifies the new URL resolves via HTTP HEAD before writing -- refuses to point a repo at a host it hasn't actually migrated to. New CLI: reuse-surface federation migrate-host --repo <slug> --to <base-url> [--from <check>] [--dry-run] [--no-verify] [--update-hub]. Inventory: cross-referenced sources.yaml against each repo's actual git origin. 11/61 repos already on Forgejo; found 2 with stale sources.yaml entries (activity-core, state-hub) despite having migrated. Fixed for real: local sources.yaml + production hub registration (hub update), verified against GET /v1/federated post-migration. config-atlas's WP-0017-T06 303 confirmed NOT a host-transition symptom (already diagnosed there as something else). Also fixed two host-agnostic gaps found while inventorying: registry_update.py and maintain_llm.py only recognized .gitea/workflows/, missing repos already on .forgejo/workflows/. Fixed a stale copy-paste example (state-hub's now-wrong old URL) in docs/RegistryFederation.md, and a pre-existing unrelated port typo (8088 vs the real llm-connect default 8080) in tools/README.md and registry/README.md. 17 new pytest cases (tests/test_forge_host.py), 106 total pass. Recomposed federated.yaml post-migration: still 61 capabilities. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
105
reuse_surface/forge_host.py
Normal file
105
reuse_surface/forge_host.py
Normal file
@@ -0,0 +1,105 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import re
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
|
||||
# Matches both the legacy Gitea-default form (.../raw/<branch>/<path>) and the
|
||||
# canonical branch-qualified form (.../raw/branch/<branch>/<path>) that both
|
||||
# Gitea and Forgejo serve without a 303 redirect. See REUSE-WP-0017-T06 for
|
||||
# why the redirect matters (urllib follows it fine; some HEAD-only probes
|
||||
# and shell tooling don't).
|
||||
_RAW_URL_RE = re.compile(
|
||||
r"^(?P<scheme>https?)://(?P<host>[^/]+)/(?P<org>[^/]+)/(?P<repo>[^/]+)"
|
||||
r"/raw/(?:branch/)?(?P<branch>[^/]+)/(?P<path>.+)$"
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class RawUrlParts:
|
||||
scheme: str
|
||||
host: str
|
||||
org: str
|
||||
repo: str
|
||||
branch: str
|
||||
path: str
|
||||
|
||||
|
||||
def parse_raw_url(url: str) -> RawUrlParts:
|
||||
match = _RAW_URL_RE.match(url)
|
||||
if not match:
|
||||
raise ValueError(f"not a recognized Gitea/Forgejo raw URL: {url}")
|
||||
return RawUrlParts(**match.groupdict())
|
||||
|
||||
|
||||
def derive_raw_url(base_url: str, org: str, repo: str, *, branch: str = "main", path: str = "registry/indexes/capabilities.yaml") -> str:
|
||||
"""Builds a raw index URL in the canonical branch-qualified form, which
|
||||
both Gitea and Forgejo serve directly (no 303 redirect)."""
|
||||
base = base_url.rstrip("/")
|
||||
return f"{base}/{org}/{repo}/raw/branch/{branch}/{path}"
|
||||
|
||||
|
||||
def rewrite_url_host(url: str, new_base_url: str) -> str:
|
||||
"""Rewrites a raw index URL's scheme+host to new_base_url, preserving
|
||||
org/repo/branch/path and normalizing to the canonical branch-qualified
|
||||
form regardless of which form the input used."""
|
||||
parts = parse_raw_url(url)
|
||||
return derive_raw_url(new_base_url, parts.org, parts.repo, branch=parts.branch, path=parts.path)
|
||||
|
||||
|
||||
def forge_base_url(explicit: str | None = None) -> str | None:
|
||||
"""Returns the configured default Forgejo base URL, or None if unset.
|
||||
Unlike hub_client.service_base_url, this has no hard requirement --
|
||||
per-repo migration is opt-in (most repos haven't moved their remote
|
||||
yet), so an unset value just means 'no default target configured',
|
||||
not an error."""
|
||||
return explicit or os.environ.get("REUSE_SURFACE_FORGE_BASE_URL") or None
|
||||
|
||||
|
||||
def probe_url(url: str, *, timeout: int = 10) -> tuple[bool, int | None]:
|
||||
"""HEAD-probes a URL, following redirects (matches establish.py's
|
||||
_probe_raw_url behavior). Returns (ok, status)."""
|
||||
request = urllib.request.Request(url, method="HEAD", headers={"User-Agent": "reuse-surface/0.1"})
|
||||
try:
|
||||
with urllib.request.urlopen(request, timeout=timeout) as response:
|
||||
return response.status == 200, response.status
|
||||
except urllib.error.HTTPError as exc:
|
||||
return False, exc.code
|
||||
except (urllib.error.URLError, TimeoutError, OSError):
|
||||
return False, None
|
||||
|
||||
|
||||
def migrate_source_host(
|
||||
sources: list[dict[str, Any]],
|
||||
repo: str,
|
||||
new_base_url: str,
|
||||
*,
|
||||
verify: bool = True,
|
||||
) -> dict[str, Any]:
|
||||
"""Rewrites one repo's entry in a sources.yaml `sources` list to point at
|
||||
new_base_url. Mutates the matching entry in place and returns a report
|
||||
dict. Raises ValueError if the repo isn't found, the existing URL can't
|
||||
be parsed, or (when verify=True) the new URL doesn't resolve -- never
|
||||
writes a URL that hasn't been confirmed reachable, since rewriting to a
|
||||
host the repo hasn't actually migrated to would silently break
|
||||
federation for it."""
|
||||
entry = next((s for s in sources if s.get("repo") == repo), None)
|
||||
if entry is None:
|
||||
raise ValueError(f"repo not found in sources: {repo}")
|
||||
old_url = entry["url"]
|
||||
new_url = rewrite_url_host(old_url, new_base_url)
|
||||
report: dict[str, Any] = {"repo": repo, "old_url": old_url, "new_url": new_url, "verified": None}
|
||||
if verify:
|
||||
ok, status = probe_url(new_url)
|
||||
report["verified"] = ok
|
||||
report["probe_status"] = status
|
||||
if not ok:
|
||||
raise ValueError(
|
||||
f"{repo}: new URL did not resolve (HTTP {status}); "
|
||||
f"not writing an unverified URL: {new_url}"
|
||||
)
|
||||
entry["url"] = new_url
|
||||
return report
|
||||
Reference in New Issue
Block a user