dotfiles/python/gitea_flake_lock.py

"""Automation helpers for flake.lock pull requests on Gitea."""

from __future__ import annotations

import subprocess
from os import getenv
from typing import Annotated

import typer

from python.gitea import GiteaClient, PullRequest, split_repo_name

DEFAULT_BASE_BRANCH = "main"
DEFAULT_BRANCH = "automation/update-flake-lock"
DEFAULT_GITEA_URL = "https://gitea.tmmworkshop.com"
PR_LABELS = ["dependencies", "automated", "flake_lock_update"]
PR_CHECK_WORKFLOWS = ["build_systems.yml", "treefmt.yml", "pytest.yml"]
PR_TITLE = "Update flake.lock"
PR_BODY = "Automated flake.lock update."

app = typer.Typer(add_completion=False)


def run_cmd(cmd: list[str], *, check: bool = True) -> subprocess.CompletedProcess[str]:
    """Run a subprocess command."""
    return subprocess.run(cmd, capture_output=True, text=True, check=check)


def ensure_flake_lock_pull_request(
    client: GiteaClient,
    *,
    owner: str,
    repo: str,
    branch: str,
    base: str,
) -> PullRequest:
    """Return an existing flake.lock PR for the branch or create one."""
    pull_requests = client.list_open_pull_requests(owner=owner, repo=repo, head=branch)
    if pull_requests:
        return pull_requests[0]

    return client.create_pull_request(
        owner=owner,
        repo=repo,
        title=PR_TITLE,
        body=PR_BODY,
        head=branch,
        base=base,
        labels=PR_LABELS,
    )


def find_flake_lock_pull_request(client: GiteaClient, *, owner: str, repo: str) -> PullRequest | None:
    """Find the first open flake.lock pull request."""
    pull_requests = client.list_open_pull_requests(owner=owner, repo=repo, labels=["flake_lock_update"])
    if not pull_requests:
        return None
    return pull_requests[0]


def dispatch_pull_request_checks(client: GiteaClient, *, owner: str, repo: str, branch: str) -> None:
    """Dispatch the workflows that normally run for pull requests."""
    for workflow in PR_CHECK_WORKFLOWS:
        client.dispatch_workflow(owner=owner, repo=repo, workflow_id=workflow, ref=branch)


def has_worktree_changes() -> bool:
    """Return whether `flake.lock` has worktree changes."""
    result = run_cmd(["git", "diff", "--quiet", "--", "flake.lock"], check=False)
    return result.returncode != 0


def commit_flake_lock_update(*, branch: str) -> None:
    """Commit the updated lock file to the automation branch."""
    run_cmd(["git", "config", "user.name", "gitea-actions[bot]"])
    run_cmd(["git", "config", "user.email", "gitea-actions@tmmworkshop.com"])
    run_cmd(["git", "checkout", "-B", branch])
    run_cmd(["git", "add", "flake.lock"])
    run_cmd(["git", "commit", "-m", "chore: update flake.lock"])


def push_branch(*, branch: str) -> None:
    """Push the automation branch to origin."""
    run_cmd(["git", "push", "origin", f"HEAD:{branch}", "--force"])


def _required_gitea_token() -> str:
    """Read the required Gitea token from the environment."""
    token = getenv("GITEA_TOKEN")
    if token:
        return token

    msg = "GITEA_TOKEN environment variable is required"
    raise RuntimeError(msg)


@app.command()
def update(
    repo: Annotated[str, typer.Option("--repo", help="Gitea repository in owner/repo form")],
    base: Annotated[str, typer.Option("--base", help="Base branch")] = DEFAULT_BASE_BRANCH,
    branch: Annotated[str, typer.Option("--branch", help="Automation branch")] = DEFAULT_BRANCH,
) -> None:
    """Commit flake.lock changes and ensure a pull request exists."""
    if not has_worktree_changes():
        typer.echo("No flake.lock changes detected")
        return

    commit_flake_lock_update(branch=branch)
    push_branch(branch=branch)

    owner, repo_name = split_repo_name(repo)
    with GiteaClient(
        base_url=getenv("GITEA_URL", DEFAULT_GITEA_URL),
        token=_required_gitea_token(),
    ) as client:
        pull_request = ensure_flake_lock_pull_request(
            client,
            owner=owner,
            repo=repo_name,
            branch=branch,
            base=base,
        )
        # We can remove this if Gitea fixes the following issue:
        # https://github.com/go-gitea/gitea/issues/33963
        dispatch_pull_request_checks(client, owner=owner, repo=repo_name, branch=branch)
    typer.echo(pull_request.html_url or f"Pull request #{pull_request.number}")


@app.command()
def merge(
    repo: Annotated[str, typer.Option("--repo", help="Gitea repository in owner/repo form")],
) -> None:
    """Merge the first open flake.lock pull request."""
    owner, repo_name = split_repo_name(repo)
    with GiteaClient(
        base_url=getenv("GITEA_URL", DEFAULT_GITEA_URL),
        token=_required_gitea_token(),
    ) as client:
        pull_request = find_flake_lock_pull_request(client, owner=owner, repo=repo_name)
        if not pull_request:
            typer.echo("No open PR found with label flake_lock_update")
            return
        client.merge_pull_request(owner=owner, repo=repo_name, number=pull_request.number, merge_method="rebase")
    typer.echo(f"Merged PR #{pull_request.number}")


if __name__ == "__main__":
    app()