GRM-33: feat: add mandatory PR review step to workflow
This commit is contained in:
+60
-8
@@ -1,14 +1,23 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Auto-merge PR by extracting task ID from branch and validating PR title.
|
||||
"""Auto-merge PR by extracting task ID from branch and constructing merge title.
|
||||
|
||||
Waits for CI checks to complete before attempting the merge.
|
||||
|
||||
PR title format: ``GRM-N: <vikunja task title>``
|
||||
Merge commit format: ``GRM-N <conventional commit message>``
|
||||
|
||||
The conventional commit message is taken from the first commit on the PR
|
||||
branch (the branch HEAD). This allows the PR title to be a human-friendly
|
||||
Vikunja task title while the squashed commit follows conventional commits.
|
||||
|
||||
Usage:
|
||||
REPO_TOKEN=<token> python3 scripts/auto_merge.py <branch> <pr_title> <repo> <pr_number> [label_name]
|
||||
"""
|
||||
|
||||
import os
|
||||
import re
|
||||
import time
|
||||
from typing import Any
|
||||
|
||||
import click
|
||||
from dotenv import load_dotenv # pyright: ignore[reportMissingImports,reportUnknownVariableType]
|
||||
@@ -22,6 +31,9 @@ READY_TO_MERGE = "ready-to-merge"
|
||||
MAX_WAIT_SECONDS = 900 # 15 minutes
|
||||
POLL_INTERVAL_SECONDS = 30
|
||||
|
||||
# PR title: GRM-N: <vikunja task title>
|
||||
PR_TITLE_RE = re.compile(r"^GRM-\d+:\s+.+")
|
||||
|
||||
load_dotenv(override=True)
|
||||
|
||||
|
||||
@@ -31,17 +43,50 @@ def extract_task_id(branch: str) -> str:
|
||||
return match.group(0) if match else ""
|
||||
|
||||
|
||||
def validate_pr_title(pr_title: str) -> None:
|
||||
"""Raise ClickException if PR title does not follow conventional commits."""
|
||||
if not CONVENTIONAL_RE.match(pr_title):
|
||||
def validate_pr_title(pr_title: str, task_id: str) -> None:
|
||||
"""Raise ClickException if PR title does not follow the required format.
|
||||
|
||||
Expected: ``GRM-N: <vikunja task title>``
|
||||
"""
|
||||
if not PR_TITLE_RE.match(pr_title):
|
||||
raise click.ClickException(
|
||||
_(
|
||||
"Oops! PR title must follow conventional commit format.\n"
|
||||
" Expected: <type>: <description>\n"
|
||||
"Oops! PR title must follow format 'GRM-N: <task title>'.\n"
|
||||
" Expected: {task_id}: <task title>\n"
|
||||
" Got: {pr_title}",
|
||||
task_id=task_id,
|
||||
pr_title=pr_title,
|
||||
)
|
||||
)
|
||||
if not pr_title.startswith(f"{task_id}:"):
|
||||
raise click.ClickException(
|
||||
_(
|
||||
"Oops! PR title task ID mismatch.\n"
|
||||
" Branch task ID: {task_id}\n"
|
||||
" PR title: {pr_title}",
|
||||
task_id=task_id,
|
||||
pr_title=pr_title,
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def extract_conventional_msg(commits: list[dict[str, Any]]) -> str:
|
||||
"""Extract the conventional commit message from PR commits.
|
||||
|
||||
Iterates commits in reverse order (newest first) to find the first
|
||||
message matching the conventional commit format. Falls back to the
|
||||
newest commit message if none match.
|
||||
"""
|
||||
for commit in reversed(commits):
|
||||
commit_info = commit.get("commit", {})
|
||||
message = str(commit_info.get("message", "") if isinstance(commit_info, dict) else "").split("\n")[0]
|
||||
if CONVENTIONAL_RE.match(message):
|
||||
return message
|
||||
# Fallback: use the newest commit's first line
|
||||
if commits:
|
||||
commit_info = commits[-1].get("commit", {})
|
||||
return str(commit_info.get("message", "") if isinstance(commit_info, dict) else "").split("\n")[0]
|
||||
return ""
|
||||
|
||||
|
||||
def has_ready_to_merge_label(client: GiteaClient, pr_number: str) -> bool:
|
||||
@@ -138,7 +183,7 @@ def main(branch: str, pr_title: str, repo: str, pr_number: str, label_name: str)
|
||||
)
|
||||
)
|
||||
|
||||
validate_pr_title(pr_title)
|
||||
validate_pr_title(pr_title, task_id)
|
||||
|
||||
# Wait for CI checks to complete before attempting merge.
|
||||
pr = client.get_pr(pr_number)
|
||||
@@ -152,7 +197,14 @@ def main(branch: str, pr_title: str, repo: str, pr_number: str, label_name: str)
|
||||
else:
|
||||
click.echo(_("Warning: could not determine PR head SHA, proceeding without CI wait."))
|
||||
|
||||
merge_title = f"{task_id}: {pr_title}"
|
||||
# Build merge title: GRM-N <conventional commit message>
|
||||
commits = client.get_pr_commits(pr_number)
|
||||
conv_msg = extract_conventional_msg(commits)
|
||||
if not conv_msg:
|
||||
raise click.ClickException(
|
||||
_("Could not extract conventional commit message from PR commits.")
|
||||
)
|
||||
merge_title = f"{task_id} {conv_msg}"
|
||||
|
||||
try:
|
||||
client.merge_pr(pr_number, merge_title)
|
||||
|
||||
@@ -27,9 +27,14 @@ def extract_task_id(commit_msg: str) -> str:
|
||||
|
||||
|
||||
def extract_conventional_msg(commit_msg: str) -> str:
|
||||
"""Strip the GRM-N prefix from the commit subject."""
|
||||
"""Strip the GRM-N prefix from the commit subject.
|
||||
|
||||
Handles both formats:
|
||||
- ``GRM-N: <message>`` (legacy, colon-separated)
|
||||
- ``GRM-N <message>`` (current, space-separated)
|
||||
"""
|
||||
first_line = commit_msg.split("\n")[0]
|
||||
return re.sub(r"^GRM-\d+:\s*", "", first_line)
|
||||
return re.sub(r"^GRM-\d+[:\s]\s*", "", first_line)
|
||||
|
||||
|
||||
def resolve_task_id(client: VikunjaClient, task_id: str) -> int:
|
||||
|
||||
@@ -0,0 +1,140 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Post a review on a Gitea pull request.
|
||||
|
||||
Used by the GRM workflow to post structured PR reviews. The review body
|
||||
is provided via --body and inline comments via a JSON file
|
||||
(--comments-json) or stdin (--comments-stdin). This script is a thin
|
||||
CLI wrapper around ``GiteaClient.create_review`` — the actual review
|
||||
analysis is performed by the agent before invoking this tool.
|
||||
|
||||
Usage:
|
||||
REPO_TOKEN=<token> python3 scripts/review_pr.py <pr_number> <repo> \
|
||||
--event COMMENT \
|
||||
--body "Review body text" \
|
||||
--comments-json comments.json
|
||||
|
||||
The comments JSON file is a list of objects with keys:
|
||||
- path: file path in the repo
|
||||
- body: comment text
|
||||
- new_position: line number in the new file (1-based)
|
||||
- old_position: (optional) line number in the old file
|
||||
|
||||
Review focus areas (for the reviewer, not enforced by this script):
|
||||
- Functional completeness
|
||||
- Edge cases
|
||||
- Technical excellence: architecture compliance, SRP, deduplication,
|
||||
code smells, best practices, code quality, reusability, clean code,
|
||||
readability, maintainability, extensibility
|
||||
- Performance
|
||||
- Security
|
||||
- User experience
|
||||
- Documentation completeness and relevance
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from typing import Any
|
||||
|
||||
import click
|
||||
from dotenv import load_dotenv # pyright: ignore[reportMissingImports,reportUnknownVariableType]
|
||||
|
||||
from gitea_runner_manager.api_clients import GiteaClient
|
||||
from gitea_runner_manager.config import GITEA_API_URL
|
||||
from gitea_runner_manager.exceptions import APIError
|
||||
from gitea_runner_manager.i18n import _
|
||||
|
||||
load_dotenv(override=True)
|
||||
|
||||
VALID_EVENTS = ("APPROVE", "REQUEST_CHANGES", "COMMENT")
|
||||
|
||||
|
||||
def parse_comments(comments_json: str | None, comments_stdin: bool) -> list[dict[str, Any]]:
|
||||
"""Parse inline comments from a JSON file or stdin."""
|
||||
if comments_json:
|
||||
try:
|
||||
with open(comments_json) as f:
|
||||
data = json.load(f)
|
||||
except json.JSONDecodeError as e:
|
||||
raise click.ClickException(_("Invalid JSON in comments file: {error}", error=str(e))) from None
|
||||
if not isinstance(data, list):
|
||||
raise click.ClickException(_("Comments JSON must be a list of objects."))
|
||||
return data
|
||||
if comments_stdin:
|
||||
raw = sys.stdin.read().strip()
|
||||
if not raw:
|
||||
return []
|
||||
try:
|
||||
data = json.loads(raw)
|
||||
except json.JSONDecodeError as e:
|
||||
raise click.ClickException(_("Invalid JSON on stdin: {error}", error=str(e))) from None
|
||||
if not isinstance(data, list):
|
||||
raise click.ClickException(_("Stdin comments JSON must be a list of objects."))
|
||||
return data
|
||||
return []
|
||||
|
||||
|
||||
@click.command()
|
||||
@click.argument("pr_number")
|
||||
@click.argument("repo")
|
||||
@click.option(
|
||||
"--event",
|
||||
default="COMMENT",
|
||||
type=click.Choice(VALID_EVENTS),
|
||||
help="Review event type: APPROVE, REQUEST_CHANGES, or COMMENT.",
|
||||
)
|
||||
@click.option("--body", default="", help="Top-level review body text.")
|
||||
@click.option("--comments-json", default=None, help="Path to JSON file with inline comments.")
|
||||
@click.option(
|
||||
"--comments-stdin",
|
||||
is_flag=True,
|
||||
default=False,
|
||||
help="Read inline comments JSON from stdin.",
|
||||
)
|
||||
def main(
|
||||
pr_number: str,
|
||||
repo: str,
|
||||
event: str,
|
||||
body: str,
|
||||
comments_json: str | None,
|
||||
comments_stdin: bool,
|
||||
) -> None:
|
||||
token = os.environ.get("REPO_TOKEN", "")
|
||||
if not token:
|
||||
raise click.ClickException(_("ERROR: REPO_TOKEN is not set."))
|
||||
|
||||
owner, repo_name = repo.split("/")
|
||||
client = GiteaClient(GITEA_API_URL, token, owner, repo_name)
|
||||
|
||||
comments = parse_comments(comments_json, comments_stdin)
|
||||
|
||||
if event != "APPROVE" and not body and not comments:
|
||||
raise click.ClickException(_("Review body or inline comments are required for event '{event}'.", event=event))
|
||||
|
||||
try:
|
||||
review = client.create_review(pr_number, event=event, body=body, comments=comments)
|
||||
except APIError as e:
|
||||
raise click.ClickException(
|
||||
_(
|
||||
"Failed to post review: HTTP {status} — {message}",
|
||||
status=e.status,
|
||||
message=e.message,
|
||||
)
|
||||
) from None
|
||||
|
||||
review_id = review.get("id", "?")
|
||||
click.echo(
|
||||
_(
|
||||
"Review #{review_id} posted on PR #{pr_number} with event '{event}' ({num_comments} inline comments).",
|
||||
review_id=review_id,
|
||||
pr_number=pr_number,
|
||||
event=event,
|
||||
num_comments=len(comments),
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
main()
|
||||
Reference in New Issue
Block a user