fix(forgejo-package-prune): correct version enumeration + network robustness
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 2s

Forgejo has no per-package /versions endpoint — the list endpoint returns one
entry per (name, version). build_delete_plans now groups list results by name
instead of calling a 404ing /versions sub-path. Removed dead list_versions/_paginate.
Added retry + longer timeout to _api_request for slow/large registry listings
(container list was timing out). Updated fixture test to the grouped model.

Dry-run now clean: 29 candidate deletions across 5 container packages, 0 errors
(was 53× HTTP 404). ACTIVITY-WP-0020 T02 fix; enable/apply still gated on
protection-coverage review (T05).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
tegwick 2026-07-16 02:46:15 +02:00
parent 20bf3d5942
commit 6f7ca31a6d
2 changed files with 43 additions and 73 deletions

View file

@ -8,6 +8,7 @@ import json
import os
import re
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
@ -111,7 +112,8 @@ def _api_request(
url: str,
token: str,
*,
timeout: float = 30.0,
timeout: float = 60.0,
retries: int = 2,
) -> Any:
req = urllib.request.Request(
url,
@ -121,37 +123,23 @@ def _api_request(
"Accept": "application/json",
},
)
with urllib.request.urlopen(req, timeout=timeout) as resp:
body = resp.read().decode("utf-8")
return json.loads(body) if body else None
def _paginate(
base_url: str,
token: str,
path: str,
*,
page_size: int = 50,
) -> list[dict[str, Any]]:
items: list[dict[str, Any]] = []
page = 1
while True:
query = urllib.parse.urlencode({"limit": page_size, "page": page})
url = f"{base_url.rstrip('/')}{path}?{query}"
payload = _api_request("GET", url, token)
if not payload:
break
if isinstance(payload, list):
batch = payload
else:
batch = payload.get("data") or payload.get("items") or []
if not batch:
break
items.extend(batch)
if len(batch) < page_size:
break
page += 1
return items
last_exc: Exception | None = None
for attempt in range(retries + 1):
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
body = resp.read().decode("utf-8")
return json.loads(body) if body else None
except urllib.error.HTTPError:
raise # 4xx/5xx are real responses — surface them, do not retry
except (urllib.error.URLError, TimeoutError, OSError) as exc:
# Transient: connection reset, read timeout, TLS handshake timeout.
last_exc = exc
if attempt < retries:
time.sleep(2 * (attempt + 1))
continue
raise
if last_exc: # pragma: no cover - defensive
raise last_exc
def list_packages(
@ -180,23 +168,6 @@ def list_packages(
return items
def list_versions(
base_url: str,
token: str,
owner: str,
package_type: str,
name: str,
) -> list[dict[str, Any]]:
owner_q = urllib.parse.quote(owner)
type_q = urllib.parse.quote(package_type)
name_q = urllib.parse.quote(name, safe="")
return _paginate(
base_url,
token,
f"/api/v1/packages/{owner_q}/{type_q}/{name_q}/versions",
)
def delete_version(
base_url: str,
token: str,
@ -240,19 +211,17 @@ def build_delete_plans(
errors.append(f"list {package_type}: {exc}")
continue
# Forgejo's package list endpoint returns one entry per (name, version).
# Group by package name; each group is that package's version set — there
# is no separate per-package "/versions" endpoint.
by_name: dict[str, list[dict[str, Any]]] = {}
for package in packages:
name = str(package.get("name") or package.get("package_name") or "")
if not name:
continue
try:
versions = list_versions(base_url, token, owner, package_type, name)
except urllib.error.HTTPError as exc:
errors.append(f"versions {package_type}/{name}: HTTP {exc.code}")
continue
except Exception as exc: # noqa: BLE001
errors.append(f"versions {package_type}/{name}: {exc}")
continue
by_name.setdefault(name, []).append(package)
for name, versions in by_name.items():
sorted_versions = sorted(
versions,
key=lambda item: _parse_created_at(str(item.get("created_at") or "")),

View file

@ -37,24 +37,25 @@ image:
def test_build_delete_plans_respects_depth_and_protection(self) -> None:
protected = {("container", "vergabe-teilnahme", "old-prod")}
versions = [
{"version": "v4", "created_at": "2026-07-12T10:00:00Z"},
{"version": "v3", "created_at": "2026-07-11T10:00:00Z"},
{"version": "v2", "created_at": "2026-07-10T10:00:00Z"},
{"version": "v1", "created_at": "2026-07-09T10:00:00Z"},
{"version": "old-prod", "created_at": "2026-07-08T10:00:00Z"},
# Forgejo's package list endpoint returns one entry per (name, version);
# build_delete_plans groups these by name (no separate /versions call).
packages = [
{"name": "vergabe-teilnahme", "version": "v4", "created_at": "2026-07-12T10:00:00Z"},
{"name": "vergabe-teilnahme", "version": "v3", "created_at": "2026-07-11T10:00:00Z"},
{"name": "vergabe-teilnahme", "version": "v2", "created_at": "2026-07-10T10:00:00Z"},
{"name": "vergabe-teilnahme", "version": "v1", "created_at": "2026-07-09T10:00:00Z"},
{"name": "vergabe-teilnahme", "version": "old-prod", "created_at": "2026-07-08T10:00:00Z"},
]
with mock.patch.object(prune, "list_packages", return_value=[{"name": "vergabe-teilnahme"}]):
with mock.patch.object(prune, "list_versions", return_value=versions):
plans, errors = prune.build_delete_plans(
base_url="https://forgejo.example",
token="token",
owner="coulomb",
package_types=["container"],
max_versions=3,
protected=protected,
)
with mock.patch.object(prune, "list_packages", return_value=packages):
plans, errors = prune.build_delete_plans(
base_url="https://forgejo.example",
token="token",
owner="coulomb",
package_types=["container"],
max_versions=3,
protected=protected,
)
self.assertEqual(errors, [])
deletable = {(p.name, p.version) for p in plans if not p.protected}