fix(test): answer package probes from repo indexes

The per-package endpoints are anti-bot territory: packages.debian.org shed
request-level 503s for >13h (2 CI runs and 2 local quality gates red, 7
resp. 91 probes unverified), and archlinux.org connection resets are the
chronic class 00784105f already fought. One Packages.xz/.db fetch per
distro answers every probe from a CDN meant for machines, through the
existing _get retry/semaphore layer; the index is cached per run behind a
lock, and an unreadable index stays a loud unknown.

Debian goes over http because the dev container hijacks deb.debian.org
onto the package-cache frontend, which serves the name on port 80 only -
the same transport the stack's own apt list uses. The stricter index also
exposed three registry declarations that never existed in Debian stable;
they are fixed in the follow-up commit.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
kevinveenbirkenbachandClaude Opus 5 committed 2026-08-24 11:27:42 +02:00
1 parent 4e1e21bbf5
commit bd84664e8d
1 file changed
+125 -12
+125 -12
View File
@@ -11,8 +11,11 @@ mirror says nothing about the declaration.
from __future__ import annotations
import io
import json
import lzma
import re
import tarfile
import threading
import time
import urllib.error
@@ -76,17 +79,17 @@ def _get(url: str) -> tuple[int, bytes]:
"""Fetch one index URL, serialising and retrying per host.
Args:
url: absolute https URL of a package index.
url: absolute http(s) URL of a package index.
Returns:
The HTTP status and body; the body is empty for an error status.
"""
request = urllib.request.Request(url, headers={"User-Agent": _USER_AGENT}) # noqa: S310 - literal https package index URLs only
request = urllib.request.Request(url, headers={"User-Agent": _USER_AGENT}) # noqa: S310 - literal http(s) package index URLs only
host = urllib.parse.urlsplit(url).hostname or ""
with _HOST_LOCKS.setdefault(host, threading.Semaphore(_PER_HOST_REQUESTS)):
for attempt in range(_RETRY_ATTEMPTS):
try:
with urllib.request.urlopen(request, timeout=_TIMEOUT) as response: # noqa: S310 - literal https package index URLs only
with urllib.request.urlopen(request, timeout=_TIMEOUT) as response: # noqa: S310 - literal http(s) package index URLs only
return response.getcode(), response.read()
except urllib.error.HTTPError as exc:
if exc.code not in _RETRY_STATUS or attempt == _RETRY_ATTEMPTS - 1:
@@ -98,13 +101,70 @@ def _get(url: str) -> tuple[int, bytes]:
raise AssertionError("unreachable")
ARCH_MIRROR = "geo.mirror.pkgbuild.com"
_ARCH_REPOS = ("core", "extra", "multilib")
_ARCH_INDEX_CACHE: dict[str, set[str] | str] = {}
_ARCH_INDEX_LOCK = threading.Lock()
def _desc_names(desc: str) -> set[str]:
names: set[str] = set()
section = ""
for line in desc.splitlines():
if line.startswith("%") and line.endswith("%"):
section = line
elif line and section in ("%NAME%", "%PROVIDES%"):
names.add(line.split("=")[0].split("<")[0].strip())
return names
def _load_arch_index() -> None:
names: set[str] = set()
for repo in _ARCH_REPOS:
url = f"https://{ARCH_MIRROR}/{repo}/os/x86_64/{repo}.db"
try:
status, body = _get(url)
except OSError as exc:
_ARCH_INDEX_CACHE["error"] = f"{ARCH_MIRROR}: {exc}"
return
if status != 200:
_ARCH_INDEX_CACHE["error"] = f"{ARCH_MIRROR} returned HTTP {status}"
return
with tarfile.open(fileobj=io.BytesIO(body), mode="r:*") as archive:
for member in archive:
if not member.name.endswith("/desc"):
continue
handle = archive.extractfile(member)
if handle is None:
continue
names |= _desc_names(handle.read().decode("utf-8", "replace"))
_ARCH_INDEX_CACHE["names"] = names
def _arch_index_names() -> tuple[set[str] | None, str]:
"""Load the Arch repo databases once per run and answer from memory.
Returns:
The set of package and provided names in core, extra and multilib,
or None together with the fetch error when a database could not be
read.
"""
with _ARCH_INDEX_LOCK:
if not _ARCH_INDEX_CACHE:
_load_arch_index()
error = _ARCH_INDEX_CACHE.get("error")
if error is not None:
return None, str(error)
names = _ARCH_INDEX_CACHE["names"]
assert isinstance(names, set)
return names, f"{ARCH_MIRROR} core+extra+multilib"
def _probe_arch(name: str) -> tuple[bool | None, str]:
query = urllib.parse.urlencode({"name": name})
status, body = _get(f"https://archlinux.org/packages/search/json/?{query}")
if status != 200:
return None, f"archlinux.org returned HTTP {status}"
payload = json.loads(body or b"{}")
return bool(payload.get("results")), "official repositories"
names, detail = _arch_index_names()
if names is None:
return None, detail
return name in names, detail
def _probe_aur(name: str) -> tuple[bool | None, str]:
@@ -116,6 +176,61 @@ def _probe_aur(name: str) -> tuple[bool | None, str]:
return bool(payload.get("results")), "AUR"
DEBIAN_MIRROR = "deb.debian.org"
_DEBIAN_COMPONENTS = ("main", "contrib", "non-free", "non-free-firmware")
_DEBIAN_INDEX_CACHE: dict[str, set[str] | str] = {}
_DEBIAN_INDEX_LOCK = threading.Lock()
def _load_debian_index() -> None:
names: set[str] = set()
for component in _DEBIAN_COMPONENTS:
url = (
f"http://{DEBIAN_MIRROR}/debian/dists/{DEBIAN_SUITE}"
f"/{component}/binary-amd64/Packages.xz"
)
try:
status, body = _get(url)
except OSError as exc:
_DEBIAN_INDEX_CACHE["error"] = f"{DEBIAN_MIRROR}: {exc}"
return
if status != 200:
_DEBIAN_INDEX_CACHE["error"] = f"{DEBIAN_MIRROR} returned HTTP {status}"
return
for line in lzma.decompress(body).decode("utf-8", "replace").splitlines():
if line.startswith("Package: "):
names.add(line[len("Package: ") :].strip())
elif line.startswith("Provides: "):
for entry in line[len("Provides: ") :].split(","):
names.add(entry.split("(")[0].strip())
_DEBIAN_INDEX_CACHE["names"] = names
def _debian_index_names() -> tuple[set[str] | None, str]:
"""Load the Debian binary index once per run and answer from memory.
Returns:
The set of binary and virtual package names in the suite, or None
together with the fetch error when the mirror could not be read.
"""
with _DEBIAN_INDEX_LOCK:
if not _DEBIAN_INDEX_CACHE:
_load_debian_index()
error = _DEBIAN_INDEX_CACHE.get("error")
if error is not None:
return None, str(error)
names = _DEBIAN_INDEX_CACHE["names"]
assert isinstance(names, set)
return names, f"{DEBIAN_MIRROR}/{DEBIAN_SUITE}"
def _probe_debian(name: str) -> tuple[bool | None, str]:
names, detail = _debian_index_names()
if names is None:
return None, detail
return name in names, detail
def _probe_debian_like(host: str, suite: str, name: str) -> tuple[bool | None, str]:
status, body = _get(f"https://{host}/{suite}/{urllib.parse.quote(name)}")
if status == 404:
@@ -202,9 +317,7 @@ def _probe(probe: Probe) -> Outcome:
elif probe.distro == "arch":
available, detail = _probe_arch(probe.name)
elif probe.distro == "debian":
available, detail = _probe_debian_like(
"packages.debian.org", DEBIAN_SUITE, probe.name
)
available, detail = _probe_debian(probe.name)
elif probe.distro == "ubuntu":
available, detail = _probe_debian_like(
"packages.ubuntu.com", UBUNTU_SUITE, probe.name