image-tools/image_tools/bump.py

"""Move a pin in the repos that use an image, and start their own run (#1308).

When an image improves, its repo opens or updates one pull request in every
dependent (``dependents.toml``), rewriting only the pin, then dispatches the
dependent's workflow, which rebuilds on the new pin and verifies itself.

    python -m image_tools.bump <image> <new-ref> <version> --report report.md

Talks to Gitea with ``BUMP_TOKEN`` (the netork-bot account); standard library only.
"""

from __future__ import annotations

import argparse
import base64
import json
import os
import re
import tomllib
import urllib.error
import urllib.request
from collections.abc import Callable
from dataclasses import dataclass
from pathlib import Path

GITEA_URL = os.environ.get("GITEA_URL", "https://git.netork.io")
REGISTRY = "registry.netork.io"
#: A pinned reference: registry/repo:tag@sha256:<64 hex>.
_REF = re.compile(r"^registry\.netork\.io/[a-z0-9./-]+:[A-Za-z0-9._-]+@sha256:[0-9a-f]{64}$")

Api = Callable[..., object]


@dataclass(frozen=True)
class Dependent:
    repo: str
    files: tuple[str, ...]
    workflow: str


def load_dependents(path: Path) -> list[Dependent]:
    data = tomllib.loads(path.read_text())
    return [
        Dependent(d["repo"], tuple(d["files"]), d["workflow"]) for d in data.get("dependent", [])
    ]


def rewrite_pin(text: str, image: str, new_ref: str) -> str:
    """*text* with every pinned reference of *image* replaced by *new_ref*."""
    if not _REF.match(new_ref):
        raise ValueError(f"not a pinned reference: {new_ref!r}")
    # The host may be written as ${REGISTRY_HOST:-registry.netork.io} (netork's
    # compose files); that spelling stays, only the path, tag and digest move.
    pattern = re.compile(
        rf"({re.escape(REGISTRY)}\}}?/){re.escape(image)}:[A-Za-z0-9._-]+@sha256:[0-9a-f]{{64}}"
    )
    tail = new_ref.split("/", 1)[1]
    return pattern.sub(lambda m: m.group(1) + tail, text)


def _branch(image: str, version: str) -> str:
    return f"chore/{image.rsplit('/', 1)[-1]}-{version}"


def open_or_update_pr(
    api: Api, dep: Dependent, image: str, new_ref: str, version: str, report: str
) -> dict | None:
    """One PR in *dep* moving *image* to *new_ref*; None when no file pins it.

    A second run for the same version works on the branch the first one made:
    files already moved stay as they are, and the open PR gets the new report.
    """
    owner_repo = f"/repos/{dep.repo}"
    # Gitea answers 404 for a repo the token's account may not see, and for its
    # files too; that read as "no pin to move" for every dependent (2026-10-22).
    try:
        api("GET", owner_repo)
    except FileNotFoundError:
        raise PermissionError(f"{dep.repo} is not visible to the bump token's account") from None
    branch = _branch(image, version)
    try:
        api("GET", f"{owner_repo}/branches/{branch}")
        branch_exists = True
    except (FileNotFoundError, urllib.error.HTTPError):
        branch_exists = False
    ref = branch if branch_exists else "main"

    changes = {}
    for name in dep.files:
        try:
            current = api("GET", f"{owner_repo}/contents/{name}?ref={ref}")
        except FileNotFoundError:
            continue  # not there yet (a repo still being set up): no pin to move
        text = base64.b64decode(current["content"]).decode()
        moved = rewrite_pin(text, image, new_ref)
        if moved != text:
            changes[name] = (moved, current["sha"])
    if not changes and not branch_exists:
        return None

    if not branch_exists:
        api(
            "POST", f"{owner_repo}/branches", {"new_branch_name": branch, "old_branch_name": "main"}
        )
    title = f"chore(images): move {image} to {version}"
    for name, (text, sha) in changes.items():
        api(
            "PUT",
            f"{owner_repo}/contents/{name}",
            {
                "content": base64.b64encode(text.encode()).decode(),
                "sha": sha,
                "branch": branch,
                "message": title,
            },
        )

    open_prs = api("GET", f"{owner_repo}/pulls?state=open&limit=50")
    for pr in open_prs:
        if pr["head"]["ref"] == branch:
            return api("PATCH", f"{owner_repo}/pulls/{pr['number']}", {"body": report})
    return api(
        "POST",
        f"{owner_repo}/pulls",
        {"head": branch, "base": "main", "title": title, "body": report},
    )


def dispatch(api: Api, dep: Dependent, inputs: dict[str, str]) -> None:
    """Start *dep*'s own workflow on main; one without a workflow gets the PR only."""
    if not dep.workflow:
        return
    api(
        "POST",
        f"/repos/{dep.repo}/actions/workflows/{dep.workflow}/dispatches",
        {"ref": "main", "inputs": inputs},
    )


def bump_all(
    api: Api,
    deps: list[Dependent],
    image: str,
    new_ref: str,
    version: str,
    report: str,
    say: Callable[[str], None],
) -> bool:
    """A pull request in every dependent and its run started; False when one failed.

    A repo the token cannot see, or a workflow that does not exist, is named and
    the others go on: one wrong entry must not keep the rest at the old pin.
    """
    ok = True
    for dep in deps:
        try:
            pr = open_or_update_pr(api, dep, image, new_ref, version, report)
        except PermissionError as exc:
            say(f"{dep.repo}: {exc}")
            ok = False
            continue
        say(f"{dep.repo}: {'PR #' + str(pr['number']) if pr else 'no pin to move'}")
        if not pr:
            continue
        try:
            dispatch(api, dep, {"base_ref": new_ref})
        except FileNotFoundError:
            say(f"{dep.repo}: workflow {dep.workflow} not found; its pull request stands")
            ok = False
    return ok


def gitea(method: str, path: str, body: dict | None = None):  # pragma: no cover - network seam
    data = json.dumps(body).encode() if body is not None else None
    request = urllib.request.Request(
        f"{GITEA_URL}/api/v1{path}",
        data=data,
        method=method,
        headers={
            "Authorization": f"token {os.environ['BUMP_TOKEN']}",
            "Content-Type": "application/json",
        },
    )
    try:
        with urllib.request.urlopen(request, timeout=60) as response:  # noqa: S310 - https
            raw = response.read()
    except urllib.error.HTTPError as exc:
        if exc.code == 404:
            raise FileNotFoundError(path) from exc
        raise
    return json.loads(raw) if raw else {}


def main() -> int:  # pragma: no cover - CLI seam
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("image", help="e.g. netork/base-os")
    parser.add_argument("new_ref")
    parser.add_argument("version")
    parser.add_argument("--report", required=True)
    parser.add_argument("--dependents", default="dependents.toml")
    args = parser.parse_args()
    report = Path(args.report).read_text()
    deps = load_dependents(Path(args.dependents))
    ok = bump_all(gitea, deps, args.image, args.new_ref, args.version, report, print)
    return 0 if ok else 1


if __name__ == "__main__":  # pragma: no cover
    raise SystemExit(main())