sysadminandClaude Opus 4.8
91c67930ec
Adds canonical pr-queue-cleanup workflow, report verifier, reviewer-only task routing, and runbook guidance so operators can drain open PRs one canonical review at a time with fail-closed boundaries on batching and unauthorized merges. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
118 lines
4.0 KiB
Python
118 lines
4.0 KiB
Python
"""Tests for PR-only queue cleanup mode (#390)."""
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
from pr_queue_cleanup import (
|
|
CLEANUP_WORKFLOW_PATH,
|
|
STOP_AFTER_REQUEST_CHANGES,
|
|
STOP_APPROVED_NO_MERGE_AUTH,
|
|
CONTINUE_TO_SAME_PR_MERGE,
|
|
assess_pr_queue_cleanup_report,
|
|
check_cleanup_task_allowed,
|
|
resolve_cleanup_run_state,
|
|
)
|
|
from review_proofs import assess_pr_queue_cleanup_report as proofs_assess
|
|
|
|
|
|
GOOD_REPORT = f"""
|
|
Task: pr-queue-cleanup
|
|
Workflow source: {CLEANUP_WORKFLOW_PATH}
|
|
PR inventory pagination proof: inventory_complete=true, total_count=12
|
|
Selected PR: #401
|
|
Review decision: approved
|
|
Merge authorized for PR: true
|
|
Merge result: not attempted
|
|
Next suggested PR: #402
|
|
Issue mutations: none
|
|
Branch mutations: none
|
|
"""
|
|
|
|
|
|
class TestCleanupRunState(unittest.TestCase):
|
|
def test_request_changes_stops_run(self):
|
|
result = resolve_cleanup_run_state("request_changes")
|
|
self.assertEqual(result["outcome"], STOP_AFTER_REQUEST_CHANGES)
|
|
self.assertFalse(result["further_mutation_allowed"])
|
|
|
|
def test_approved_without_merge_auth_stops(self):
|
|
result = resolve_cleanup_run_state("approved")
|
|
self.assertEqual(result["outcome"], STOP_APPROVED_NO_MERGE_AUTH)
|
|
self.assertFalse(result["further_mutation_allowed"])
|
|
|
|
def test_approved_with_merge_auth_continues_to_merge(self):
|
|
result = resolve_cleanup_run_state(
|
|
"approved",
|
|
merge_authorized_for_pr=True,
|
|
merge_gates_passed=True,
|
|
)
|
|
self.assertEqual(result["outcome"], CONTINUE_TO_SAME_PR_MERGE)
|
|
self.assertTrue(result["further_mutation_allowed"])
|
|
|
|
|
|
class TestCleanupTaskGate(unittest.TestCase):
|
|
def test_author_tasks_blocked(self):
|
|
for task in (
|
|
"lock_issue",
|
|
"create_issue",
|
|
"create_branch",
|
|
"create_pr",
|
|
"commit_files",
|
|
):
|
|
allowed, reasons = check_cleanup_task_allowed(task)
|
|
self.assertFalse(allowed, task)
|
|
self.assertTrue(reasons)
|
|
|
|
|
|
class TestCleanupReportVerifier(unittest.TestCase):
|
|
def test_good_report_passes(self):
|
|
result = assess_pr_queue_cleanup_report(GOOD_REPORT)
|
|
self.assertTrue(result["proven"])
|
|
self.assertFalse(result["block"])
|
|
|
|
def test_multi_pr_selection_blocked(self):
|
|
report = GOOD_REPORT + "\nSelected PR: #403\n"
|
|
result = assess_pr_queue_cleanup_report(report)
|
|
self.assertFalse(result["proven"])
|
|
self.assertIn("multiple prs", result["reasons"][0].lower())
|
|
|
|
def test_missing_pagination_proof_blocked(self):
|
|
report = GOOD_REPORT.replace(
|
|
"PR inventory pagination proof: inventory_complete=true, total_count=12",
|
|
"PR inventory pagination proof: inventory listed",
|
|
)
|
|
result = assess_pr_queue_cleanup_report(report)
|
|
self.assertFalse(result["proven"])
|
|
self.assertTrue(
|
|
any("pagination" in r.lower() for r in result["reasons"])
|
|
)
|
|
|
|
def test_merge_without_authorization_blocked(self):
|
|
report = GOOD_REPORT.replace(
|
|
"Merge authorized for PR: true",
|
|
"Merge authorized for PR: false",
|
|
).replace("Merge result: not attempted", "Merge result: merged abc123")
|
|
result = assess_pr_queue_cleanup_report(report)
|
|
self.assertFalse(result["proven"])
|
|
self.assertTrue(
|
|
any("merge" in r.lower() for r in result["reasons"])
|
|
)
|
|
|
|
def test_issue_mutations_forbidden(self):
|
|
report = GOOD_REPORT.replace(
|
|
"Issue mutations: none",
|
|
"Issue mutations: gitea_mark_issue",
|
|
)
|
|
result = assess_pr_queue_cleanup_report(report)
|
|
self.assertFalse(result["proven"])
|
|
|
|
def test_review_proofs_wrapper_matches_module(self):
|
|
direct = assess_pr_queue_cleanup_report(GOOD_REPORT)
|
|
wrapped = proofs_assess(GOOD_REPORT)
|
|
self.assertEqual(direct["proven"], wrapped["proven"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main() |