468 lines
17 KiB
Python
468 lines
17 KiB
Python
#!/usr/bin/env python3
|
|
"""Automated PR review: check architecture compliance, best practices, and quality.
|
|
|
|
Fetches the PR diff via the Gitea API, runs a series of automated checks,
|
|
and posts a structured review using GiteaClient.create_review.
|
|
|
|
Checks performed:
|
|
1. Architecture compliance — no business logic in CLI, no direct subprocess
|
|
calls outside executor, no hardcoded config that should be in config.py
|
|
2. Best practices — no bare except, no print() (use click.echo), no TODO/FIXME
|
|
left in merged code, no functions > 50 lines
|
|
3. Security — no secrets in code, no shell=True, no eval/exec
|
|
4. Documentation — new CLI commands documented, new modules in architecture.md
|
|
5. Test coverage — 100% enforced by pytest-cov (checked in quality job)
|
|
6. Commit conventions — conventional commit format on branch commits
|
|
|
|
Usage:
|
|
REPO_TOKEN=<token> python3 scripts/ci/pr_review.py <pr_number> <owner/repo>
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
from dataclasses import dataclass, field
|
|
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)
|
|
|
|
# Files that are exempt from certain checks
|
|
WORKFLOW_ONLY_SUFFIXES = (".yml", ".yaml", ".md", ".json", ".toml", ".cfg", ".ini", ".txt")
|
|
PYTHON_SUFFIX = ".py"
|
|
|
|
# Architecture rules
|
|
CLI_FILE = "src/gitea_runner_manager/cli.py"
|
|
EXECUTOR_FILE = "src/gitea_runner_manager/executor.py"
|
|
CONFIG_FILE = "src/gitea_runner_manager/config.py"
|
|
|
|
# Patterns that indicate business logic in CLI (should be in runner_manager.py)
|
|
BUSINESS_LOGIC_IN_CLI = [
|
|
(r"subprocess\.(run|call|Popen|check_output|check_call)", "subprocess call in CLI — delegate to executor.py"),
|
|
(r"\bos\.system\b", "os.system call in CLI — delegate to executor.py"),
|
|
(r"\bansible-playbook\b", "ansible-playbook reference in CLI — delegate to executor.py"),
|
|
]
|
|
|
|
# Patterns that indicate bad practices
|
|
BAD_PRACTICES = [
|
|
(r"\bprint\s*\(", "print() found — use click.echo() for user output"),
|
|
(r"\beval\s*\(", "eval() found — security risk, avoid dynamic code execution"),
|
|
(r"\bexec\s*\(", "exec() found — security risk, avoid dynamic code execution"),
|
|
(r"shell\s*=\s*True", "shell=True found — security risk, use shell=False with list args"),
|
|
(r"except\s*:", "bare except found — catch specific exceptions"),
|
|
(r"except\s+Exception\s*:", "broad Exception catch — catch specific exceptions"),
|
|
(r"#\s*(TODO|FIXME|HACK|XXX)", "TODO/FIXME found — resolve before merging"),
|
|
]
|
|
|
|
# Patterns for hardcoded config values that should be in config.py
|
|
HARDCODED_CONFIG = [
|
|
(r"https?://[a-z]+\.[a-z]+\.[a-z]+", "hardcoded URL — move to config.py with env var override"),
|
|
]
|
|
|
|
|
|
@dataclass
|
|
class ReviewResult:
|
|
"""Result of automated review checks."""
|
|
|
|
issues: list[dict[str, Any]] = field(default_factory=list)
|
|
summary: list[str] = field(default_factory=list)
|
|
|
|
@property
|
|
def has_issues(self) -> bool:
|
|
return bool(self.issues)
|
|
|
|
def add_issue(self, file_path: str, line: int, message: str, severity: str = "warning") -> None:
|
|
self.issues.append(
|
|
{
|
|
"path": file_path,
|
|
"body": f"[{severity}] {message}",
|
|
"new_position": line,
|
|
}
|
|
)
|
|
|
|
def add_summary(self, text: str) -> None:
|
|
self.summary.append(text)
|
|
|
|
|
|
def is_python_file(path: str) -> bool:
|
|
"""Check if a file is a Python source file."""
|
|
return path.endswith(PYTHON_SUFFIX) and not path.startswith("tests/")
|
|
|
|
|
|
def is_workflow_only(path: str) -> bool:
|
|
"""Check if a file is workflow/config/docs only (not Python source)."""
|
|
return path.endswith(WORKFLOW_ONLY_SUFFIXES) or path.startswith((".gitea/", "docs/", "ansible/"))
|
|
|
|
|
|
def check_architecture_compliance(files: list[dict[str, Any]], result: ReviewResult) -> None:
|
|
"""Check that changes follow the documented architecture."""
|
|
for f in files:
|
|
path = f.get("filename", "")
|
|
if not is_python_file(path):
|
|
continue
|
|
|
|
patch = f.get("patch", "")
|
|
if not patch:
|
|
continue
|
|
|
|
lines = patch.split("\n")
|
|
current_line = 0
|
|
for line in lines:
|
|
if line.startswith("@@"):
|
|
match = re.search(r"\+(\d+)", line)
|
|
if match:
|
|
current_line = int(match.group(1)) - 1
|
|
continue
|
|
if line.startswith("+") and not line.startswith("+++"):
|
|
current_line += 1
|
|
content = line[1:]
|
|
|
|
# Check for business logic in CLI
|
|
if path == CLI_FILE:
|
|
for pattern, msg in BUSINESS_LOGIC_IN_CLI:
|
|
if re.search(pattern, content):
|
|
result.add_issue(path, current_line, msg, "error")
|
|
|
|
if not result.issues:
|
|
result.add_summary("- Architecture compliance: OK")
|
|
|
|
|
|
def check_best_practices(files: list[dict[str, Any]], result: ReviewResult) -> None:
|
|
"""Check for common code quality issues."""
|
|
for f in files:
|
|
path = f.get("filename", "")
|
|
if not is_python_file(path):
|
|
continue
|
|
|
|
patch = f.get("patch", "")
|
|
if not patch:
|
|
continue
|
|
|
|
lines = patch.split("\n")
|
|
current_line = 0
|
|
for line in lines:
|
|
if line.startswith("@@"):
|
|
match = re.search(r"\+(\d+)", line)
|
|
if match:
|
|
current_line = int(match.group(1)) - 1
|
|
continue
|
|
if line.startswith("+") and not line.startswith("+++"):
|
|
current_line += 1
|
|
content = line[1:]
|
|
|
|
for pattern, msg in BAD_PRACTICES:
|
|
if re.search(pattern, content):
|
|
result.add_issue(path, current_line, msg, "warning")
|
|
|
|
if not any(i["body"].startswith("[warning]") for i in result.issues):
|
|
result.add_summary("- Best practices: OK")
|
|
|
|
|
|
def check_security(files: list[dict[str, Any]], result: ReviewResult) -> None:
|
|
"""Check for security issues in changed files."""
|
|
for f in files:
|
|
path = f.get("filename", "")
|
|
if not is_python_file(path):
|
|
continue
|
|
|
|
patch = f.get("patch", "")
|
|
if not patch:
|
|
continue
|
|
|
|
lines = patch.split("\n")
|
|
current_line = 0
|
|
for line in lines:
|
|
if line.startswith("@@"):
|
|
match = re.search(r"\+(\d+)", line)
|
|
if match:
|
|
current_line = int(match.group(1)) - 1
|
|
continue
|
|
if line.startswith("+") and not line.startswith("+++"):
|
|
current_line += 1
|
|
content = line[1:]
|
|
|
|
# Check for hardcoded secrets
|
|
secret_re = r'(token|password|secret|key)\s*=\s*["\'][^"\']{8,}["\']' # nosec B105
|
|
is_secret = re.search(secret_re, content, re.IGNORECASE)
|
|
is_comment = content.strip().startswith("#")
|
|
is_example = "your-" in content or "example" in content
|
|
if is_secret and not is_comment and not is_example:
|
|
result.add_issue(
|
|
path,
|
|
current_line,
|
|
"potential hardcoded secret — use environment variable",
|
|
"error",
|
|
)
|
|
|
|
if not any(i["body"].startswith("[error]") and "secret" in i["body"] for i in result.issues):
|
|
result.add_summary("- Security: OK")
|
|
|
|
|
|
def check_function_length(files: list[dict[str, Any]], result: ReviewResult) -> None:
|
|
"""Check that no new function is excessively long (> 50 lines)."""
|
|
for f in files:
|
|
path = f.get("filename", "")
|
|
if not is_python_file(path):
|
|
continue
|
|
|
|
patch = f.get("patch", "")
|
|
if not patch:
|
|
continue
|
|
|
|
# Count consecutive added lines within a function
|
|
lines = patch.split("\n")
|
|
current_line = 0
|
|
func_start = 0
|
|
func_name = ""
|
|
added_in_func = 0
|
|
|
|
for line in lines:
|
|
if line.startswith("@@"):
|
|
if func_name and added_in_func > 50:
|
|
result.add_issue(
|
|
path,
|
|
func_start,
|
|
f"function '{func_name}' adds {added_in_func} lines — consider splitting (> 50 lines)",
|
|
"warning",
|
|
)
|
|
match = re.search(r"\+(\d+)", line)
|
|
if match:
|
|
current_line = int(match.group(1)) - 1
|
|
func_name = ""
|
|
added_in_func = 0
|
|
continue
|
|
if line.startswith("+") and not line.startswith("+++"):
|
|
current_line += 1
|
|
content = line[1:]
|
|
func_match = re.match(r"\s*def\s+(\w+)\s*\(", content)
|
|
if func_match:
|
|
if func_name and added_in_func > 50:
|
|
result.add_issue(
|
|
path,
|
|
func_start,
|
|
f"function '{func_name}' adds {added_in_func} lines — consider splitting (> 50 lines)",
|
|
"warning",
|
|
)
|
|
func_name = func_match.group(1)
|
|
func_start = current_line
|
|
added_in_func = 0
|
|
else:
|
|
added_in_func += 1
|
|
elif line.startswith(" ") or line.startswith("-"):
|
|
pass # context or removed line
|
|
|
|
# Check last function
|
|
if func_name and added_in_func > 50:
|
|
result.add_issue(
|
|
path,
|
|
func_start,
|
|
f"function '{func_name}' adds {added_in_func} lines — consider splitting (> 50 lines)",
|
|
"warning",
|
|
)
|
|
|
|
|
|
def check_documentation(files: list[dict[str, Any]], result: ReviewResult) -> None:
|
|
"""Check that documentation is updated for relevant changes."""
|
|
has_src_changes = any(
|
|
is_python_file(f.get("filename", "")) and f.get("filename", "").startswith("src/") for f in files
|
|
)
|
|
has_doc_changes = any(
|
|
f.get("filename", "").startswith("docs/") or f.get("filename", "") in ("README.md", "AGENTS.md", "CHANGELOG.md")
|
|
for f in files
|
|
)
|
|
has_ansible_changes = any(f.get("filename", "").startswith("ansible/") for f in files)
|
|
|
|
if has_src_changes and not has_doc_changes:
|
|
result.add_summary("- Documentation: WARNING — source files changed but no docs updated")
|
|
elif has_ansible_changes and not has_doc_changes:
|
|
result.add_summary("- Documentation: WARNING — Ansible role changed but no docs updated")
|
|
else:
|
|
result.add_summary("- Documentation: OK")
|
|
|
|
|
|
def check_test_coverage(files: list[dict[str, Any]], result: ReviewResult) -> None:
|
|
"""Check that tests are updated for source changes."""
|
|
has_src_changes = any(
|
|
is_python_file(f.get("filename", "")) and f.get("filename", "").startswith("src/") for f in files
|
|
)
|
|
has_test_changes = any(f.get("filename", "").startswith("tests/") for f in files)
|
|
|
|
if has_src_changes and not has_test_changes:
|
|
result.add_summary("- Tests: WARNING — source files changed but no test files updated")
|
|
else:
|
|
result.add_summary("- Tests: OK")
|
|
|
|
|
|
def check_commit_conventions(client: GiteaClient, pr_number: str, result: ReviewResult) -> None:
|
|
"""Check that PR commits follow conventional commit format.
|
|
|
|
Verifies that at least one commit on the PR branch matches the
|
|
conventional commit pattern (type: description). Merge commits
|
|
and revert commits are exempt.
|
|
"""
|
|
try:
|
|
commits = client.get_pr_commits(pr_number)
|
|
except APIError as e:
|
|
result.add_summary(f"- Commit conventions: ERROR — could not fetch commits: {e.message}")
|
|
return
|
|
|
|
if not commits:
|
|
result.add_summary("- Commit conventions: OK (no commits to check)")
|
|
return
|
|
|
|
from gitea_runner_manager.config import CONVENTIONAL_RE
|
|
|
|
has_conventional = False
|
|
non_conventional: list[str] = []
|
|
|
|
for commit in commits:
|
|
commit_info = commit.get("commit", {})
|
|
message = str(commit_info.get("message", "") if isinstance(commit_info, dict) else "").split("\n")[0]
|
|
# Skip merge commits and revert commits
|
|
if message.startswith(("Merge", "Revert")):
|
|
continue
|
|
if CONVENTIONAL_RE.match(message):
|
|
has_conventional = True
|
|
else:
|
|
non_conventional.append(message[:60])
|
|
|
|
if has_conventional:
|
|
result.add_summary("- Commit conventions: OK")
|
|
elif non_conventional:
|
|
result.add_summary(
|
|
f"- Commit conventions: WARNING — no conventional commit found. "
|
|
f"Non-conventional commits: {', '.join(non_conventional[:3])}"
|
|
)
|
|
else:
|
|
result.add_summary("- Commit conventions: OK (all commits are merges/reverts)")
|
|
|
|
|
|
def run_review(client: GiteaClient, pr_number: str) -> ReviewResult:
|
|
"""Run all review checks and return the result."""
|
|
result = ReviewResult()
|
|
|
|
try:
|
|
files = client.get_pr_files(pr_number)
|
|
except APIError as e:
|
|
result.add_summary(f"- ERROR: Could not fetch PR files: {e.message}")
|
|
return result
|
|
|
|
if not files:
|
|
result.add_summary("- No files changed in this PR")
|
|
return result
|
|
|
|
# Run all checks
|
|
check_architecture_compliance(files, result)
|
|
check_best_practices(files, result)
|
|
check_security(files, result)
|
|
check_function_length(files, result)
|
|
check_documentation(files, result)
|
|
check_test_coverage(files, result)
|
|
check_commit_conventions(client, pr_number, result)
|
|
|
|
return result
|
|
|
|
|
|
def build_review_body(result: ReviewResult) -> str:
|
|
"""Build the review body text from the review result."""
|
|
lines = ["## Automated PR Review", ""]
|
|
|
|
for item in result.summary:
|
|
lines.append(item)
|
|
|
|
if result.issues:
|
|
lines.append("")
|
|
lines.append(f"**{len(result.issues)} issue(s) found:**")
|
|
lines.append("")
|
|
for issue in result.issues:
|
|
lines.append(f"- `{issue['path']}:{issue['new_position']}` — {issue['body']}")
|
|
else:
|
|
lines.append("")
|
|
lines.append("No issues found by automated checks.")
|
|
|
|
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("```")
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
def post_review(client: GiteaClient, pr_number: str, result: ReviewResult) -> dict[str, Any]:
|
|
"""Post the review to the PR.
|
|
|
|
Uses REQUEST_CHANGES when issues are found, COMMENT otherwise.
|
|
Never uses APPROVE — the bot shares the PR author's token, so
|
|
Gitea rejects self-approval. The actual APPROVE must come from
|
|
the manual review step.
|
|
"""
|
|
body = build_review_body(result)
|
|
event = "REQUEST_CHANGES" if result.has_issues else "COMMENT"
|
|
comments = result.issues if result.has_issues else []
|
|
|
|
return client.create_review(pr_number, event=event, body=body, comments=comments)
|
|
|
|
|
|
@click.command()
|
|
@click.argument("pr_number")
|
|
@click.argument("repo")
|
|
@click.option("--dry-run", is_flag=True, default=False, help="Print review without posting.")
|
|
def main(pr_number: str, repo: str, dry_run: bool) -> None:
|
|
"""Run automated PR review and post results to Gitea."""
|
|
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)
|
|
|
|
result = run_review(client, pr_number)
|
|
|
|
body = build_review_body(result)
|
|
event = "REQUEST_CHANGES" if result.has_issues else "COMMENT"
|
|
|
|
click.echo(f"Review event: {event}")
|
|
click.echo(f"Issues found: {len(result.issues)}")
|
|
click.echo("")
|
|
click.echo(body)
|
|
|
|
if dry_run:
|
|
click.echo("\n[dry-run] Review not posted.")
|
|
return
|
|
|
|
try:
|
|
review = post_review(client, pr_number, result)
|
|
except APIError as e:
|
|
if "approve" in e.message.lower() or "422" in str(e.status):
|
|
click.echo(_("Note: Self-approval not allowed. Posting COMMENT instead."))
|
|
review = client.create_review(pr_number, event="COMMENT", body=body)
|
|
else:
|
|
raise
|
|
review_id = review.get("id", "?")
|
|
click.echo(
|
|
_(
|
|
"\nReview #{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(result.issues),
|
|
)
|
|
)
|
|
|
|
|
|
if __name__ == "__main__": # pragma: no cover
|
|
main() # pragma: no cover
|