GRM-57: Fully automate PR merge workflow #83

Merged
emil merged 1 commits from GRM-60-automate-workflow into master 2026-06-22 10:45:27 +00:00
16 changed files with 335 additions and 1541 deletions
-34
View File
@@ -1,34 +0,0 @@
name: Auto-merge
on:
pull_request:
types: [labeled, unlabeled]
jobs:
merge:
# Always run — the Python script checks for the label via API.
# Gitea's `labeled` event payload may not populate pull_request.labels
# correctly, so we can't rely on the YAML-level condition.
if: github.event.label.name == 'ready-to-merge'
runs-on: docker
timeout-minutes: 15
steps:
- uses: actions/checkout@v4
- name: Install dependencies
run: python3 -m pip install --break-system-packages requests python-dotenv click
- name: Squash merge with task ID
env:
REPO_TOKEN: ${{ secrets.REPO_TOKEN }}
VIKUNJA_TOKEN: ${{ secrets.VIKUNJA_TOKEN }}
PYTHONPATH: src
HEAD_REF: ${{ github.head_ref }}
PR_TITLE: ${{ github.event.pull_request.title }}
REPOSITORY: ${{ github.repository }}
PR_NUMBER: ${{ github.event.number }}
run: |
python3 scripts/ci/auto_merge.py \
"$HEAD_REF" \
"$PR_TITLE" \
"$REPOSITORY" \
"$PR_NUMBER" \
"ready-to-merge"
+31
View File
@@ -172,3 +172,34 @@ jobs:
python3 scripts/ci/pr_review.py \
"${{ github.event.number }}" \
"${{ github.repository }}"
auto-merge:
# Auto-merge runs after all CI checks pass. It reads the task ID
# from .taskid file, validates the PR title, and squash-merges.
# No manual label or review needed — CI is the quality gate.
needs: [quality, detect-changes, pr-review]
if: github.event_name == 'pull_request'
runs-on: docker
timeout-minutes: 10
steps:
- uses: actions/checkout@v4
with:
fetch-depth: 0
token: ${{ secrets.REPO_TOKEN }}
- name: Install dependencies
run: python3 -m pip install --break-system-packages requests python-dotenv click
- name: Squash merge with task ID
env:
REPO_TOKEN: ${{ secrets.REPO_TOKEN }}
VIKUNJA_TOKEN: ${{ secrets.VIKUNJA_TOKEN }}
PYTHONPATH: src
HEAD_REF: ${{ github.head_ref }}
PR_TITLE: ${{ github.event.pull_request.title }}
REPOSITORY: ${{ github.repository }}
PR_NUMBER: ${{ github.event.number }}
run: |
python3 scripts/ci/auto_merge.py \
"$HEAD_REF" \
"$PR_TITLE" \
"$REPOSITORY" \
"$PR_NUMBER"
+1
View File
@@ -0,0 +1 @@
GRM-57
-130
View File
@@ -1,130 +0,0 @@
# Review Checklist
This checklist is **mandatory** for every PR. The automated `pr-review` CI
job checks items marked **[auto]**. The agent must verify all items
marked **[manual]** before posting an APPROVE review.
The `review_pr.py` script requires `--checklist-confirmed` for APPROVE
events. This flag attests that every category below has been reviewed.
---
## 1. Architecture Compliance [auto + manual]
- [ ] **No business logic in CLI** (`cli.py`): no `subprocess`, no
`os.system`, no `ansible-playbook` — delegate to `executor.py`
- [ ] **No hardcoded URLs or config values** that belong in `config.py`
with env var overrides
- [ ] **Layer boundaries respected**: CLI → runner_manager → executor →
subprocess/Ansible. No skipping layers.
- [ ] **Single Responsibility**: each module/function has one reason to
change. If a function does two things, split it.
- [ ] **No circular imports** introduced
## 2. Code Quality and Best Practices [auto + manual]
- [ ] **No `print()`** in `src/` — use `click.echo()` for user output
- [ ] **No bare `except:`** — catch specific exceptions
- [ ] **No broad `except Exception:`** without justification
- [ ] **No `TODO`/`FIXME`/`HACK`/`XXX`** left in merged code
- [ ] **No functions > 50 lines** (excluding docstrings and decorators)
- [ ] **No dead code** — unused imports, unreachable branches, commented-out code
- [ ] **No copy-paste duplication** — extract shared logic into a helper
- [ ] **Idiomatic Python** — use comprehensions, context managers, dataclasses
- [ ] **Type hints** on all public functions
- [ ] **No `Any` type without justification** — document why if used
- [ ] **Error handling complete** — all failure paths handled, no silent failures
- [ ] **Cleanup in error paths** — files closed, connections released, temp files removed
## 3. Security [auto + manual]
- [ ] **No hardcoded secrets** (tokens, passwords, keys in string literals)
- [ ] **No `shell=True`** in subprocess calls — use argument lists
- [ ] **No `eval()` or `exec()`** — use `ast.literal_eval` if parsing literals
- [ ] **No secrets in logs or process arguments** — pass via env vars or files
- [ ] **Input validation** on all external inputs (CLI args, API responses, file contents)
- [ ] **No injection vectors** — parameterize subprocess args, SQL queries, etc.
- [ ] **File paths validated** — no path traversal (use `Path.resolve()`, check boundaries)
## 4. Internationalization (i18n) [auto + manual]
- [ ] **All user-facing strings wrapped in `_()`**`click.echo(_("..."))`,
error messages, help text, prompts
- [ ] **No raw English strings** in `click.echo()`, `click.ClickException()`,
or `raise` messages visible to users
- [ ] **String interpolation uses named placeholders**: `_("Hello {name}", name=x)`
not `f"Hello {x}"` for translatable strings
## 5. Testability and Test Coverage [auto + manual]
- [ ] **Source file changes include corresponding test updates**
- [ ] **100% coverage maintained** (enforced by `pytest-cov`)
- [ ] **Tests are fast** (< 10 seconds total, enforced by `check_test_speed.py`)
- [ ] **Edge cases tested**: empty inputs, boundary values, error paths, None/Optional
- [ ] **No flaky tests** — no `sleep()`, no race conditions, no external dependencies
- [ ] **Test names describe the scenario**: `test_<condition>_<expected_result>`
## 6. Performance [manual]
- [ ] **No unnecessary allocations** in hot paths — use generators for large datasets,
avoid reading entire files into memory
- [ ] **Correct data structures** — O(1) lookups use `set`/`dict`, not `list`;
`dict` for key-value, `set` for membership, `list` for ordered iteration
- [ ] **No N+1 query patterns** in API calls or file I/O — batch operations where possible
- [ ] **No blocking I/O on hot paths** without justification — CLI startup, command execution
## 7. User Experience [manual]
- [ ] **Clear error messages** — tell the user what went wrong and how to fix it.
Example: "Error: Config file not found at /etc/grm.conf. Create it with: grm config init"
- [ ] **Consistent CLI flag naming**`--long-name` with `--short` aliases
- [ ] **Help text on all commands and options**`--help` should be useful
- [ ] **No silent failures** — if something fails, the user should know
- [ ] **Output is actionable** — not just "Error" but "Error: X failed because Y. Try Z."
## 8. Documentation [auto + manual]
- [ ] **Source changes include doc updates** — README, wiki, AGENTS.md as needed
- [ ] **New functions/classes have docstrings** — Google style
- [ ] **Public API changes documented** in CHANGELOG (auto-generated by git-cliff)
- [ ] **AGENTS.md updated** if workflow, conventions, or processes changed
- [ ] **No stale documentation** — if code changed, docs must reflect it
## 9. Workflow Compliance [manual]
- [ ] **PR title matches Vikunja task title** (`GRM-N: <task title>`)
- [ ] **Commit messages follow conventional format** (`type: description`)
- [ ] **No force-push after review** — creates new commits and re-trigger CI
- [ ] **Branch is up to date** with master before merging
- [ ] **No merge commits** in the PR branch — use squash merge via auto-merge
## 10. Extensibility and Maintainability [manual]
- [ ] **Open/Closed Principle** — code is open for extension, closed for modification.
New behavior via new functions/classes, not by modifying existing ones
- [ ] **No magic numbers** — constants are named and documented
- [ ] **Configuration over hardcoding** — use `config.py` with env var overrides
- [ ] **Future-proof error handling** — don't catch specific error messages that may change
- [ ] **Dependencies are justified** — no new dependency without rationale
## 11. Resource Management [auto + manual]
- [ ] **File handles closed** — use `with` statements or explicit `close()` in `finally`
- [ ] **Subprocess resources cleaned up** — call `.wait()` or `.communicate()`
- [ ] **Temporary files deleted** — use `tempfile.TemporaryDirectory()` or cleanup in `finally`
- [ ] **No resource leaks in error paths**`try/finally` or context managers for cleanup
## 12. Backwards Compatibility [manual]
- [ ] **No breaking changes to public API** — or documented as major version bump
- [ ] **Removed functions deprecated first** — with `DeprecationWarning` and removal timeline
- [ ] **Default values added** instead of new required arguments
- [ ] **Return types stable** — no changes without major version bump
- [ ] **Behavior changes documented** — no silent behavior changes in existing functions
## 13. Logging and Observability [manual]
- [ ] **No sensitive data in logs** — tokens, passwords, PII excluded
- [ ] **Sufficient detail for debugging** — context, state, values logged at DEBUG level
- [ ] **Log levels appropriate** — DEBUG for internals, INFO for user actions, WARNING for recoverable issues
- [ ] **No log spam** — loops don't log per iteration, use DEBUG for high-frequency events
+37 -133
View File
@@ -1,23 +1,26 @@
#!/usr/bin/env python3
"""Auto-merge PR by extracting task ID from branch and constructing merge title.
"""Auto-merge PR when all CI checks pass.
Waits for CI checks to complete before attempting the merge.
Runs as the final job in ci.yml. Reads the task ID from ``.taskid`` file
(falling back to branch name extraction for backwards compatibility),
validates the PR title, and squash-merges with a conventional commit
message prefixed by the task ID.
PR title format: ``GRM-N: <vikunja task title>``
Merge commit format: ``GRM-N <conventional commit message>``
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.
The conventional commit message is extracted from the PR commits.
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]
REPO_TOKEN=<token> python3 scripts/ci/auto_merge.py <branch> <pr_title> <repo> <pr_number>
"""
import os
import re
import subprocess # nosec B404
import time
from pathlib import Path
from typing import Any
import click
@@ -35,9 +38,10 @@ from gitea_runner_manager.config import (
from gitea_runner_manager.exceptions import APIError
from gitea_runner_manager.i18n import _
READY_TO_MERGE = "ready-to-merge"
MAX_WAIT_SECONDS = 600 # 10 minutes max — CI may still be running when label is added
POLL_INTERVAL_SECONDS = 15 # Poll every 15 seconds
TASKID_FILE = ".taskid"
PR_TITLE_RE = re.compile(r"^GRM-\d+:\s+.+")
load_dotenv(override=True)
def run_cmd(args: list[str], check: bool = True) -> subprocess.CompletedProcess[str]:
@@ -54,14 +58,25 @@ def run_cmd(args: list[str], check: bool = True) -> subprocess.CompletedProcess[
return result
# PR title: GRM-N: <vikunja task title>
PR_TITLE_RE = re.compile(r"^GRM-\d+:\s+.+")
def read_taskid(branch: str) -> str:
"""Read task ID from .taskid file, falling back to branch name extraction.
load_dotenv(override=True)
The .taskid file is a simple text file containing just the task ID
(e.g., ``GRM-60``). If the file doesn't exist, extract from the
branch name as a backwards-compatibility fallback.
"""
path = Path(TASKID_FILE)
if path.exists():
task_id = path.read_text(encoding="utf-8").strip()
if task_id:
return task_id
# Fallback: extract from branch name
match = TASK_ID_RE.search(branch)
return match.group(0) if match else ""
def extract_task_id(branch: str) -> str:
"""Extract GRM-N task identifier from branch name."""
"""Extract GRM-N task identifier from branch name (legacy fallback)."""
match = TASK_ID_RE.search(branch)
return match.group(0) if match else ""
@@ -134,28 +149,6 @@ def validate_pr_title_matches_vikunja(pr_title: str, task_id: str) -> None:
)
def has_approval_review(client: GiteaClient, pr_number: str) -> bool:
"""Check whether the PR has at least one substantive APPROVE review.
A substantive review has a body longer than 20 characters (not just
"LGTM" or "OK"). This ensures the reviewer actually reviewed the PR
rather than rubber-stamping it.
Returns False if no APPROVE review is found — the caller should
block the merge in that case.
"""
reviews = client.get_pr_reviews(pr_number)
for r in reviews:
state = r.get("state", "")
if state == "APPROVED":
body = str(r.get("body", "")).strip()
if len(body) > 20 or r.get("comments", []):
return True
return False
def extract_conventional_msg(commits: list[dict[str, Any]]) -> str:
"""Extract the conventional commit message from PR commits.
@@ -175,73 +168,12 @@ def extract_conventional_msg(commits: list[dict[str, Any]]) -> str:
return ""
def has_ready_to_merge_label(client: GiteaClient, pr_number: str) -> bool:
"""Check whether the PR has the ready-to-merge label via the API."""
labels = client.get_pr_labels(pr_number)
return any(label.get("name") == READY_TO_MERGE for label in labels)
def wait_for_ci(
client: GiteaClient, sha: str, max_wait: int = MAX_WAIT_SECONDS, poll_interval: int = POLL_INTERVAL_SECONDS
) -> bool:
"""Poll commit statuses until all CI checks are complete (not pending).
Returns True if all checks are successful, False if any failed or timed out.
Uses the combined status endpoint which returns one entry per context
(deduplicated server-side). Filters to "CI /" contexts only, excluding
"Auto-merge / merge" and other non-CI contexts.
"""
elapsed = 0
while elapsed < max_wait:
statuses = client.get_commit_status(sha)
if not statuses:
click.echo(_("No CI checks reported yet, waiting..."))
time.sleep(poll_interval)
elapsed += poll_interval
continue
# Combined endpoint already deduplicates — one entry per context.
# Filter to CI contexts only (excludes "Auto-merge / merge" etc).
ci_statuses = {s.get("context", ""): s for s in statuses if s.get("context", "").startswith("CI /")}
if not ci_statuses:
click.echo(_("No CI checks found yet, waiting..."))
time.sleep(poll_interval)
elapsed += poll_interval
continue
pending = [ctx for ctx, s in ci_statuses.items() if s.get("status") in ("pending", "waiting")]
if not pending:
# All CI checks are complete — check if they all succeeded.
# "skipped" jobs are considered passing (conditional jobs that didn't run).
failed = [ctx for ctx, s in ci_statuses.items() if s.get("status") not in ("success", "ok", "skipped")]
if failed:
click.echo(_("CI checks failed: {failed}", failed=", ".join(sorted(failed))))
return False
click.echo(_("All CI checks passed."))
return True
click.echo(
_(
"Waiting for CI checks: {pending} ({elapsed}s elapsed)",
pending=", ".join(sorted(pending)),
elapsed=elapsed,
)
)
time.sleep(poll_interval)
elapsed += poll_interval
click.echo(_("Timed out waiting for CI checks after {max_wait}s.", max_wait=max_wait))
return False
@click.command()
@click.argument("branch")
@click.argument("pr_title")
@click.argument("repo")
@click.argument("pr_number")
@click.argument("label_name", required=False, default="")
def main(branch: str, pr_title: str, repo: str, pr_number: str, label_name: str) -> None:
def main(branch: str, pr_title: str, repo: str, pr_number: str) -> None:
token = os.environ.get("REPO_TOKEN", "")
if not token:
raise click.ClickException(_("ERROR: REPO_TOKEN is not set."))
@@ -249,47 +181,20 @@ def main(branch: str, pr_title: str, repo: str, pr_number: str, label_name: str)
owner, repo_name = repo.split("/")
client = GiteaClient(GITEA_API_URL, token, owner, repo_name)
# Gitea Actions may not populate github.event.label.name; fall back to API check.
if label_name != READY_TO_MERGE and not has_ready_to_merge_label(client, pr_number):
click.echo(_("Label '{label}' is not '{rtm}', skipping.", label=label_name, rtm=READY_TO_MERGE))
return
task_id = extract_task_id(branch)
task_id = read_taskid(branch)
if not task_id:
raise click.ClickException(
_(
"Oops! No task ID (GRM-N) found in branch name '{branch}'.",
"Oops! No task ID found in .taskid file or branch name '{branch}'.",
branch=branch,
)
)
click.echo(_("Task ID: {task_id}", task_id=task_id))
validate_pr_title(pr_title, task_id)
validate_pr_title_matches_vikunja(pr_title, task_id)
# Enforce APPROVE review before merge (Gap 2 fix)
if not has_approval_review(client, pr_number):
raise click.ClickException(
_(
"Cannot merge: PR #{pr_number} has no APPROVE review. "
"Please review and approve before adding the ready-to-merge label.",
pr_number=pr_number,
)
)
click.echo(_("PR has at least one APPROVE review."))
# Wait for CI checks to complete before attempting merge.
pr = client.get_pr(pr_number)
sha = pr.get("head", {}).get("sha", "")
if sha:
click.echo(_("Waiting for CI checks on commit {sha}...", sha=sha[:8]))
if not wait_for_ci(client, sha):
raise click.ClickException(
_("Cannot merge: CI checks did not pass. Please fix failing checks and re-label.")
)
else:
click.echo(_("Warning: could not determine PR head SHA, proceeding without CI wait."))
# Build merge title: GRM-N <conventional commit message>
# 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:
@@ -311,8 +216,7 @@ def main(branch: str, pr_title: str, repo: str, pr_number: str, label_name: str)
except (APIError, Exception) as retry_err:
raise click.ClickException(
_(
"Merge failed after rebase retry: {error}\n"
"Please rebase the PR manually and re-add the ready-to-merge label.",
"Merge failed after rebase retry: {error}\nPlease rebase the PR manually.",
error=str(retry_err),
)
) from None
-2
View File
@@ -24,7 +24,6 @@ Classification strategy (safe-by-default):
- tests/** — Test files
- hooks/** — Git hooks
- AGENTS.md — Agent conventions
- REVIEW_CHECKLIST.md — Review checklist
- README.md — README (lean, links to wiki)
- CHANGELOG.md — Changelog (generated)
- TROUBLESHOOTING.md — Troubleshooting guide
@@ -75,7 +74,6 @@ WORKFLOW_ONLY_PATTERNS = frozenset(
# Documentation
"docs/",
"AGENTS.md",
"REVIEW_CHECKLIST.md",
"README.md",
"CHANGELOG.md",
"TROUBLESHOOTING.md",
+1 -7
View File
@@ -500,13 +500,7 @@ def build_review_body(result: ReviewResult) -> str:
lines.append("")
lines.append("---")
lines.append("**Manual review required:** Before approving, review every category in")
lines.append("[REVIEW_CHECKLIST.md](REVIEW_CHECKLIST.md) and confirm with:")
lines.append("```bash")
lines.append("python3 scripts/ci/review_pr.py <PR> <owner/repo> \\")
lines.append(" --event APPROVE --checklist-confirmed \\")
lines.append(' --body "<substantive review summary>"')
lines.append("```")
lines.append("**Auto-merge:** If all CI checks pass, this PR will be merged automatically.")
return "\n".join(lines)
-184
View File
@@ -1,184 +0,0 @@
#!/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).
.. note::
The ``tea`` CLI v0.14.1 only supports interactive reviews (no
``--approve``/``--comment`` flags), so this script uses
``GiteaClient`` (direct HTTP API) for posting reviews. When a newer
version of tea adds non-interactive review support, this can be
switched to use ``TeaCLI.review_pr()``.
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
For APPROVE events, --checklist-confirmed is required. This attests
that the reviewer has gone through every category in
REVIEW_CHECKLIST.md. The review body must also be substantive
(> 20 characters) trivial "LGTM" approvals are rejected by the
auto-merge gate.
"""
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.",
)
@click.option(
"--checklist-confirmed",
is_flag=True,
default=False,
help="Required for APPROVE: confirms all REVIEW_CHECKLIST.md categories reviewed.",
)
@click.option(
"--checklist-categories",
default="",
help="Comma-separated list of checklist categories reviewed (e.g., '1,2,3,4,5,6,7,8,9,10,11,12,13'). "
"Required for APPROVE: must list at least 8 of 13 categories.",
)
def main(
pr_number: str,
repo: str,
event: str,
body: str,
comments_json: str | None,
comments_stdin: bool,
checklist_confirmed: bool,
checklist_categories: str,
) -> 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))
if event == "APPROVE":
if not checklist_confirmed:
raise click.ClickException(
_(
"APPROVE requires --checklist-confirmed. "
"Review every category in REVIEW_CHECKLIST.md before approving."
)
)
# Validate that at least 8 of 13 checklist categories were reviewed
categories = [c.strip() for c in checklist_categories.split(",") if c.strip()] if checklist_categories else []
if len(categories) < 8:
raise click.ClickException(
_(
"APPROVE requires --checklist-categories with at least 8 of 13 categories reviewed. "
"Provide comma-separated category numbers (e.g., '1,2,3,4,5,6,7,8'). "
"Got {count} categories: {cats}",
count=len(categories),
cats=checklist_categories or "(none)",
)
)
if len(body.strip()) <= 50 and not comments:
raise click.ClickException(
_(
"APPROVE review body must be substantive (> 50 characters) "
"or include inline comments. Trivial approvals are rejected. "
"Current body is {len} characters.",
len=len(body.strip()),
)
)
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()
+5 -53
View File
@@ -1,9 +1,9 @@
#!/usr/bin/env python3
"""Configure GRM repository: branch protection + labels via Gitea REST API.
"""Configure GRM repository: branch protection + repo settings via Gitea REST API.
Uses the ``tea`` Gitea CLI for label creation and the ``GiteaClient`` for
branch protection and repo settings (tea only supports basic protect/unprotect,
not the detailed config we need with status checks and required approvals).
Uses ``GiteaClient`` for branch protection and repo settings.
The ``tea`` CLI is used for label creation if available, with a
fallback to ``GiteaClient`` if tea is not installed.
Usage:
REPO_TOKEN=<token> python3 scripts/configure_repo.py
@@ -20,14 +20,12 @@ from gitea_runner_manager.api_clients import GiteaClient
from gitea_runner_manager.config import (
BRANCH_PROTECTION_CONFIG,
GITEA_API_URL,
LABEL_CONFIG,
REPO_NAME,
REPO_OWNER,
REPO_SETTINGS_CONFIG,
)
from gitea_runner_manager.exceptions import APIError
from gitea_runner_manager.i18n import _
from scripts.gitea_cli import TeaCLI, TeaCLIError
load_dotenv(override=True)
@@ -46,48 +44,17 @@ def _handle_http_error(e: APIError) -> None:
raise click.ClickException(_("HTTP error: {status}{message}", status=e.status, message=e.message))
def _ensure_label_via_tea(tea: TeaCLI, repo: str, name: str, color: str, description: str) -> bool:
"""Create a label via tea if it doesn't already exist.
Returns True if created, False if it already existed.
Falls back to GiteaClient if tea is not installed or fails.
"""
try:
existing = tea.list_labels(repo)
if any(label.get("name") == name for label in existing):
return False
tea.create_label(repo, name=name, color=color, description=description)
return True
except (TeaCLIError, FileNotFoundError):
# Fall back to GiteaClient if tea is not installed or fails
return _ensure_label_via_client(name, color, description)
def _ensure_label_via_client(name: str, color: str, description: str) -> bool:
"""Fallback: create label via GiteaClient. Returns True if created."""
client = GiteaClient(
GITEA_API_URL,
os.environ.get("REPO_TOKEN", ""),
REPO_OWNER,
REPO_NAME,
)
result = client.ensure_label(name=name, color=color, description=description)
return result is not None
def main() -> None:
token = os.environ.get("REPO_TOKEN", "")
if not token:
raise click.ClickException(_("ERROR: REPO_TOKEN is not set."))
repo = f"{REPO_OWNER}/{REPO_NAME}"
client = GiteaClient(GITEA_API_URL, token, REPO_OWNER, REPO_NAME)
tea = TeaCLI(repo=repo)
try:
click.echo(_("Configuring branch protection for {branch}...", branch="master"))
client.ensure_branch_protection("master", BRANCH_PROTECTION_CONFIG)
click.echo(_(" - Direct pushes: BLOCKED (require PR)"))
click.echo(_(" - Direct pushes: BLOCKED (require PR, whitelisted users can push)"))
click.echo(
_(
" - Required approvals: {count}",
@@ -100,21 +67,6 @@ def main() -> None:
checks = ", ".join(cast(list[str], BRANCH_PROTECTION_CONFIG["status_check_contexts"]))
click.echo(_(" - Required status checks: {checks}", checks=checks))
click.echo("")
label_name = cast(str, LABEL_CONFIG["name"])
click.echo(_("Creating {label} label...", label=label_name))
created = _ensure_label_via_tea(
tea,
repo,
name=label_name,
color=cast(str, LABEL_CONFIG["color"]),
description=cast(str, LABEL_CONFIG["description"]),
)
if created:
click.echo(_(" Label '{label}' created.", label=label_name))
else:
click.echo(_(" Label '{label}' already exists.", label=label_name))
click.echo("")
click.echo(_("Configuring repository settings..."))
client.update_repo_settings(cast(dict[str, object], REPO_SETTINGS_CONFIG))
-6
View File
@@ -38,12 +38,6 @@ BRANCH_PROTECTION_CONFIG: dict[str, object] = {
"block_on_official_review_requests": True,
}
LABEL_CONFIG: dict[str, object] = {
"name": "ready-to-merge",
"color": "2ecc71",
"description": "Auto-merge PR when all CI checks pass",
}
REPO_SETTINGS_CONFIG: dict[str, object] = {
"default_delete_branch_after_merge": True,
}
+227 -551
View File
@@ -1,669 +1,345 @@
"""Unit tests for scripts/ci/auto_merge.py."""
import http
from unittest.mock import MagicMock, patch
import click
import pytest
from click.testing import CliRunner
from gitea_runner_manager.config import CONVENTIONAL_RE, TASK_ID_RE
from gitea_runner_manager.exceptions import APIError
from scripts.ci.auto_merge import (
PR_TITLE_RE,
extract_conventional_msg,
extract_task_id,
has_approval_review,
has_ready_to_merge_label,
main,
read_taskid,
run_cmd,
validate_pr_title,
validate_pr_title_matches_vikunja,
wait_for_ci,
)
CI_SUCCESS = "success"
CI_PENDING = "pending"
CI_FAILURE = "failure"
# -- read_taskid --
def _status(context: str, status: str, updated_at: str = "2026-01-01T00:00:00Z") -> dict[str, str]:
return {"context": context, "status": status, "updated_at": updated_at}
class TestReadTaskid:
def test_reads_from_file(self, tmp_path, monkeypatch) -> None: # type: ignore[no-untyped-def]
monkeypatch.chdir(tmp_path)
(tmp_path / ".taskid").write_text("GRM-60\n")
assert read_taskid("some-branch") == "GRM-60"
def test_falls_back_to_branch_name(self, tmp_path, monkeypatch) -> None: # type: ignore[no-untyped-def]
monkeypatch.chdir(tmp_path)
assert read_taskid("GRM-19-fix-bug") == "GRM-19"
def test_returns_empty_when_no_file_no_match(self, tmp_path, monkeypatch) -> None: # type: ignore[no-untyped-def]
monkeypatch.chdir(tmp_path)
assert read_taskid("feature-branch") == ""
def test_empty_file_falls_back_to_branch(self, tmp_path, monkeypatch) -> None: # type: ignore[no-untyped-def]
monkeypatch.chdir(tmp_path)
(tmp_path / ".taskid").write_text("\n")
assert read_taskid("GRM-42-test") == "GRM-42"
def _commit(message: str) -> dict[str, dict[str, str]]:
return {"commit": {"message": message}}
class TestRegexes:
def test_task_id_re_matches(self) -> None:
assert TASK_ID_RE.search("GRM-19-fix-bug")
assert TASK_ID_RE.search("feature/GRM-42")
def test_task_id_re_no_match(self) -> None:
assert not TASK_ID_RE.search("feature-no-id")
def test_conventional_re_matches(self) -> None:
assert CONVENTIONAL_RE.match("feat: add feature")
assert CONVENTIONAL_RE.match("fix(api): handle timeout")
def test_conventional_re_rejects(self) -> None:
assert not CONVENTIONAL_RE.match("random message")
assert not CONVENTIONAL_RE.match("feat:")
def test_pr_title_re_matches(self) -> None:
assert PR_TITLE_RE.match("GRM-19: Some task title")
assert PR_TITLE_RE.match("GRM-42: double space title")
def test_pr_title_re_rejects(self) -> None:
assert not PR_TITLE_RE.match("fix: resolve timeout")
assert not PR_TITLE_RE.match("GRM-19:No space after colon")
assert not PR_TITLE_RE.match("random message")
# -- extract_task_id (legacy fallback) --
class TestExtractTaskId:
def test_extracts_from_branch(self) -> None:
assert extract_task_id("GRM-19-fix-bug") == "GRM-19"
assert extract_task_id("GRM-123") == "GRM-123"
def test_extracts_from_feature_branch(self) -> None:
assert extract_task_id("feature/GRM-42-add-x") == "GRM-42"
def test_returns_empty_when_no_match(self) -> None:
assert extract_task_id("feature-branch") == ""
def test_returns_empty_when_missing(self) -> None:
assert extract_task_id("feature-no-id") == ""
# -- validate_pr_title --
class TestValidatePrTitle:
def test_valid_title_passes(self) -> None:
validate_pr_title("GRM-19: Some task title", "GRM-19")
def test_valid_title(self) -> None:
validate_pr_title("GRM-19: Add new feature", "GRM-19")
def test_valid_title_with_scope_passes(self) -> None:
validate_pr_title("GRM-42: Add --url option", "GRM-42")
def test_missing_colon(self) -> None:
with pytest.raises(click.ClickException, match="format"):
validate_pr_title("GRM-19 Add new feature", "GRM-19")
def test_invalid_format_raises(self) -> None:
with pytest.raises(click.ClickException) as exc:
validate_pr_title("random message", "GRM-19")
assert "GRM-N" in str(exc.value)
def test_task_id_mismatch(self) -> None:
with pytest.raises(click.ClickException, match="mismatch"):
validate_pr_title("GRM-20: Add feature", "GRM-19")
def test_invalid_format_conventional_raises(self) -> None:
with pytest.raises(click.ClickException) as exc:
validate_pr_title("fix: resolve timeout", "GRM-19")
assert "GRM-N" in str(exc.value)
def test_task_id_mismatch_raises(self) -> None:
with pytest.raises(click.ClickException) as exc:
validate_pr_title("GRM-42: Some task title", "GRM-19")
assert "mismatch" in str(exc.value)
def test_no_task_id_in_title(self) -> None:
with pytest.raises(click.ClickException, match="format"):
validate_pr_title("Add new feature", "GRM-19")
class TestExtractConventionalMsg:
def test_returns_newest_conventional(self) -> None:
"""Iterates in reverse — picks the newest conventional commit."""
commits = [
_commit("fix: resolve timeout"),
_commit("random message"),
_commit("feat: add thing"),
]
assert extract_conventional_msg(commits) == "feat: add thing"
def test_falls_back_to_newest_commit(self) -> None:
commits = [
_commit("another random"),
_commit("random message"),
]
assert extract_conventional_msg(commits) == "random message"
def test_uses_first_line_only(self) -> None:
commits = [_commit("fix: resolve timeout\n\nBody text here")]
assert extract_conventional_msg(commits) == "fix: resolve timeout"
def test_empty_commits_returns_empty(self) -> None:
assert extract_conventional_msg([]) == ""
def test_commit_with_scope(self) -> None:
commits = [_commit("feat(api): new endpoint")]
assert extract_conventional_msg(commits) == "feat(api): new endpoint"
def test_missing_commit_key(self) -> None:
commits = [{}] # type: ignore[list-item]
assert extract_conventional_msg(commits) == ""
class TestHasReadyToMergeLabel:
def test_label_present(self) -> None:
client = MagicMock()
client.get_pr_labels.return_value = [{"name": "bug"}, {"name": "ready-to-merge"}]
assert has_ready_to_merge_label(client, "5") is True
def test_label_absent(self) -> None:
client = MagicMock()
client.get_pr_labels.return_value = [{"name": "bug"}]
assert has_ready_to_merge_label(client, "5") is False
def test_no_labels(self) -> None:
client = MagicMock()
client.get_pr_labels.return_value = []
assert has_ready_to_merge_label(client, "5") is False
class TestHasApprovalReview:
def test_has_substantive_approved(self) -> None:
client = MagicMock()
client.get_pr_reviews.return_value = [
{"state": "APPROVED", "body": "All comments addressed. LGTM.", "comments": []},
{"state": "COMMENT"},
]
assert has_approval_review(client, "5") is True
def test_has_approved_with_inline_comments(self) -> None:
client = MagicMock()
client.get_pr_reviews.return_value = [
{"state": "APPROVED", "body": "", "comments": [{"body": "good"}]},
]
assert has_approval_review(client, "5") is True
def test_trivial_approved_without_comments_returns_false(self) -> None:
"""A bare 'LGTM' approval (< 20 chars) without comments is not substantive."""
client = MagicMock()
client.get_pr_reviews.return_value = [
{"state": "APPROVED", "body": "LGTM", "comments": []},
]
assert has_approval_review(client, "5") is False
def test_no_approved_returns_false(self) -> None:
"""No APPROVE review means merge is blocked — no fallback."""
client = MagicMock()
client.get_pr_reviews.return_value = [{"state": "COMMENT"}]
assert has_approval_review(client, "5") is False
def test_changes_requested_blocks_merge(self) -> None:
"""REQUEST_CHANGES blocks merge."""
client = MagicMock()
client.get_pr_reviews.return_value = [{"state": "REQUEST_CHANGES", "body": "Fix this"}]
assert has_approval_review(client, "5") is False
def test_no_reviews_returns_false(self) -> None:
"""No reviews at all means no APPROVE — merge is blocked."""
client = MagicMock()
client.get_pr_reviews.return_value = []
assert has_approval_review(client, "5") is False
# -- validate_pr_title_matches_vikunja --
class TestValidatePrTitleMatchesVikunja:
@patch("scripts.ci.auto_merge.get_vikunja_task_title", return_value="")
@patch.dict("os.environ", {}, clear=True)
def test_no_vikunja_token_skips(self, mock_get: MagicMock) -> None:
"""Should skip validation when VIKUNJA_TOKEN is not set."""
validate_pr_title_matches_vikunja("GRM-19: Some title", "GRM-19")
def test_skips_when_no_token(self) -> None:
# Should not raise — just warn
validate_pr_title_matches_vikunja("GRM-19: test", "GRM-19")
@patch("scripts.ci.auto_merge.get_vikunja_task_title", return_value="Some task title")
@patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"}, clear=True)
def test_matching_title_passes(self, mock_get: MagicMock) -> None:
validate_pr_title_matches_vikunja("GRM-19: Some task title", "GRM-19")
@patch("scripts.ci.auto_merge.get_vikunja_task_title", return_value="Some task title")
@patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"}, clear=True)
def test_mismatched_title_raises(self, mock_get: MagicMock) -> None:
with pytest.raises(click.ClickException) as exc:
validate_pr_title_matches_vikunja("GRM-19: Different title", "GRM-19")
assert "does not match" in str(exc.value)
@patch("scripts.ci.auto_merge.get_vikunja_task_title", return_value="")
@patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"}, clear=True)
def test_task_not_found_skips(self, mock_get: MagicMock) -> None:
"""Should skip validation when Vikunja task is not found."""
validate_pr_title_matches_vikunja("GRM-19: Some title", "GRM-19")
class TestGetVikunjaTaskTitle:
@patch.dict("os.environ", {}, clear=True)
def test_no_token_returns_empty(self) -> None:
from scripts.ci.auto_merge import get_vikunja_task_title
assert get_vikunja_task_title("GRM-19") == ""
@patch("scripts.ci.auto_merge.VikunjaClient")
@patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"}, clear=True)
def test_finds_task(self, mock_client_cls: MagicMock) -> None:
from scripts.ci.auto_merge import get_vikunja_task_title
def test_matches(self, mock_client_cls: MagicMock) -> None:
mock_client = MagicMock()
mock_client.list_project_tasks.return_value = [
{"identifier": "GRM-19", "title": "Some task title"},
{"id": 1, "identifier": "GRM-19", "title": "Add new feature"},
]
mock_client_cls.return_value = mock_client
assert get_vikunja_task_title("GRM-19") == "Some task title"
validate_pr_title_matches_vikunja("GRM-19: Add new feature", "GRM-19")
@patch("scripts.ci.auto_merge.VikunjaClient")
@patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"}, clear=True)
@patch("scripts.ci.auto_merge.VikunjaClient")
def test_mismatch_raises(self, mock_client_cls: MagicMock) -> None:
mock_client = MagicMock()
mock_client.list_project_tasks.return_value = [
{"id": 1, "identifier": "GRM-19", "title": "Different title"},
]
mock_client_cls.return_value = mock_client
with pytest.raises(click.ClickException, match="does not match"):
validate_pr_title_matches_vikunja("GRM-19: Add new feature", "GRM-19")
@patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"}, clear=True)
@patch("scripts.ci.auto_merge.VikunjaClient")
def test_task_not_found_returns_empty(self, mock_client_cls: MagicMock) -> None:
from scripts.ci.auto_merge import get_vikunja_task_title
mock_client = MagicMock()
mock_client.list_project_tasks.return_value = [
{"identifier": "GRM-20", "title": "Other task"},
]
mock_client_cls.return_value = mock_client
assert get_vikunja_task_title("GRM-19") == ""
@patch("scripts.ci.auto_merge.VikunjaClient")
@patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"}, clear=True)
def test_paginates_to_find_task(self, mock_client_cls: MagicMock) -> None:
from scripts.ci.auto_merge import get_vikunja_task_title
mock_client = MagicMock()
# First page: full page of 50 tasks, no match; second page: match
page1 = [{"identifier": f"GRM-{i}", "title": f"task {i}"} for i in range(50)]
page2 = [{"identifier": "GRM-99", "title": "Found task"}]
mock_client.list_project_tasks.side_effect = [page1, page2]
mock_client_cls.return_value = mock_client
assert get_vikunja_task_title("GRM-99") == "Found task"
@patch("scripts.ci.auto_merge.VikunjaClient")
@patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"}, clear=True)
def test_empty_pages_returns_empty(self, mock_client_cls: MagicMock) -> None:
from scripts.ci.auto_merge import get_vikunja_task_title
"""When Vikunja task is not found, returns empty string (skip validation)."""
mock_client = MagicMock()
mock_client.list_project_tasks.return_value = []
mock_client_cls.return_value = mock_client
assert get_vikunja_task_title("GRM-19") == ""
# Should not raise — just warn
validate_pr_title_matches_vikunja("GRM-99: test", "GRM-99")
@patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"}, clear=True)
@patch("scripts.ci.auto_merge.VikunjaClient")
def test_task_found_on_second_page(self, mock_client_cls: MagicMock) -> None:
"""Pagination: task found on page 2."""
mock_client = MagicMock()
page1 = [{"id": i, "identifier": f"GRM-{i}", "title": f"Title {i}"} for i in range(50)]
page2 = [{"id": 100, "identifier": "GRM-99", "title": "Found me"}]
mock_client.list_project_tasks.side_effect = [page1, page2]
mock_client_cls.return_value = mock_client
# Should not raise — title matches
validate_pr_title_matches_vikunja("GRM-99: Found me", "GRM-99")
class TestWaitForCi:
def test_all_pass_immediately(self) -> None:
client = MagicMock()
client.get_commit_status.return_value = [
_status("CI / quality (pull_request)", CI_SUCCESS),
_status("CI / molecule-tests (0) (pull_request)", CI_SUCCESS),
@patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"}, clear=True)
@patch("scripts.ci.auto_merge.VikunjaClient")
def test_task_not_found_partial_page(self, mock_client_cls: MagicMock) -> None:
"""Pagination stops when page has fewer than DEFAULT_PER_PAGE results."""
mock_client = MagicMock()
mock_client.list_project_tasks.return_value = [
{"id": 1, "identifier": "GRM-1", "title": "Title 1"},
]
assert wait_for_ci(client, "abc123", max_wait=10) is True
mock_client_cls.return_value = mock_client
# Should not raise — returns empty, skips validation
validate_pr_title_matches_vikunja("GRM-99: test", "GRM-99")
def test_waits_then_passes(self) -> None:
client = MagicMock()
client.get_commit_status.side_effect = [
[_status("CI / quality (pull_request)", CI_PENDING)],
[_status("CI / quality (pull_request)", CI_SUCCESS)],
# -- extract_conventional_msg --
class TestExtractConventionalMsg:
def test_finds_conventional(self) -> None:
commits = [
{"commit": {"message": "fix: resolve timeout"}},
{"commit": {"message": "merge branch"}},
]
with patch("scripts.ci.auto_merge.time.sleep"):
assert wait_for_ci(client, "abc123", max_wait=10, poll_interval=5) is True
assert extract_conventional_msg(commits) == "fix: resolve timeout"
def test_fails_on_failed_check(self) -> None:
client = MagicMock()
client.get_commit_status.return_value = [
_status("CI / quality (pull_request)", CI_SUCCESS),
_status("CI / molecule-tests (0) (pull_request)", CI_FAILURE),
def test_finds_latest_conventional(self) -> None:
commits = [
{"commit": {"message": "merge branch"}},
{"commit": {"message": "feat: add feature"}},
]
assert wait_for_ci(client, "abc123", max_wait=10) is False
assert extract_conventional_msg(commits) == "feat: add feature"
def test_times_out(self) -> None:
client = MagicMock()
client.get_commit_status.return_value = [
_status("CI / quality (pull_request)", CI_PENDING),
def test_falls_back_to_newest(self) -> None:
commits = [
{"commit": {"message": "random message"}},
]
with patch("scripts.ci.auto_merge.time.sleep"):
assert wait_for_ci(client, "abc123", max_wait=5) is False
assert extract_conventional_msg(commits) == "random message"
def test_no_statuses_waits(self) -> None:
client = MagicMock()
client.get_commit_status.side_effect = [
[],
[_status("CI / quality (pull_request)", CI_SUCCESS)],
def test_empty_commits(self) -> None:
assert extract_conventional_msg([]) == ""
def test_multiline_message(self) -> None:
commits = [
{"commit": {"message": "feat: add feature\n\nBody text."}},
]
with patch("scripts.ci.auto_merge.time.sleep"):
assert wait_for_ci(client, "abc123", max_wait=10, poll_interval=5) is True
def test_ignores_non_ci_contexts(self) -> None:
client = MagicMock()
client.get_commit_status.return_value = [
_status("Auto-merge / merge (pull_request)", CI_PENDING),
_status("CI / quality (pull_request)", CI_SUCCESS),
]
assert wait_for_ci(client, "abc123", max_wait=10) is True
def test_deduplicates_by_latest(self) -> None:
"""Combined endpoint returns one entry per context; if multiple
entries appear, the last one wins (dict comprehension)."""
client = MagicMock()
client.get_commit_status.return_value = [
_status("CI / quality (pull_request)", CI_PENDING, "2026-01-01T00:00:00Z"),
_status("CI / quality (pull_request)", CI_SUCCESS, "2026-01-01T00:01:00Z"),
]
assert wait_for_ci(client, "abc123", max_wait=10) is True
def test_skipped_jobs_count_as_passing(self) -> None:
"""Conditional jobs that are skipped should not block merge."""
client = MagicMock()
client.get_commit_status.return_value = [
_status("CI / quality (pull_request)", CI_SUCCESS),
_status("CI / badges (pull_request)", "skipped"),
_status("CI / molecule-tests (pull_request)", "skipped"),
_status("CI / discover-runners (pull_request)", "skipped"),
]
assert wait_for_ci(client, "abc123", max_wait=10) is True
def test_only_non_ci_contexts_waits_then_ci_appears(self) -> None:
client = MagicMock()
client.get_commit_status.side_effect = [
[_status("Auto-merge / merge (pull_request)", CI_PENDING)],
[_status("CI / quality (pull_request)", CI_SUCCESS)],
]
with patch("scripts.ci.auto_merge.time.sleep"):
assert wait_for_ci(client, "abc123", max_wait=10, poll_interval=5) is True
assert extract_conventional_msg(commits) == "feat: add feature"
def _mock_pr(sha: str = "abc123def456") -> dict[str, object]:
return {"head": {"sha": sha}}
# -- run_cmd --
def _mock_ci_passing() -> list[dict[str, str]]:
return [_status("CI / quality (pull_request)", CI_SUCCESS)]
class TestRunCmd:
def test_success(self) -> None:
result = run_cmd(["echo", "hello"])
assert result.returncode == 0
def test_failure_raises(self) -> None:
with pytest.raises(click.ClickException, match="Command failed"):
run_cmd(["false"])
def test_failure_no_check(self) -> None:
result = run_cmd(["false"], check=False)
assert result.returncode != 0
def _mock_commits() -> list[dict[str, dict[str, str]]]:
return [_commit("fix: resolve timeout")]
# -- main (integration) --
class TestMain:
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch("scripts.ci.auto_merge.validate_pr_title_matches_vikunja")
@patch("scripts.ci.auto_merge.has_approval_review", return_value=True)
@patch.dict("os.environ", {"REPO_TOKEN": "tok", "VIKUNJA_TOKEN": ""}, clear=True)
@patch("scripts.ci.auto_merge.GiteaClient")
def test_successful_flow_with_label_arg(
self, mock_client_cls: MagicMock, mock_approval: MagicMock, mock_vikunja: MagicMock
) -> None:
def test_full_merge_flow(self, mock_client_cls: MagicMock, tmp_path, monkeypatch) -> None: # type: ignore[no-untyped-def]
monkeypatch.chdir(tmp_path)
(tmp_path / ".taskid").write_text("GRM-19\n")
mock_client = MagicMock()
mock_client.get_pr_commits.return_value = [
{"commit": {"message": "fix: resolve timeout"}},
]
mock_client_cls.return_value = mock_client
mock_client.get_pr.return_value = _mock_pr()
mock_client.get_commit_status.return_value = _mock_ci_passing()
mock_client.get_pr_commits.return_value = _mock_commits()
runner = CliRunner()
result = runner.invoke(
main,
["GRM-19-fix-bug", "GRM-19: Some task title", "owner/repo", "7", "ready-to-merge"],
["GRM-19-fix-bug", "GRM-19: Fix timeout", "owner/repo", "7"],
)
assert result.exit_code == 0
assert "squash-merged" in result.output
assert result.exit_code == 0, result.output
mock_client.merge_pr.assert_called_once_with("7", "GRM-19: fix: resolve timeout")
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch("scripts.ci.auto_merge.validate_pr_title_matches_vikunja")
@patch("scripts.ci.auto_merge.has_approval_review", return_value=True)
@patch("scripts.ci.auto_merge.GiteaClient")
def test_successful_flow_label_fallback(
self, mock_client_cls: MagicMock, mock_approval: MagicMock, mock_vikunja: MagicMock
) -> None:
"""Label not passed via arg, but PR has ready-to-merge via API."""
mock_client = MagicMock()
mock_client.get_pr_labels.return_value = [{"name": "ready-to-merge"}]
mock_client.get_pr.return_value = _mock_pr()
mock_client.get_commit_status.return_value = _mock_ci_passing()
mock_client.get_pr_commits.return_value = _mock_commits()
mock_client_cls.return_value = mock_client
runner = CliRunner()
result = runner.invoke(
main,
["GRM-19-fix-bug", "GRM-19: Some task title", "owner/repo", "7"],
)
assert result.exit_code == 0
assert "squash-merged" in result.output
mock_client.merge_pr.assert_called_once_with("7", "GRM-19: fix: resolve timeout")
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch("scripts.ci.auto_merge.validate_pr_title_matches_vikunja")
@patch("scripts.ci.auto_merge.has_approval_review", return_value=True)
@patch("scripts.ci.auto_merge.GiteaClient")
def test_wrong_label_skips_merge(
self, mock_client_cls: MagicMock, mock_approval: MagicMock, mock_vikunja: MagicMock
) -> None:
"""Label is not ready-to-merge and PR doesn't have it via API either."""
mock_client = MagicMock()
mock_client.get_pr_labels.return_value = [{"name": "bug"}]
mock_client_cls.return_value = mock_client
runner = CliRunner()
result = runner.invoke(
main,
["GRM-19-fix-bug", "GRM-19: Some task title", "owner/repo", "7", "bug"],
)
assert result.exit_code == 0
assert "skipping" in result.output
mock_client.merge_pr.assert_not_called()
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch("scripts.ci.auto_merge.validate_pr_title_matches_vikunja")
@patch("scripts.ci.auto_merge.has_approval_review", return_value=True)
@patch("scripts.ci.auto_merge.GiteaClient")
def test_empty_label_falls_back_to_api(
self, mock_client_cls: MagicMock, mock_approval: MagicMock, mock_vikunja: MagicMock
) -> None:
"""Gitea Actions doesn't populate label name, but API shows ready-to-merge."""
mock_client = MagicMock()
mock_client.get_pr_labels.return_value = [{"name": "ready-to-merge"}]
mock_client.get_pr.return_value = _mock_pr()
mock_client.get_commit_status.return_value = _mock_ci_passing()
mock_client.get_pr_commits.return_value = _mock_commits()
mock_client_cls.return_value = mock_client
runner = CliRunner()
result = runner.invoke(
main,
["GRM-19-fix-bug", "GRM-19: Some task title", "owner/repo", "7", ""],
)
assert result.exit_code == 0
assert "squash-merged" in result.output
@patch.dict("os.environ", {"REPO_TOKEN": ""}, clear=True)
def test_missing_token_exits(self) -> None:
def test_no_token_raises(self) -> None:
runner = CliRunner()
result = runner.invoke(main, ["branch", "title", "owner/repo", "1"])
assert result.exit_code == 1
result = runner.invoke(main, ["GRM-19-fix", "GRM-19: test", "owner/repo", "7"])
assert result.exit_code != 0
assert "REPO_TOKEN" in result.output
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch.dict("os.environ", {"REPO_TOKEN": "tok"}, clear=True)
@patch("scripts.ci.auto_merge.GiteaClient")
def test_missing_task_id_exits(self, mock_client_cls: MagicMock) -> None:
mock_client = MagicMock()
mock_client.get_pr_labels.return_value = [{"name": "ready-to-merge"}]
mock_client_cls.return_value = mock_client
def test_no_task_id_raises(self, mock_client_cls: MagicMock, tmp_path, monkeypatch) -> None: # type: ignore[no-untyped-def]
monkeypatch.chdir(tmp_path)
# No .taskid file, no GRM-N in branch name
runner = CliRunner()
result = runner.invoke(main, ["feature-no-id", "GRM-19: bug", "owner/repo", "1"])
assert result.exit_code == 1
assert "task ID" in result.output
result = runner.invoke(main, ["feature-branch", "GRM-19: test", "owner/repo", "7"])
assert result.exit_code != 0
assert "No task ID" in result.output
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch.dict("os.environ", {"REPO_TOKEN": "tok"}, clear=True)
@patch("scripts.ci.auto_merge.GiteaClient")
def test_invalid_pr_title_exits(self, mock_client_cls: MagicMock) -> None:
mock_client = MagicMock()
mock_client.get_pr_labels.return_value = [{"name": "ready-to-merge"}]
mock_client_cls.return_value = mock_client
def test_invalid_pr_title_raises(self, mock_client_cls: MagicMock, tmp_path, monkeypatch) -> None: # type: ignore[no-untyped-def]
monkeypatch.chdir(tmp_path)
(tmp_path / ".taskid").write_text("GRM-19\n")
runner = CliRunner()
result = runner.invoke(main, ["GRM-19-fix", "random title", "owner/repo", "1"])
assert result.exit_code == 1
assert "GRM-N" in result.output
result = runner.invoke(main, ["GRM-19-fix", "Bad title", "owner/repo", "7"])
assert result.exit_code != 0
assert "format" in result.output.lower()
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch.dict("os.environ", {"REPO_TOKEN": "tok"}, clear=True)
@patch("scripts.ci.auto_merge.GiteaClient")
def test_pr_title_task_id_mismatch_exits(self, mock_client_cls: MagicMock) -> None:
"""PR title has a different task ID than the branch."""
def test_merge_behind_master_rebases(self, mock_client_cls: MagicMock, tmp_path, monkeypatch) -> None: # type: ignore[no-untyped-def]
monkeypatch.chdir(tmp_path)
(tmp_path / ".taskid").write_text("GRM-19\n")
mock_client = MagicMock()
mock_client.get_pr_labels.return_value = [{"name": "ready-to-merge"}]
mock_client.get_pr_commits.return_value = [
{"commit": {"message": "fix: resolve timeout"}},
]
mock_client.merge_pr.side_effect = [
APIError(405, "HEAD branch is behind master"),
None, # Second call succeeds
]
mock_client_cls.return_value = mock_client
runner = CliRunner()
result = runner.invoke(main, ["GRM-19-fix", "GRM-42: Some task title", "owner/repo", "1"])
assert result.exit_code == 1
assert "mismatch" in result.output
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch("scripts.ci.auto_merge.validate_pr_title_matches_vikunja")
@patch("scripts.ci.auto_merge.has_approval_review", return_value=False)
with patch("scripts.ci.auto_merge.run_cmd") as mock_run:
runner = CliRunner()
result = runner.invoke(
main,
["GRM-19-fix-bug", "GRM-19: Fix timeout", "owner/repo", "7"],
)
assert result.exit_code == 0, result.output
assert mock_client.merge_pr.call_count == 2
# Should have fetched, rebased, and pushed
assert mock_run.call_count == 3
@patch.dict("os.environ", {"REPO_TOKEN": "tok"}, clear=True)
@patch("scripts.ci.auto_merge.GiteaClient")
def test_no_approval_review_blocks_merge(
self, mock_client_cls: MagicMock, mock_approval: MagicMock, mock_vikunja: MagicMock
) -> None:
"""No APPROVE review — merge should be blocked."""
def test_merge_failure_raises(self, mock_client_cls: MagicMock, tmp_path, monkeypatch) -> None: # type: ignore[no-untyped-def]
monkeypatch.chdir(tmp_path)
(tmp_path / ".taskid").write_text("GRM-19\n")
mock_client = MagicMock()
mock_client.get_pr_labels.return_value = [{"name": "ready-to-merge"}]
mock_client.get_pr_commits.return_value = [
{"commit": {"message": "fix: resolve timeout"}},
]
mock_client.merge_pr.side_effect = APIError(409, "Conflict")
mock_client_cls.return_value = mock_client
runner = CliRunner()
result = runner.invoke(main, ["GRM-19-fix", "GRM-19: Some task title", "owner/repo", "1"])
assert result.exit_code == 1
assert "APPROVE review" in result.output
mock_client.merge_pr.assert_not_called()
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch("scripts.ci.auto_merge.validate_pr_title_matches_vikunja")
@patch("scripts.ci.auto_merge.has_approval_review", return_value=True)
runner = CliRunner()
result = runner.invoke(
main,
["GRM-19-fix-bug", "GRM-19: Fix timeout", "owner/repo", "7"],
)
assert result.exit_code != 0
assert "Merge failed" in result.output
@patch.dict("os.environ", {"REPO_TOKEN": "tok"}, clear=True)
@patch("scripts.ci.auto_merge.GiteaClient")
def test_empty_commits_exits(
self, mock_client_cls: MagicMock, mock_approval: MagicMock, mock_vikunja: MagicMock
) -> None:
"""PR has no commits — cannot extract conventional message."""
def test_no_conventional_msg_raises(self, mock_client_cls: MagicMock, tmp_path, monkeypatch) -> None: # type: ignore[no-untyped-def]
"""When no conventional commit message is found in PR commits, raises."""
monkeypatch.chdir(tmp_path)
(tmp_path / ".taskid").write_text("GRM-19\n")
mock_client = MagicMock()
mock_client.get_pr_labels.return_value = [{"name": "ready-to-merge"}]
mock_client.get_pr.return_value = _mock_pr()
mock_client.get_commit_status.return_value = _mock_ci_passing()
mock_client.get_pr_commits.return_value = []
mock_client_cls.return_value = mock_client
runner = CliRunner()
result = runner.invoke(main, ["GRM-19-fix", "GRM-19: Some task title", "owner/repo", "1"])
assert result.exit_code == 1
assert "conventional commit" in result.output
mock_client.merge_pr.assert_not_called()
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch("scripts.ci.auto_merge.validate_pr_title_matches_vikunja")
@patch("scripts.ci.auto_merge.has_approval_review", return_value=True)
@patch("scripts.ci.auto_merge.GiteaClient")
def test_merge_pr_failure_raises_click(
self, mock_client_cls: MagicMock, mock_approval: MagicMock, mock_vikunja: MagicMock
) -> None:
mock_client = MagicMock()
mock_client.get_pr_labels.return_value = [{"name": "ready-to-merge"}]
mock_client.get_pr.return_value = _mock_pr()
mock_client.get_commit_status.return_value = _mock_ci_passing()
mock_client.get_pr_commits.return_value = _mock_commits()
mock_client.merge_pr.side_effect = APIError(http.HTTPStatus.INTERNAL_SERVER_ERROR, "server error")
mock_client_cls.return_value = mock_client
runner = CliRunner()
result = runner.invoke(main, ["GRM-19-fix", "GRM-19: Some task title", "owner/repo", "1"])
assert result.exit_code == 1
assert "HTTP" in result.output
result = runner.invoke(
main,
["GRM-19-fix-bug", "GRM-19: Fix timeout", "owner/repo", "7"],
)
assert result.exit_code != 0
assert "conventional commit" in result.output.lower()
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch("scripts.ci.auto_merge.validate_pr_title_matches_vikunja")
@patch("scripts.ci.auto_merge.has_approval_review", return_value=True)
@patch.dict("os.environ", {"REPO_TOKEN": "tok"}, clear=True)
@patch("scripts.ci.auto_merge.GiteaClient")
def test_merge_pr_json_parse_failure(
self, mock_client_cls: MagicMock, mock_approval: MagicMock, mock_vikunja: MagicMock
) -> None:
mock_client = MagicMock()
mock_client.get_pr_labels.return_value = [{"name": "ready-to-merge"}]
mock_client.get_pr.return_value = _mock_pr()
mock_client.get_commit_status.return_value = _mock_ci_passing()
mock_client.get_pr_commits.return_value = _mock_commits()
mock_client.merge_pr.side_effect = APIError(http.HTTPStatus.BAD_GATEWAY, "bad gateway")
mock_client_cls.return_value = mock_client
runner = CliRunner()
result = runner.invoke(main, ["GRM-19-fix", "GRM-19: Some task title", "owner/repo", "1"])
assert result.exit_code == 1
assert str(http.HTTPStatus.BAD_GATEWAY) in result.output
def test_rebase_retry_failure_raises(self, mock_client_cls: MagicMock, tmp_path, monkeypatch) -> None: # type: ignore[no-untyped-def]
"""When rebase retry also fails, raises with helpful message."""
monkeypatch.chdir(tmp_path)
(tmp_path / ".taskid").write_text("GRM-19\n")
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch("scripts.ci.auto_merge.validate_pr_title_matches_vikunja")
@patch("scripts.ci.auto_merge.has_approval_review", return_value=True)
@patch("scripts.ci.auto_merge.GiteaClient")
def test_ci_failure_blocks_merge(
self, mock_client_cls: MagicMock, mock_approval: MagicMock, mock_vikunja: MagicMock
) -> None:
"""CI checks fail — merge should not be attempted."""
mock_client = MagicMock()
mock_client.get_pr_labels.return_value = [{"name": "ready-to-merge"}]
mock_client.get_pr.return_value = _mock_pr()
mock_client.get_commit_status.return_value = [
_status("CI / quality (pull_request)", CI_FAILURE),
mock_client.get_pr_commits.return_value = [
{"commit": {"message": "fix: resolve timeout"}},
]
mock_client.merge_pr.side_effect = APIError(405, "HEAD branch is behind master")
mock_client_cls.return_value = mock_client
runner = CliRunner()
result = runner.invoke(main, ["GRM-19-fix", "GRM-19: Some task title", "owner/repo", "1"])
assert result.exit_code == 1
assert "CI checks did not pass" in result.output
mock_client.merge_pr.assert_not_called()
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch("scripts.ci.auto_merge.validate_pr_title_matches_vikunja")
@patch("scripts.ci.auto_merge.has_approval_review", return_value=True)
@patch("scripts.ci.auto_merge.GiteaClient")
def test_no_sha_proceeds_without_wait(
self, mock_client_cls: MagicMock, mock_approval: MagicMock, mock_vikunja: MagicMock
) -> None:
"""PR head SHA missing — should proceed without waiting."""
mock_client = MagicMock()
mock_client.get_pr_labels.return_value = [{"name": "ready-to-merge"}]
mock_client.get_pr.return_value = {"head": {}}
mock_client.get_pr_commits.return_value = _mock_commits()
mock_client_cls.return_value = mock_client
runner = CliRunner()
result = runner.invoke(main, ["GRM-19-fix", "GRM-19: Some task title", "owner/repo", "1"])
assert result.exit_code == 0
assert "squash-merged" in result.output
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch("scripts.ci.auto_merge.validate_pr_title_matches_vikunja")
@patch("scripts.ci.auto_merge.has_approval_review", return_value=True)
@patch("scripts.ci.auto_merge.GiteaClient")
def test_merge_405_behind_retries(
self, mock_client_cls: MagicMock, mock_approval: MagicMock, mock_vikunja: MagicMock
) -> None:
"""405 'behind' error should trigger rebase and retry."""
mock_client = MagicMock()
mock_client.get_pr_labels.return_value = [{"name": "ready-to-merge"}]
mock_client.get_pr.return_value = _mock_pr()
mock_client.get_commit_status.return_value = _mock_ci_passing()
mock_client.get_pr_commits.return_value = _mock_commits()
# First merge_pr raises 405 "behind", second succeeds
mock_client.merge_pr.side_effect = [
APIError(http.HTTPStatus.METHOD_NOT_ALLOWED, "head branch is behind base"),
None,
]
mock_client_cls.return_value = mock_client
with patch("scripts.ci.auto_merge.run_cmd") as mock_run_cmd:
mock_run_cmd.return_value = MagicMock(returncode=0, stdout="", stderr="")
with patch("scripts.ci.auto_merge.run_cmd") as mock_run:
mock_run.side_effect = click.ClickException("git rebase failed")
runner = CliRunner()
result = runner.invoke(main, ["GRM-19-fix", "GRM-19: Some task title", "owner/repo", "1"])
assert result.exit_code == 0
assert "Rebased" in result.output or "rebase" in result.output.lower()
assert mock_client.merge_pr.call_count == 2
result = runner.invoke(
main,
["GRM-19-fix-bug", "GRM-19: Fix timeout", "owner/repo", "7"],
)
assert result.exit_code != 0
assert "rebase" in result.output.lower()
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch("scripts.ci.auto_merge.validate_pr_title_matches_vikunja")
@patch("scripts.ci.auto_merge.has_approval_review", return_value=True)
@patch("scripts.ci.auto_merge.GiteaClient")
def test_merge_405_behind_rebase_fails(
self, mock_client_cls: MagicMock, mock_approval: MagicMock, mock_vikunja: MagicMock
) -> None:
"""405 'behind' with rebase failure should raise ClickException."""
mock_client = MagicMock()
mock_client.get_pr_labels.return_value = [{"name": "ready-to-merge"}]
mock_client.get_pr.return_value = _mock_pr()
mock_client.get_commit_status.return_value = _mock_ci_passing()
mock_client.get_pr_commits.return_value = _mock_commits()
mock_client.merge_pr.side_effect = APIError(http.HTTPStatus.METHOD_NOT_ALLOWED, "head branch is behind base")
mock_client_cls.return_value = mock_client
with patch("scripts.ci.auto_merge.run_cmd") as mock_run_cmd:
mock_run_cmd.side_effect = click.ClickException("rebase failed")
runner = CliRunner()
result = runner.invoke(main, ["GRM-19-fix", "GRM-19: Some task title", "owner/repo", "1"])
assert result.exit_code == 1
assert "rebase" in result.output.lower() or "retry" in result.output.lower()
def test_run_cmd_success(self) -> None:
"""run_cmd should return CompletedProcess on success."""
with patch("scripts.ci.auto_merge.subprocess.run") as mock_run:
mock_run.return_value = MagicMock(returncode=0, stdout="ok", stderr="")
result = run_cmd(["echo", "ok"])
assert result.returncode == 0
def test_main_module_block() -> None:
"""Test that the __main__ block can be executed."""
import scripts.ci.auto_merge as am
def test_run_cmd_failure_raises(self) -> None:
"""run_cmd should raise ClickException on non-zero exit."""
with patch("scripts.ci.auto_merge.subprocess.run") as mock_run:
mock_run.return_value = MagicMock(returncode=1, stdout="", stderr="error")
with pytest.raises(click.ClickException):
run_cmd(["false"])
with open(am.__file__) as f:
source = f.read()
source = source.replace('if __name__ == "__main__":\n main()\n', "")
namespace = dict(am.__dict__)
exec(compile(source, am.__file__, "exec"), namespace)
# Verify main is callable
assert callable(namespace["main"])
-4
View File
@@ -59,10 +59,6 @@ class TestIsUserFacing:
"""api_clients.py is used only by CI/CD scripts, not by the GRM CLI."""
assert is_user_facing("src/gitea_runner_manager/api_clients.py") is False
def test_review_checklist_is_not_user_facing(self) -> None:
"""REVIEW_CHECKLIST.md is agent infrastructure, not user-facing."""
assert is_user_facing("REVIEW_CHECKLIST.md") is False
def test_docs_are_not_user_facing(self) -> None:
assert is_user_facing("docs/user/getting-started.md") is False
-5
View File
@@ -6,7 +6,6 @@ from gitea_runner_manager.config import (
DEFAULT_PER_PAGE,
DEFAULT_TIMEOUT,
GITEA_API_URL,
LABEL_CONFIG,
REPO_NAME,
REPO_OWNER,
TASK_ID_RE,
@@ -55,7 +54,3 @@ class TestConfigConstants:
assert len(contexts) == 4
assert "CI / quality (pull_request)" in contexts
assert any("molecule-tests" in c for c in contexts)
def test_label_config(self) -> None:
assert LABEL_CONFIG["name"] == "ready-to-merge"
assert LABEL_CONFIG["color"] == "2ecc71"
+30 -151
View File
@@ -1,6 +1,5 @@
"""Unit tests for scripts/configure_repo.py."""
import http
from unittest.mock import MagicMock, patch
import click
@@ -8,181 +7,61 @@ import pytest
from gitea_runner_manager.config import BRANCH_PROTECTION_CONFIG, REPO_SETTINGS_CONFIG
from gitea_runner_manager.exceptions import APIError
from scripts.configure_repo import (
_ensure_label_via_client,
_ensure_label_via_tea,
_handle_http_error,
main,
)
from scripts.gitea_cli import TeaCLIError
from scripts.configure_repo import _handle_http_error, main
class TestHandleHttpError:
def test_handle_http_error_403(self) -> None:
err = APIError(http.HTTPStatus.FORBIDDEN, "Forbidden")
with pytest.raises(click.ClickException) as exc:
_handle_http_error(err)
msg = str(exc.value)
assert "admin rights" in msg
assert "Settings → Branches" in msg
def test_forbidden_raises_click_exception(self) -> None:
with pytest.raises(click.ClickException, match="Forbidden"):
_handle_http_error(APIError(403, "Forbidden"))
def test_handle_http_error_other(self) -> None:
err = APIError(http.HTTPStatus.INTERNAL_SERVER_ERROR, "Internal Server Error")
with pytest.raises(click.ClickException) as exc:
_handle_http_error(err)
assert str(http.HTTPStatus.INTERNAL_SERVER_ERROR) in str(exc.value)
def test_handle_http_error_json_parse_fails(self) -> None:
err = APIError(http.HTTPStatus.BAD_GATEWAY, "bad gateway")
with pytest.raises(click.ClickException) as exc:
_handle_http_error(err)
assert str(http.HTTPStatus.BAD_GATEWAY) in str(exc.value)
class TestEnsureLabelViaTea:
def test_creates_new_label(self) -> None:
mock_tea = MagicMock()
mock_tea.list_labels.return_value = [{"name": "bug"}]
result = _ensure_label_via_tea(mock_tea, "owner/repo", "ready-to-merge", "2ecc71", "desc")
assert result is True
mock_tea.create_label.assert_called_once()
def test_label_already_exists(self) -> None:
mock_tea = MagicMock()
mock_tea.list_labels.return_value = [{"name": "ready-to-merge"}]
result = _ensure_label_via_tea(mock_tea, "owner/repo", "ready-to-merge", "2ecc71", "desc")
assert result is False
mock_tea.create_label.assert_not_called()
def test_tea_error_falls_back_to_client(self) -> None:
mock_tea = MagicMock()
mock_tea.list_labels.side_effect = TeaCLIError("network error")
with patch("scripts.configure_repo._ensure_label_via_client", return_value=True) as mock_fallback:
result = _ensure_label_via_tea(mock_tea, "owner/repo", "ready-to-merge", "2ecc71", "desc")
assert result is True
mock_fallback.assert_called_once_with("ready-to-merge", "2ecc71", "desc")
def test_tea_not_installed_falls_back_to_client(self) -> None:
"""When tea CLI is not installed (FileNotFoundError), fall back to GiteaClient."""
mock_tea = MagicMock()
mock_tea.list_labels.side_effect = FileNotFoundError("[Errno 2] No such file or directory: 'tea'")
with patch("scripts.configure_repo._ensure_label_via_client", return_value=True) as mock_fallback:
result = _ensure_label_via_tea(mock_tea, "owner/repo", "ready-to-merge", "2ecc71", "desc")
assert result is True
mock_fallback.assert_called_once_with("ready-to-merge", "2ecc71", "desc")
class TestEnsureLabelViaClient:
@patch.dict("os.environ", {"REPO_TOKEN": "tok"}, clear=True)
@patch("scripts.configure_repo.GiteaClient")
def test_creates_label(self, mock_client_cls: MagicMock) -> None:
mock_client = MagicMock()
mock_client.ensure_label.return_value = {"id": 1}
mock_client_cls.return_value = mock_client
result = _ensure_label_via_client("bug", "ff0000", "A bug")
assert result is True
@patch.dict("os.environ", {"REPO_TOKEN": "tok"}, clear=True)
@patch("scripts.configure_repo.GiteaClient")
def test_label_already_exists(self, mock_client_cls: MagicMock) -> None:
mock_client = MagicMock()
mock_client.ensure_label.return_value = None
mock_client_cls.return_value = mock_client
result = _ensure_label_via_client("bug", "ff0000", "A bug")
assert result is False
def test_other_error_raises_click_exception(self) -> None:
with pytest.raises(click.ClickException, match="HTTP error"):
_handle_http_error(APIError(500, "Server error"))
class TestMain:
def test_main_missing_token(self) -> None:
with patch.dict("os.environ", {}, clear=True):
with pytest.raises(click.ClickException) as exc:
main()
assert "REPO_TOKEN" in str(exc.value)
@patch.dict("os.environ", {"REPO_TOKEN": "tok"}, clear=True)
@patch("scripts.configure_repo.TeaCLI")
@patch("scripts.configure_repo.GiteaClient")
def test_main_success(self, mock_client_cls: MagicMock, mock_tea_cls: MagicMock) -> None:
def test_main_success(self, mock_client_cls: MagicMock) -> None:
mock_client = MagicMock()
mock_client_cls.return_value = mock_client
mock_tea = MagicMock()
mock_tea.list_labels.return_value = [] # No existing labels
mock_tea_cls.return_value = mock_tea
main()
with click.Context(click.Command("test")):
main()
mock_client.ensure_branch_protection.assert_called_once_with("master", BRANCH_PROTECTION_CONFIG)
mock_tea.create_label.assert_called_once()
mock_client.update_repo_settings.assert_called_once_with(REPO_SETTINGS_CONFIG)
@patch.dict("os.environ", {"REPO_TOKEN": "tok"}, clear=True)
@patch("scripts.configure_repo.TeaCLI")
@patch("scripts.configure_repo.GiteaClient")
def test_main_label_already_exists(self, mock_client_cls: MagicMock, mock_tea_cls: MagicMock) -> None:
def test_main_api_error(self, mock_client_cls: MagicMock) -> None:
mock_client = MagicMock()
mock_client.ensure_branch_protection.side_effect = APIError(403, "Forbidden")
mock_client_cls.return_value = mock_client
mock_tea = MagicMock()
mock_tea.list_labels.return_value = [{"name": "ready-to-merge"}]
mock_tea_cls.return_value = mock_tea
main()
with pytest.raises(click.ClickException, match="Forbidden"):
main()
mock_client.ensure_branch_protection.assert_called_once_with("master", BRANCH_PROTECTION_CONFIG)
mock_tea.create_label.assert_not_called()
mock_client.update_repo_settings.assert_called_once_with(REPO_SETTINGS_CONFIG)
@patch.dict("os.environ", {"REPO_TOKEN": "tok"}, clear=True)
@patch("scripts.configure_repo.TeaCLI")
@patch("scripts.configure_repo.GiteaClient")
def test_main_tea_error_falls_back(self, mock_client_cls: MagicMock, mock_tea_cls: MagicMock) -> None:
mock_client = MagicMock()
mock_client.ensure_label.return_value = {"id": 1}
mock_client_cls.return_value = mock_client
mock_tea = MagicMock()
mock_tea.list_labels.side_effect = TeaCLIError("network error")
mock_tea_cls.return_value = mock_tea
main()
mock_client.ensure_branch_protection.assert_called_once()
# Fallback to GiteaClient for label creation
mock_client.ensure_label.assert_called_once()
mock_client.update_repo_settings.assert_called_once()
@patch.dict("os.environ", {"REPO_TOKEN": "tok"}, clear=True)
@patch("scripts.configure_repo.TeaCLI")
@patch("scripts.configure_repo.GiteaClient")
def test_main_api_error(self, mock_client_cls: MagicMock, mock_tea_cls: MagicMock) -> None:
mock_client = MagicMock()
mock_client.ensure_branch_protection.side_effect = APIError(http.HTTPStatus.FORBIDDEN, "Forbidden")
mock_client_cls.return_value = mock_client
mock_tea = MagicMock()
mock_tea_cls.return_value = mock_tea
with pytest.raises(click.ClickException) as exc:
@patch.dict("os.environ", {}, clear=True)
def test_main_no_token(self) -> None:
with pytest.raises(click.ClickException, match="REPO_TOKEN"):
main()
assert "HTTP" in str(exc.value)
def test_main_module_block() -> None:
with patch.dict("os.environ", {"REPO_TOKEN": "tok"}, clear=True):
with patch("scripts.configure_repo.TeaCLI") as mock_tea_cls:
with patch("scripts.configure_repo.GiteaClient") as mock_client_cls:
mock_client = MagicMock()
mock_client_cls.return_value = mock_client
mock_tea = MagicMock()
mock_tea.list_labels.return_value = []
mock_tea_cls.return_value = mock_tea
import scripts.configure_repo as cr
with patch("scripts.configure_repo.GiteaClient") as mock_client_cls:
mock_client = MagicMock()
mock_client_cls.return_value = mock_client
import scripts.configure_repo as cr
with open(cr.__file__) as f:
source = f.read()
# Remove __main__ block so exec doesn't call main() before we inject the mock
source = source.replace('if __name__ == "__main__":\n main()\n', "")
namespace = dict(cr.__dict__)
exec(compile(source, cr.__file__, "exec"), namespace)
namespace["GiteaClient"] = mock_client_cls
namespace["TeaCLI"] = mock_tea_cls
namespace["main"]()
mock_client.ensure_branch_protection.assert_called_once_with("master", BRANCH_PROTECTION_CONFIG)
mock_client.update_repo_settings.assert_called_once_with(REPO_SETTINGS_CONFIG)
with open(cr.__file__) as f:
source = f.read()
source = source.replace('if __name__ == "__main__":\n main() # pragma: no cover\n', "")
namespace = dict(cr.__dict__)
exec(compile(source, cr.__file__, "exec"), namespace)
namespace["GiteaClient"] = mock_client_cls
namespace["main"]()
mock_client.ensure_branch_protection.assert_called_once_with("master", BRANCH_PROTECTION_CONFIG)
mock_client.update_repo_settings.assert_called_once_with(REPO_SETTINGS_CONFIG)
+3 -4
View File
@@ -507,12 +507,11 @@ class TestBuildReviewBody:
body = build_review_body(result)
assert "No issues found" in body
def test_body_contains_checklist_reference(self) -> None:
"""Review body must reference REVIEW_CHECKLIST.md for manual review."""
def test_body_contains_auto_merge_note(self) -> None:
"""Review body must mention auto-merge."""
result = ReviewResult()
body = build_review_body(result)
assert "REVIEW_CHECKLIST.md" in body
assert "--checklist-confirmed" in body
assert "Auto-merge" in body
class TestRunReview:
-277
View File
@@ -1,277 +0,0 @@
"""Unit tests for scripts/ci/review_pr.py."""
import http
import json
from unittest.mock import MagicMock, patch
import click
import pytest
from click.testing import CliRunner
from gitea_runner_manager.exceptions import APIError
from scripts.ci.review_pr import main, parse_comments
class TestParseComments:
def test_parse_from_json_file(self, tmp_path) -> None:
comments = [{"path": "a.py", "body": "fix", "new_position": 1}]
f = tmp_path / "comments.json"
f.write_text(json.dumps(comments))
assert parse_comments(str(f), False) == comments
def test_parse_from_stdin(self) -> None:
comments = [{"path": "a.py", "body": "fix", "new_position": 1}]
with patch("scripts.ci.review_pr.sys.stdin") as mock_stdin:
mock_stdin.read.return_value = json.dumps(comments)
assert parse_comments(None, True) == comments
def test_no_comments_returns_empty(self) -> None:
assert parse_comments(None, False) == []
def test_invalid_json_file_raises(self, tmp_path) -> None:
f = tmp_path / "comments.json"
f.write_text("not json{")
with pytest.raises(click.ClickException):
parse_comments(str(f), False)
def test_non_list_json_raises(self, tmp_path) -> None:
f = tmp_path / "comments.json"
f.write_text(json.dumps({"path": "a.py"}))
with pytest.raises(click.ClickException):
parse_comments(str(f), False)
def test_stdin_non_list_raises(self) -> None:
with patch("scripts.ci.review_pr.sys.stdin") as mock_stdin:
mock_stdin.read.return_value = json.dumps({"path": "a.py"})
with pytest.raises(click.ClickException):
parse_comments(None, True)
def test_stdin_invalid_json_raises(self) -> None:
with patch("scripts.ci.review_pr.sys.stdin") as mock_stdin:
mock_stdin.read.return_value = "not json{"
with pytest.raises(click.ClickException):
parse_comments(None, True)
def test_stdin_empty_returns_empty(self) -> None:
with patch("scripts.ci.review_pr.sys.stdin") as mock_stdin:
mock_stdin.read.return_value = " "
assert parse_comments(None, True) == []
class TestMain:
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch("scripts.ci.review_pr.GiteaClient")
def test_successful_comment_review(self, mock_client_cls: MagicMock) -> None:
mock_client = MagicMock()
mock_client.create_review.return_value = {"id": 42}
mock_client_cls.return_value = mock_client
runner = CliRunner()
result = runner.invoke(
main,
["5", "owner/repo", "--event", "COMMENT", "--body", "LGTM"],
)
assert result.exit_code == 0
assert "Review #42" in result.output
mock_client.create_review.assert_called_once_with("5", event="COMMENT", body="LGTM", comments=[])
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch("scripts.ci.review_pr.GiteaClient")
def test_successful_approve_review(self, mock_client_cls: MagicMock) -> None:
"""APPROVE requires --checklist-confirmed, --checklist-categories, and substantive body."""
mock_client = MagicMock()
mock_client.create_review.return_value = {"id": 7}
mock_client_cls.return_value = mock_client
runner = CliRunner()
result = runner.invoke(
main,
[
"5",
"owner/repo",
"--event",
"APPROVE",
"--checklist-confirmed",
"--checklist-categories",
"1,2,3,4,5,6,7,8,9,10,11,12,13",
"--body",
"All 13 checklist categories verified. Architecture OK, tests pass.",
],
)
assert result.exit_code == 0
assert "Review #7" in result.output
mock_client.create_review.assert_called_once_with(
"5",
event="APPROVE",
body="All 13 checklist categories verified. Architecture OK, tests pass.",
comments=[],
)
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch("scripts.ci.review_pr.GiteaClient")
def test_approve_without_checklist_confirmed_fails(self, mock_client_cls: MagicMock) -> None:
"""APPROVE without --checklist-confirmed is rejected."""
mock_client = MagicMock()
mock_client_cls.return_value = mock_client
runner = CliRunner()
result = runner.invoke(
main,
["5", "owner/repo", "--event", "APPROVE", "--body", "Looks good to me"],
)
assert result.exit_code != 0
assert "checklist" in result.output.lower()
mock_client.create_review.assert_not_called()
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch("scripts.ci.review_pr.GiteaClient")
def test_approve_without_checklist_categories_fails(self, mock_client_cls: MagicMock) -> None:
"""APPROVE without --checklist-categories is rejected."""
mock_client = MagicMock()
mock_client_cls.return_value = mock_client
runner = CliRunner()
result = runner.invoke(
main,
[
"5",
"owner/repo",
"--event",
"APPROVE",
"--checklist-confirmed",
"--body",
"All categories verified. Architecture OK, tests pass, docs updated.",
],
)
assert result.exit_code != 0
assert "checklist-categories" in result.output.lower()
mock_client.create_review.assert_not_called()
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch("scripts.ci.review_pr.GiteaClient")
def test_approve_with_too_few_categories_fails(self, mock_client_cls: MagicMock) -> None:
"""APPROVE with fewer than 8 categories is rejected."""
mock_client = MagicMock()
mock_client_cls.return_value = mock_client
runner = CliRunner()
result = runner.invoke(
main,
[
"5",
"owner/repo",
"--event",
"APPROVE",
"--checklist-confirmed",
"--checklist-categories",
"1,2,3",
"--body",
"All categories verified. Architecture OK, tests pass, docs updated.",
],
)
assert result.exit_code != 0
assert "8 of 13" in result.output
mock_client.create_review.assert_not_called()
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch("scripts.ci.review_pr.GiteaClient")
def test_approve_with_trivial_body_fails(self, mock_client_cls: MagicMock) -> None:
"""APPROVE with trivial body (< 50 chars) and no comments is rejected."""
mock_client = MagicMock()
mock_client_cls.return_value = mock_client
runner = CliRunner()
result = runner.invoke(
main,
[
"5",
"owner/repo",
"--event",
"APPROVE",
"--checklist-confirmed",
"--checklist-categories",
"1,2,3,4,5,6,7,8",
"--body",
"LGTM",
],
)
assert result.exit_code != 0
assert "substantive" in result.output.lower()
mock_client.create_review.assert_not_called()
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch("scripts.ci.review_pr.GiteaClient")
def test_successful_with_inline_comments(self, mock_client_cls: MagicMock, tmp_path) -> None:
comments = [{"path": "a.py", "body": "fix", "new_position": 1}]
f = tmp_path / "comments.json"
f.write_text(json.dumps(comments))
mock_client = MagicMock()
mock_client.create_review.return_value = {"id": 9}
mock_client_cls.return_value = mock_client
runner = CliRunner()
result = runner.invoke(
main,
["5", "owner/repo", "--comments-json", str(f)],
)
assert result.exit_code == 0
mock_client.create_review.assert_called_once_with("5", event="COMMENT", body="", comments=comments)
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch("scripts.ci.review_pr.GiteaClient")
def test_successful_with_stdin_comments(self, mock_client_cls: MagicMock) -> None:
comments = [{"path": "a.py", "body": "fix", "new_position": 1}]
mock_client = MagicMock()
mock_client.create_review.return_value = {"id": 11}
mock_client_cls.return_value = mock_client
runner = CliRunner()
result = runner.invoke(
main,
["5", "owner/repo", "--comments-stdin"],
input=json.dumps(comments),
)
assert result.exit_code == 0
mock_client.create_review.assert_called_once_with("5", event="COMMENT", body="", comments=comments)
@patch.dict("os.environ", {"REPO_TOKEN": ""}, clear=True)
def test_missing_token_exits(self) -> None:
runner = CliRunner()
result = runner.invoke(main, ["5", "owner/repo", "--body", "x"])
assert result.exit_code == 1
assert "REPO_TOKEN" in result.output
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch("scripts.ci.review_pr.GiteaClient")
def test_no_body_or_comments_for_comment_event(self, mock_client_cls: MagicMock) -> None:
mock_client = MagicMock()
mock_client_cls.return_value = mock_client
runner = CliRunner()
result = runner.invoke(main, ["5", "owner/repo", "--event", "COMMENT"])
assert result.exit_code == 1
assert "required" in result.output
mock_client.create_review.assert_not_called()
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch("scripts.ci.review_pr.GiteaClient")
def test_no_body_or_comments_for_request_changes(self, mock_client_cls: MagicMock) -> None:
mock_client = MagicMock()
mock_client_cls.return_value = mock_client
runner = CliRunner()
result = runner.invoke(main, ["5", "owner/repo", "--event", "REQUEST_CHANGES"])
assert result.exit_code == 1
assert "required" in result.output
mock_client.create_review.assert_not_called()
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch("scripts.ci.review_pr.GiteaClient")
def test_api_error_raises_click(self, mock_client_cls: MagicMock) -> None:
mock_client = MagicMock()
mock_client.create_review.side_effect = APIError(http.HTTPStatus.INTERNAL_SERVER_ERROR, "server error")
mock_client_cls.return_value = mock_client
runner = CliRunner()
result = runner.invoke(main, ["5", "owner/repo", "--body", "x"])
assert result.exit_code == 1
assert "HTTP" in result.output
@patch.dict("os.environ", {"REPO_TOKEN": "tok"})
@patch("scripts.ci.review_pr.GiteaClient")
def test_invalid_event_choice(self, mock_client_cls: MagicMock) -> None:
mock_client = MagicMock()
mock_client_cls.return_value = mock_client
runner = CliRunner()
result = runner.invoke(main, ["5", "owner/repo", "--event", "Bogus"])
assert result.exit_code != 0
mock_client.create_review.assert_not_called()