weblib-ci/sync_blocked_label.py

#!/usr/bin/env python3
"""Keep `Status/Blocked` in step with Gitea's dependency graph.

Gitea knows which issues block which, and will grey out a pull request's merge
button accordingly, but it does not surface that as a label -- so a list view
gives no hint that half of it cannot be started. This reconciles the two.

Standard library only, so it runs in any runner image and can be driven by
hand.

## The one rule worth arguing about

**An issue with no dependencies at all is never touched.** Only issues that
have at least one dependency are managed:

* any blocker not positively closed -> ensure the label is present
* every blocker closed              -> ensure the label is absent

The asymmetry is deliberate. A dependency in *another* repo is not fully
visible to the Actions token, which is scoped to one repo, so "not open" and
"closed" are not the same statement.

That matters because `Status/Blocked` is also applied by hand for reasons the
graph knows nothing about -- weblib-archive#29 and #30 are blocked on a
decision, not on an issue. A rule of "no open blockers means not blocked"
would strip those the first time it ran. Anything with a dependency edge is
fair game because the edge is the statement of intent; anything without one is
somebody's judgement and stays.

Usage:
    sync_blocked_label.py --repo weblib/cfbypass [--repo weblib/weblib-archive]
                          [--dry-run] [--label Status/Blocked]

Reads the token from --token, else $GITEA_TOKEN, else $GITHUB_TOKEN.
"""

import argparse
import json
import os
import sys
import urllib.error
import urllib.parse
import urllib.request

DEFAULT_LABEL = "Status/Blocked"
DEFAULT_HOST = "https://git.chaosbit.de"


class Forge:
    def __init__(self, base, token):
        self.base = base.rstrip("/") + "/api/v1"
        self.token = token

    def _call(self, method, path, body=None):
        url = f"{self.base}/{path.lstrip('/')}"
        data = json.dumps(body).encode() if body is not None else None
        req = urllib.request.Request(url, data=data, method=method)
        req.add_header("Content-Type", "application/json")
        req.add_header("Authorization", f"token {self.token}")
        try:
            with urllib.request.urlopen(req, timeout=30) as res:
                raw = res.read()
                return json.loads(raw) if raw else None
        except urllib.error.HTTPError as e:
            raise SystemExit(f"{method} {path} failed: {e.code} {e.read()[:200].decode('utf-8', 'replace')}")

    def get(self, path):
        return self._call("GET", path)

    def post(self, path, body):
        return self._call("POST", path, body)

    def put(self, path, body):
        return self._call("PUT", path, body)


def label_id(forge, repo, name):
    """The label's id, or None if this repo has no such label.

    Returning None rather than exiting matters when several repos are passed:
    aborting on the third would leave the first two already modified, which is
    a worse state than doing nothing. weblib-viewer has no labels at all.
    """
    for label in forge.get(f"repos/{repo}/labels?limit=100"):
        if label["name"] == name:
            return label["id"]
    return None


def open_items(forge, repo):
    """Open issues *and* pull requests. They share the /issues endpoint."""
    out = []
    page = 1
    while True:
        batch = forge.get(f"repos/{repo}/issues?state=open&limit=50&page={page}")
        if not batch:
            break
        out.extend(batch)
        if len(batch) < 50:
            break
        page += 1
    return out


def reconcile(forge, repo, label_name, dry_run):
    lid = label_id(forge, repo, label_name)
    if lid is None:
        print(f"  skipped: no {label_name!r} label in this repo", flush=True)
        return []
    changed = []
    for item in open_items(forge, repo):
        number = item["number"]
        deps = forge.get(f"repos/{repo}/issues/{number}/dependencies") or []
        if not deps:
            # No edges, no opinion. See the docstring.
            continue
        # Fail safe: only an *explicitly closed* dependency counts as done.
        # A cross-repo blocker is not fully visible to the Actions token, which
        # is scoped to one repo -- weblib-archive#30 is blocked by
        # weblib/cfbypass#1, and the runner read that entry as not-open and
        # took the label off while cfbypass#1 was still open. Anything whose
        # state we cannot positively confirm keeps the issue blocked.
        blockers = [d for d in deps if d.get("state") != "closed"]
        labelled = any(l["name"] == label_name for l in item.get("labels") or [])
        if blockers and not labelled:
            action, detail = "add", f"blocked by {', '.join('#%d' % d['number'] for d in blockers)}"
            if not dry_run:
                forge.post(f"repos/{repo}/issues/{number}/labels", {"labels": [lid]})
        elif not blockers and labelled:
            action, detail = "remove", "every blocker is closed"
            if not dry_run:
                # PUT the whole set minus this label, rather than
                # DELETE .../labels/{id}. Under Gitea Actions the automatic
                # token can *add* an issue label but not delete one -- the
                # DELETE fails whatever `permissions:` the workflow declares,
                # while the replace succeeds. Measured on 1.25.2; a run by
                # hand with a personal token works either way, so this only
                # ever showed up in CI.
                keep = [l["id"] for l in item.get("labels") or []
                        if l["name"] != label_name]
                forge.put(f"repos/{repo}/issues/{number}/labels", {"labels": keep})
        else:
            continue
        changed.append((number, action, detail))
        print(f"  {'would ' if dry_run else ''}{action:<6} {label_name} on {repo}#{number}"
              f"  ({detail})", flush=True)
    return changed


def main(argv=None):
    ap = argparse.ArgumentParser(description=__doc__,
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--repo", action="append", default=[],
                    help="owner/name; repeatable. Defaults to $GITHUB_REPOSITORY.")
    ap.add_argument("--host", default=os.environ.get("GITEA_HOST") or DEFAULT_HOST)
    ap.add_argument("--token", default=os.environ.get("GITEA_TOKEN")
                    or os.environ.get("GITHUB_TOKEN"))
    ap.add_argument("--label", default=DEFAULT_LABEL)
    ap.add_argument("--dry-run", action="store_true")
    args = ap.parse_args(argv)

    repos = args.repo or ([os.environ["GITHUB_REPOSITORY"]]
                          if os.environ.get("GITHUB_REPOSITORY") else [])
    if not repos:
        ap.error("no --repo given and GITHUB_REPOSITORY is not set")
    if not args.token:
        ap.error("no token: pass --token or set GITEA_TOKEN/GITHUB_TOKEN")

    forge = Forge(args.host, args.token)
    total = 0
    for repo in repos:
        print(f"{repo}:", flush=True)
        total += len(reconcile(forge, repo, args.label, args.dry_run))
    print(f"--> {total} change(s){' (dry run, nothing applied)' if args.dry_run else ''}",
          flush=True)
    return 0


if __name__ == "__main__":
    sys.exit(main())