Complete WP-0015: forgejo_hubs migration + closeout (T05/T07)

migrations/0007_forgejo_hubs.sql: table forgejo_hubs, auto-populated
via a BEFORE INSERT trigger on phase_manifests that reads repo_hub/
repo_hub_uri straight out of the manifest JSONB (no top-level columns
needed). ON CONFLICT DO NOTHING -- a hub already seen is left alone;
correcting a URI is a SECURITY DEFINER governance action
(correct_forgejo_hub_uri), not a plain UPDATE, matching every other
governance-action pattern in this project. Thin Python wrappers added
to registry.py.

tests/test_forgejo_hubs.py (6 Docker-gated tests) and
tests/test_reference_docs.py (13 tests, no Docker needed -- smoke-tests
every real specs/policies/specs/profiles/ file, not just the two
exercised incidentally by T03's Control Plane tests).

All seven WP-0015 tasks done; workplan marked finished. Final suite:
94 passing offline, 183 passing under the service extras venv. No
stray Docker containers left running.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
tegwick 2026-08-03 23:57:46 +02:00
parent eaea89e125
commit de308f8947
6 changed files with 419 additions and 5 deletions

View file

@ -29,7 +29,7 @@ from dataclasses import dataclass
from typing import Any
from psycopg import Connection
from psycopg.errors import UniqueViolation
from psycopg.errors import RaiseException, UniqueViolation
from psycopg.types.json import Jsonb
from . import validation
@ -296,3 +296,45 @@ def promote_extension_canonical(
"SELECT set_extension_status(%s, %s, %s, %s)",
(extension_id, version, "canonical", approved_by),
)
def get_forgejo_hub(conn: Connection, hub_slug: str) -> dict[str, Any] | None:
"""Look up a hub's currently recorded service URI (WP-0015-T05).
Purely an admin/repair convenience a Phase Manifest never needs
this to be reachable to be independently verified, since it already
carries `repo_hub_uri`/`repo_name` directly at registration time
(`specs/PhaseLifecycleUseCases.md` use case 9). Rows are created
automatically by `phase_manifests_ensure_forgejo_hub`
(`migrations/0007_forgejo_hubs.sql`) the first time a hub is seen
there is no explicit "register a hub" function to pair with this one.
"""
row = conn.execute(
"SELECT hub_slug, service_uri, first_seen_at, updated_at "
"FROM forgejo_hubs WHERE hub_slug = %s",
(hub_slug,),
).fetchone()
if row is None:
return None
hub_slug_, service_uri, first_seen_at, updated_at = row
return {
"hub_slug": hub_slug_,
"service_uri": service_uri,
"first_seen_at": first_seen_at,
"updated_at": updated_at,
}
def correct_forgejo_hub_uri(conn: Connection, hub_slug: str, new_uri: str) -> None:
"""Correct a hub's recorded service URI (e.g. after a domain move).
Calls the database's `correct_forgejo_hub_uri` function rather than
an UPDATE the application role has no UPDATE grant on
`forgejo_hubs` at all, matching every other governance-action pattern
in this project (`revoke_sub_credential`,
`promote_extension_canonical`).
"""
try:
conn.execute("SELECT correct_forgejo_hub_uri(%s, %s)", (hub_slug, new_uri))
except RaiseException as exc:
raise RegistrationError(f"unknown forgejo hub: {hub_slug!r}") from exc