Compare commits
1 Commits
fix/1445-a
...
fix/1615
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
6cfa167956 |
@@ -1,7 +1,14 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Review Gate — Poka-yoke for unreviewed merges.
|
||||
Fails if the current PR has fewer than 1 approving review.
|
||||
Enhanced to prevent rubber-stamping of PRs with no changes.
|
||||
|
||||
Issue #1615: feat: prevent rubber-stamping of PRs with no changes (#1445)
|
||||
|
||||
Checks:
|
||||
1. Empty PR - 0 additions, 0 deletions, 0 files
|
||||
2. Approval - No APPROVED reviews
|
||||
3. Rubber-stamp - Trivial PR + empty review comment
|
||||
|
||||
Usage in Gitea workflow:
|
||||
- name: Review Approval Gate
|
||||
@@ -13,16 +20,21 @@ Usage in Gitea workflow:
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
import subprocess
|
||||
from urllib import request, error
|
||||
|
||||
# Configuration
|
||||
GITEA_TOKEN = os.environ.get("GITEA_TOKEN", "")
|
||||
GITEA_URL = os.environ.get("GITEA_URL", "https://forge.alexanderwhitestone.com")
|
||||
REPO = os.environ.get("GITEA_REPO", "")
|
||||
PR_NUMBER = os.environ.get("PR_NUMBER", "")
|
||||
|
||||
# Thresholds
|
||||
MIN_APPROVALS = 1
|
||||
TRIVIAL_THRESHOLD = 10 # Lines changed
|
||||
|
||||
|
||||
def api_call(method, path):
|
||||
"""Make authenticated Gitea API call."""
|
||||
url = f"{GITEA_URL}/api/v1{path}"
|
||||
headers = {"Authorization": f"token {GITEA_TOKEN}"}
|
||||
req = request.Request(url, method=method, headers=headers)
|
||||
@@ -33,38 +45,188 @@ def api_call(method, path):
|
||||
return {"error": e.read().decode(), "status": e.code}
|
||||
|
||||
|
||||
def get_pr_details(repo, pr_number):
|
||||
"""Get PR details including diff stats."""
|
||||
return api_call("GET", f"/repos/{repo}/pulls/{pr_number}")
|
||||
|
||||
|
||||
def get_pr_files(repo, pr_number):
|
||||
"""Get list of files changed in PR."""
|
||||
return api_call("GET", f"/repos/{repo}/pulls/{pr_number}/files")
|
||||
|
||||
|
||||
def get_pr_reviews(repo, pr_number):
|
||||
"""Get reviews for a PR."""
|
||||
return api_call("GET", f"/repos/{repo}/pulls/{pr_number}/reviews")
|
||||
|
||||
|
||||
def check_empty_pr(pr_details, pr_files):
|
||||
"""Check if PR has no actual changes."""
|
||||
additions = pr_details.get("additions", 0)
|
||||
deletions = pr_details.get("deletions", 0)
|
||||
changed_files = pr_details.get("changed_files", 0)
|
||||
|
||||
if additions == 0 and deletions == 0 and changed_files == 0:
|
||||
return {
|
||||
"check": "empty_pr",
|
||||
"passed": False,
|
||||
"message": f"EMPTY PR: {additions} additions, {deletions} deletions, {changed_files} files",
|
||||
"severity": "error"
|
||||
}
|
||||
|
||||
return {
|
||||
"check": "empty_pr",
|
||||
"passed": True,
|
||||
"message": f"PR has changes: {additions} additions, {deletions} deletions, {changed_files} files",
|
||||
"severity": "info"
|
||||
}
|
||||
|
||||
|
||||
def check_approvals(reviews):
|
||||
"""Check if PR has required approvals."""
|
||||
approvals = [r for r in reviews if r.get("state") == "APPROVED"]
|
||||
|
||||
if len(approvals) >= MIN_APPROVALS:
|
||||
return {
|
||||
"check": "approvals",
|
||||
"passed": True,
|
||||
"message": f"PR has {len(approvals)} approving review(s)",
|
||||
"severity": "info"
|
||||
}
|
||||
else:
|
||||
return {
|
||||
"check": "approvals",
|
||||
"passed": False,
|
||||
"message": f"PR has {len(approvals)} approving review(s), need {MIN_APPROVALS}",
|
||||
"severity": "error"
|
||||
}
|
||||
|
||||
|
||||
def check_rubber_stamp(pr_details, reviews):
|
||||
"""Check for rubber-stamping (trivial PR + empty review comment)."""
|
||||
additions = pr_details.get("additions", 0)
|
||||
deletions = pr_details.get("deletions", 0)
|
||||
total_changes = additions + deletions
|
||||
|
||||
# Check if PR is trivial (small changes)
|
||||
is_trivial = total_changes <= TRIVIAL_THRESHOLD
|
||||
|
||||
# Check if any approval has empty comment
|
||||
rubber_stamp = False
|
||||
for review in reviews:
|
||||
if review.get("state") == "APPROVED":
|
||||
body = review.get("body", "").strip()
|
||||
if not body:
|
||||
rubber_stamp = True
|
||||
break
|
||||
|
||||
if is_trivial and rubber_stamp:
|
||||
return {
|
||||
"check": "rubber_stamp",
|
||||
"passed": False,
|
||||
"message": f"RUBBER-STAMP WARNING: Trivial PR ({total_changes} lines) with empty approval comment",
|
||||
"severity": "warning"
|
||||
}
|
||||
|
||||
return {
|
||||
"check": "rubber_stamp",
|
||||
"passed": True,
|
||||
"message": "No rubber-stamping detected",
|
||||
"severity": "info"
|
||||
}
|
||||
|
||||
|
||||
def main():
|
||||
"""Main entry point."""
|
||||
if not GITEA_TOKEN:
|
||||
print("ERROR: GITEA_TOKEN not set")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
if not REPO:
|
||||
print("ERROR: GITEA_REPO not set")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
pr_number = PR_NUMBER
|
||||
if not pr_number:
|
||||
# Try to infer from Gitea Actions environment
|
||||
pr_number = os.environ.get("GITEA_PULL_REQUEST_INDEX", "")
|
||||
|
||||
|
||||
if not pr_number:
|
||||
print("ERROR: Could not determine PR number")
|
||||
sys.exit(1)
|
||||
|
||||
reviews = api_call("GET", f"/repos/{REPO}/pulls/{pr_number}/reviews")
|
||||
|
||||
print(f"Review Gate: Checking PR #{pr_number} in {REPO}")
|
||||
print("=" * 60)
|
||||
|
||||
# Get PR details
|
||||
pr_details = get_pr_details(REPO, pr_number)
|
||||
if isinstance(pr_details, dict) and "error" in pr_details:
|
||||
print(f"ERROR fetching PR details: {pr_details}")
|
||||
sys.exit(1)
|
||||
|
||||
# Get PR files
|
||||
pr_files = get_pr_files(REPO, pr_number)
|
||||
if isinstance(pr_files, dict) and "error" in pr_files:
|
||||
print(f"ERROR fetching PR files: {pr_files}")
|
||||
sys.exit(1)
|
||||
|
||||
# Get PR reviews
|
||||
reviews = get_pr_reviews(REPO, pr_number)
|
||||
if isinstance(reviews, dict) and "error" in reviews:
|
||||
print(f"ERROR fetching reviews: {reviews}")
|
||||
sys.exit(1)
|
||||
|
||||
approvals = [r for r in reviews if r.get("state") == "APPROVED"]
|
||||
if len(approvals) >= 1:
|
||||
print(f"OK: PR #{pr_number} has {len(approvals)} approving review(s).")
|
||||
|
||||
# Run checks
|
||||
checks = []
|
||||
|
||||
# Check 1: Empty PR
|
||||
empty_check = check_empty_pr(pr_details, pr_files)
|
||||
checks.append(empty_check)
|
||||
|
||||
# Check 2: Approvals
|
||||
approval_check = check_approvals(reviews)
|
||||
checks.append(approval_check)
|
||||
|
||||
# Check 3: Rubber-stamping
|
||||
rubber_check = check_rubber_stamp(pr_details, reviews)
|
||||
checks.append(rubber_check)
|
||||
|
||||
# Print results
|
||||
print("\nCheck Results:")
|
||||
print("-" * 60)
|
||||
|
||||
errors = 0
|
||||
warnings = 0
|
||||
|
||||
for check in checks:
|
||||
status = "✅ PASS" if check["passed"] else "❌ FAIL"
|
||||
if check["severity"] == "warning" and check["passed"]:
|
||||
status = "⚠️ WARN"
|
||||
|
||||
print(f"{status} [{check['check']}] {check['message']}")
|
||||
|
||||
if not check["passed"]:
|
||||
if check["severity"] == "error":
|
||||
errors += 1
|
||||
elif check["severity"] == "warning":
|
||||
warnings += 1
|
||||
|
||||
print("-" * 60)
|
||||
|
||||
# Final decision
|
||||
if errors > 0:
|
||||
print(f"\n❌ BLOCKED: {errors} error(s), {warnings} warning(s)")
|
||||
print("Merges are not permitted until errors are resolved.")
|
||||
sys.exit(1)
|
||||
elif warnings > 0:
|
||||
print(f"\n⚠️ WARNING: {warnings} warning(s)")
|
||||
print("PR can merge but review warnings above.")
|
||||
sys.exit(0)
|
||||
else:
|
||||
print(f"BLOCKED: PR #{pr_number} has no approving reviews.")
|
||||
print("Merges are not permitted without at least one approval.")
|
||||
sys.exit(1)
|
||||
print(f"\n✅ OK: All checks passed")
|
||||
print("PR is ready for merge.")
|
||||
sys.exit(0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
main()
|
||||
Reference in New Issue
Block a user