rune/scripts/create_release.py

#!/usr/bin/env python3
"""Idempotent release creation for the Rune release pipeline (#192).

Extracted from the 'Create release from merged pull request' step in
``.gitea/workflows/auto-version.yml`` so both the pull_request:closed path and
the push-triggered safety net cut releases through the same code. Keeps the
tag-exists idempotency check (retry/replay safety).

Env-driven: GITEA_TOKEN, RELEASE_VERSION, PR_TITLE/PR_BODY (optional — the
push safety net has no PR context and falls back to a generic body),
RELEASE_SERVER_URL + RELEASE_REPO. The release API URLs are derived in-script
from those base coordinates (spec-gate finding 1, #192: the workflow never
passed LIST_API_URL/RELEASE_API_URL — KeyError at runtime on both paths), so
the script is self-sufficient; explicit LIST_API_URL/RELEASE_API_URL env
values still override the derived defaults for direct/manual use.
"""

from __future__ import annotations

import argparse
import json
import os
import sys
import urllib.error
import urllib.request
from typing import Any
import sys as _sys
_SCRIPT_DIR = str(__import__('pathlib').Path(__file__).resolve().parent)
if _SCRIPT_DIR not in _sys.path:
    _sys.path.insert(0, _SCRIPT_DIR)
import gitea_auth

# Env-contract (#192 spec-gate finding 1): derived in-script, never required
# from the caller. Guarded by tests/test_workflow_script_contract.py.
DERIVED_ENV: dict[str, str] = {
    "LIST_API_URL": "RELEASE_SERVER_URL + RELEASE_REPO",
    "RELEASE_API_URL": "RELEASE_SERVER_URL + RELEASE_REPO",
}


def derive_api_urls(server_url: str, repo: str) -> tuple[str, str]:
    """Derive (list_api_url, release_api_url) from the workflow base coordinates."""
    base = server_url.rstrip("/") + "/api/v1/repos/" + repo
    return base + "/releases?limit=50", base + "/releases"


def build_release_payload(version: str, pr_title: str, pr_body: str) -> dict[str, Any]:
    if pr_title:
        body = f"## {pr_title}\n\n{pr_body}"
    else:
        body = "## Release (push-triggered safety net)\n\nCut automatically by the release safety net (#192) after a merge whose closed-event never arrived."
    return {
        "tag_name": f"v{version}",
        "target_commitish": "master",
        "name": f"Rune v{version}",
        "body": body,
        "draft": False,
        "prerelease": False,
    }


def release_exists(releases: list[dict[str, Any]], version: str) -> bool:
    """True when a release for ``v<version>`` already exists (idempotency check)."""
    return any(r.get("tag_name") == f"v{version}" for r in releases)


def list_releases(list_api_url: str, token: str) -> list[dict[str, Any]]:
    req = urllib.request.Request(list_api_url, headers={"Authorization": "token " + token})
    with urllib.request.urlopen(req, timeout=30) as r:
        return json.loads(r.read())


def create_release(release_api_url: str, token: str, payload: dict[str, Any]) -> None:
    req = urllib.request.Request(
        release_api_url,
        data=json.dumps(payload).encode(),
        headers={"Authorization": "token " + token, "Content-Type": "application/json"},
        method="POST",
    )
    with urllib.request.urlopen(req, timeout=60) as r:
        print(f"release created ({r.status})")


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="Idempotent Gitea release creation (#192)")
    parser.add_argument("--dry-run", action="store_true", help="print the payload without executing any API call")
    args = parser.parse_args(argv)

    version = os.environ["RELEASE_VERSION"]
    payload = build_release_payload(version, os.environ.get("PR_TITLE", ""), os.environ.get("PR_BODY", ""))
    if args.dry_run:
        print(json.dumps(payload))
        return 0

    token = gitea_auth.get_token()

    # Derived URLs (explicit env still overrides — for direct/manual use).
    server_url = os.environ["RELEASE_SERVER_URL"]
    repo = os.environ["RELEASE_REPO"]
    list_api_url = os.environ.get("LIST_API_URL") or derive_api_urls(server_url, repo)[0]
    release_api_url = os.environ.get("RELEASE_API_URL") or derive_api_urls(server_url, repo)[1]

    # Skip idempotently if a release for this tag already exists (retry/replay safety).
    releases = list_releases(list_api_url, token)
    if release_exists(releases, version):
        print(f"Release v{version} already exists — skipping creation")
        return 0

    create_release(release_api_url, token, payload)
    return 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except urllib.error.HTTPError as exc:
        print(
            f"create_release: API failed ({exc.code}): {exc.read().decode('utf-8', 'replace')[:300]}", file=sys.stderr
        )
        sys.exit(1)
    except urllib.error.URLError as exc:
        # Connection refused / DNS failure etc. (HTTPError is a URLError
        # subclass, so this only fires for transport-level failures).
        print(f"create_release: API failed (transport error): {exc.reason}", file=sys.stderr)
        sys.exit(1)