139 lines
4.4 KiB
Python
139 lines
4.4 KiB
Python
"""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_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 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,
|
|
)
|
|
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()
|