tegwick
f9d957a221
reuse_surface/stats.py: _hub_summary() now reports composed_at, stale,
age_days, freshness_threshold_days (REUSE_SURFACE_FRESHNESS_DAYS env,
default 7), and a computed stale_warning. New hub_client.hub_federated()
backs it. format_stats_markdown surfaces a STALE marker when triggered.
.forgejo/workflows/ci.yml: new informational (non-failing) hub freshness
check against the live production hub on every push -- prints a
:⚠️: annotation when stale, never fails the build.
docs/RegistryFederation.md: new section tying together the webhook (T02),
scheduled fallback (T03), and freshness visibility (T06) into one
explanation. docs/deploy/reuse-kubernetes.md: updated for the T03 Forgejo
migration and the now-automated image.yaml build; image promotion
checklist updated for the known /health ingress bug (verify via
/v1/repos or /v1/federated instead).
14 new pytest cases, 173 total pass. Live-verified against production:
reuse-surface stats correctly showed composed_at/age_days for the real
federated index. Separately discovered and confirmed (via a live signed
webhook test) that reuse-surface-env moving to ExternalSecret/OpenBao
custody (railiance-apps commit 706f6c7, found while updating these docs)
did not break the T02/T03 webhook -- the synced value still matches what
the hub actually uses.
REUSE-WP-0019 is now fully complete (T01-T06). SCOPE.md and
docs/IntentScopeGapAnalysis.md updated to reflect closure.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
113 lines
No EOL
3.4 KiB
Python
113 lines
No EOL
3.4 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import urllib.error
|
|
import urllib.request
|
|
from typing import Any
|
|
|
|
|
|
def service_base_url(explicit: str | None = None) -> str:
|
|
base = (explicit or os.environ.get("REUSE_SURFACE_URL", "")).rstrip("/")
|
|
if not base:
|
|
raise ValueError(
|
|
"service URL not configured; set REUSE_SURFACE_URL or pass --base-url"
|
|
)
|
|
return base
|
|
|
|
|
|
def service_token() -> str | None:
|
|
return os.environ.get("REUSE_SURFACE_TOKEN")
|
|
|
|
|
|
def _request(
|
|
method: str,
|
|
url: str,
|
|
*,
|
|
token: str | None = None,
|
|
body: dict[str, Any] | None = None,
|
|
) -> tuple[int, Any]:
|
|
headers = {"Accept": "application/json", "User-Agent": "reuse-surface/0.1"}
|
|
data = None
|
|
if body is not None:
|
|
headers["Content-Type"] = "application/json"
|
|
data = json.dumps(body).encode("utf-8")
|
|
if token:
|
|
headers["Authorization"] = f"Bearer {token}"
|
|
request = urllib.request.Request(url, data=data, headers=headers, method=method)
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=30) as response:
|
|
raw = response.read().decode("utf-8")
|
|
return response.status, json.loads(raw) if raw else None
|
|
except urllib.error.HTTPError as exc:
|
|
raw = exc.read().decode("utf-8")
|
|
try:
|
|
payload = json.loads(raw) if raw else {"message": exc.reason}
|
|
except json.JSONDecodeError:
|
|
payload = {"message": raw or exc.reason}
|
|
return exc.code, payload
|
|
|
|
|
|
def hub_status(base_url: str | None = None) -> tuple[int, Any]:
|
|
return _request("GET", f"{service_base_url(base_url)}/health")
|
|
|
|
|
|
def hub_list(base_url: str | None = None) -> tuple[int, Any]:
|
|
return _request("GET", f"{service_base_url(base_url)}/v1/repos")
|
|
|
|
|
|
def hub_federated(base_url: str | None = None) -> tuple[int, Any]:
|
|
return _request("GET", f"{service_base_url(base_url)}/v1/federated")
|
|
|
|
|
|
def hub_show(repo: str, base_url: str | None = None) -> tuple[int, Any]:
|
|
return _request("GET", f"{service_base_url(base_url)}/v1/repos/{repo}")
|
|
|
|
|
|
def hub_register(payload: dict[str, Any], base_url: str | None = None) -> tuple[int, Any]:
|
|
token = service_token()
|
|
if not token:
|
|
raise ValueError("REUSE_SURFACE_TOKEN is required for register")
|
|
return _request(
|
|
"POST",
|
|
f"{service_base_url(base_url)}/v1/repos",
|
|
token=token,
|
|
body=payload,
|
|
)
|
|
|
|
|
|
def hub_update(
|
|
repo: str, payload: dict[str, Any], base_url: str | None = None
|
|
) -> tuple[int, Any]:
|
|
token = service_token()
|
|
if not token:
|
|
raise ValueError("REUSE_SURFACE_TOKEN is required for update")
|
|
return _request(
|
|
"PATCH",
|
|
f"{service_base_url(base_url)}/v1/repos/{repo}",
|
|
token=token,
|
|
body=payload,
|
|
)
|
|
|
|
|
|
def hub_record_reuse_event(
|
|
payload: dict[str, Any], base_url: str | None = None
|
|
) -> tuple[int, Any]:
|
|
token = service_token()
|
|
if not token:
|
|
raise ValueError("REUSE_SURFACE_TOKEN is required for record-reuse")
|
|
return _request(
|
|
"POST",
|
|
f"{service_base_url(base_url)}/v1/reuse-events",
|
|
token=token,
|
|
body=payload,
|
|
)
|
|
|
|
|
|
def hub_list_reuse_events(
|
|
capability_id: str | None = None, base_url: str | None = None
|
|
) -> tuple[int, Any]:
|
|
url = f"{service_base_url(base_url)}/v1/reuse-events"
|
|
if capability_id:
|
|
url += f"?capability_id={capability_id}"
|
|
return _request("GET", url) |