test: expand PR-queue-cleanup coverage to full acceptance matrix (#390)
Superset of the initial test slice: adds capability-map and router routing cases (reviewer-role enforcement, author wrong_role_stop), the full terminal-chain matrix (comment/skip stop, gates-failed stop, merge completed/blocker stop, unknown-decision fail-closed), and report verifier cases for two terminal mutations, missing next-suggested-PR, branch mutations, and missing workflow citation. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
Claude Opus 4.8
+205
-54
@@ -6,113 +6,264 @@ from pathlib import Path
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from pr_queue_cleanup import (
|
||||
CLEANUP_FORBIDDEN_TASKS,
|
||||
CLEANUP_WORKFLOW_PATH,
|
||||
STOP_AFTER_REQUEST_CHANGES,
|
||||
STOP_APPROVED_NO_MERGE_AUTH,
|
||||
CONTINUE_TO_SAME_PR_MERGE,
|
||||
STOP_AFTER_DECISION,
|
||||
STOP_AFTER_MERGE,
|
||||
STOP_AFTER_MERGE_BLOCKER,
|
||||
STOP_AFTER_REQUEST_CHANGES,
|
||||
STOP_APPROVED_MERGE_GATES_FAILED,
|
||||
STOP_APPROVED_NO_MERGE_AUTH,
|
||||
STOP_GATE_NOT_PROVEN,
|
||||
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
|
||||
from role_session_router import REVIEWER_TASKS, route_task_session
|
||||
from task_capability_map import required_permission, required_role
|
||||
|
||||
|
||||
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 TestCleanupCapabilityMapping(unittest.TestCase):
|
||||
def test_cleanup_task_is_reviewer_role(self):
|
||||
for key in ("pr_queue_cleanup", "pr-queue-cleanup"):
|
||||
self.assertEqual(required_permission(key), "gitea.pr.review")
|
||||
self.assertEqual(required_role(key), "reviewer")
|
||||
|
||||
def test_cleanup_task_registered_as_reviewer_route(self):
|
||||
self.assertIn("pr_queue_cleanup", REVIEWER_TASKS)
|
||||
|
||||
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,
|
||||
def test_author_session_gets_wrong_role_stop(self):
|
||||
result = route_task_session(
|
||||
"pr_queue_cleanup",
|
||||
active_profile="prgs-author",
|
||||
active_role_kind="author",
|
||||
allowed_in_current_session=False,
|
||||
)
|
||||
self.assertEqual(result["outcome"], CONTINUE_TO_SAME_PR_MERGE)
|
||||
self.assertTrue(result["further_mutation_allowed"])
|
||||
self.assertEqual(result["route_result"], "wrong_role_stop")
|
||||
self.assertFalse(result["downstream_allowed"])
|
||||
|
||||
def test_reviewer_session_allowed(self):
|
||||
result = route_task_session(
|
||||
"pr_queue_cleanup",
|
||||
active_profile="prgs-reviewer",
|
||||
active_role_kind="reviewer",
|
||||
allowed_in_current_session=True,
|
||||
)
|
||||
self.assertEqual(result["route_result"], "allowed_current_session")
|
||||
self.assertTrue(result["downstream_allowed"])
|
||||
|
||||
|
||||
class TestCleanupTaskGate(unittest.TestCase):
|
||||
def test_author_tasks_blocked(self):
|
||||
for task in (
|
||||
"claim_issue",
|
||||
"mark_issue",
|
||||
"lock_issue",
|
||||
"create_issue",
|
||||
"create_branch",
|
||||
"push_branch",
|
||||
"create_pr",
|
||||
"commit_files",
|
||||
):
|
||||
allowed, reasons = check_cleanup_task_allowed(task)
|
||||
self.assertFalse(allowed, task)
|
||||
self.assertTrue(reasons)
|
||||
self.assertTrue(reasons, task)
|
||||
|
||||
def test_review_task_allowed(self):
|
||||
allowed, reasons = check_cleanup_task_allowed("review_pr")
|
||||
self.assertTrue(allowed)
|
||||
self.assertEqual(reasons, [])
|
||||
|
||||
def test_forbidden_set_covers_issue_filing_and_impl(self):
|
||||
self.assertIn("create_issue", CLEANUP_FORBIDDEN_TASKS)
|
||||
self.assertIn("create_branch", CLEANUP_FORBIDDEN_TASKS)
|
||||
self.assertIn("commit_files", CLEANUP_FORBIDDEN_TASKS)
|
||||
|
||||
|
||||
class TestCleanupRunState(unittest.TestCase):
|
||||
def test_request_changes_stops_run(self):
|
||||
state = resolve_cleanup_run_state("request_changes")
|
||||
self.assertEqual(state["outcome"], STOP_AFTER_REQUEST_CHANGES)
|
||||
self.assertFalse(state["further_mutation_allowed"])
|
||||
|
||||
def test_comment_and_skip_stop_run(self):
|
||||
for decision in ("comment", "skip"):
|
||||
state = resolve_cleanup_run_state(decision)
|
||||
self.assertEqual(state["outcome"], STOP_AFTER_DECISION)
|
||||
self.assertFalse(state["further_mutation_allowed"])
|
||||
|
||||
def test_approved_without_merge_auth_stops(self):
|
||||
state = resolve_cleanup_run_state("approved", merge_gates_passed=True)
|
||||
self.assertEqual(state["outcome"], STOP_APPROVED_NO_MERGE_AUTH)
|
||||
self.assertFalse(state["further_mutation_allowed"])
|
||||
|
||||
def test_approved_with_merge_auth_continues_to_merge(self):
|
||||
state = resolve_cleanup_run_state(
|
||||
"approved",
|
||||
merge_authorized_for_pr=True,
|
||||
merge_gates_passed=True,
|
||||
)
|
||||
self.assertEqual(state["outcome"], CONTINUE_TO_SAME_PR_MERGE)
|
||||
self.assertTrue(state["further_mutation_allowed"])
|
||||
self.assertEqual(state["allowed_mutation"], "merge same PR only")
|
||||
|
||||
def test_approved_with_auth_but_failed_gates_stops(self):
|
||||
state = resolve_cleanup_run_state(
|
||||
"approved",
|
||||
merge_authorized_for_pr=True,
|
||||
merge_gates_passed=False,
|
||||
)
|
||||
self.assertEqual(state["outcome"], STOP_APPROVED_MERGE_GATES_FAILED)
|
||||
self.assertFalse(state["further_mutation_allowed"])
|
||||
|
||||
def test_merge_completed_stops(self):
|
||||
state = resolve_cleanup_run_state(
|
||||
"approved",
|
||||
merge_authorized_for_pr=True,
|
||||
merge_gates_passed=True,
|
||||
merge_completed=True,
|
||||
)
|
||||
self.assertEqual(state["outcome"], STOP_AFTER_MERGE)
|
||||
self.assertFalse(state["further_mutation_allowed"])
|
||||
|
||||
def test_merge_blocker_stops(self):
|
||||
state = resolve_cleanup_run_state(
|
||||
"approved",
|
||||
merge_authorized_for_pr=True,
|
||||
merge_gates_passed=True,
|
||||
merge_blocker=True,
|
||||
)
|
||||
self.assertEqual(state["outcome"], STOP_AFTER_MERGE_BLOCKER)
|
||||
self.assertFalse(state["further_mutation_allowed"])
|
||||
|
||||
def test_unknown_decision_fails_closed(self):
|
||||
state = resolve_cleanup_run_state("ship_it")
|
||||
self.assertEqual(state["outcome"], STOP_GATE_NOT_PROVEN)
|
||||
self.assertFalse(state["further_mutation_allowed"])
|
||||
|
||||
|
||||
def _clean_report(**overrides):
|
||||
lines = {
|
||||
"task": "Task: pr-queue-cleanup",
|
||||
"workflow": f"Workflow source: {CLEANUP_WORKFLOW_PATH} (hash abc123def456)",
|
||||
"pagination": (
|
||||
"PR inventory pagination proof: inventory_complete=true, final "
|
||||
"page, total_count 15"
|
||||
),
|
||||
"selected": "Selected PR: #371",
|
||||
"decision": "Review decision: approved",
|
||||
"merge_auth": "Merge authorized for PR: false",
|
||||
"merge_result": "Merge result: not attempted",
|
||||
"next": "Next suggested PR: #372",
|
||||
"issue_mut": "Issue mutations: none",
|
||||
"branch_mut": "Branch mutations: none",
|
||||
}
|
||||
lines.update(overrides)
|
||||
return "\n".join(v for v in lines.values() if v)
|
||||
|
||||
|
||||
class TestCleanupReportVerifier(unittest.TestCase):
|
||||
def test_good_report_passes(self):
|
||||
result = assess_pr_queue_cleanup_report(GOOD_REPORT)
|
||||
self.assertTrue(result["proven"])
|
||||
def test_clean_single_pr_report_passes(self):
|
||||
result = assess_pr_queue_cleanup_report(_clean_report())
|
||||
self.assertTrue(result["proven"], result["reasons"])
|
||||
self.assertFalse(result["block"])
|
||||
|
||||
def test_approved_and_merged_with_authorization_passes(self):
|
||||
report = _clean_report(
|
||||
merge_auth="Merge authorized for PR: true",
|
||||
merge_result="Merge result: merged 0123456789ab",
|
||||
)
|
||||
result = assess_pr_queue_cleanup_report(report)
|
||||
self.assertTrue(result["proven"], result["reasons"])
|
||||
|
||||
def test_request_changes_then_stop_passes(self):
|
||||
report = _clean_report(decision="Review decision: request_changes")
|
||||
result = assess_pr_queue_cleanup_report(report)
|
||||
self.assertTrue(result["proven"], result["reasons"])
|
||||
|
||||
def test_multi_pr_selection_blocked(self):
|
||||
report = GOOD_REPORT + "\nSelected PR: #403\n"
|
||||
report = _clean_report() + "\nSelected PR: #403\n"
|
||||
result = assess_pr_queue_cleanup_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertIn("multiple prs", result["reasons"][0].lower())
|
||||
self.assertTrue(
|
||||
any("multiple prs" in r.lower() for r in result["reasons"]),
|
||||
result["reasons"],
|
||||
)
|
||||
|
||||
def test_two_terminal_mutations_blocked(self):
|
||||
report = _clean_report() + "\nReview decision: request_changes"
|
||||
result = assess_pr_queue_cleanup_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(
|
||||
any("terminal review mutations" in r for r in result["reasons"]),
|
||||
result["reasons"],
|
||||
)
|
||||
|
||||
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",
|
||||
report = _clean_report(
|
||||
pagination="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"])
|
||||
any("pagination" in r.lower() for r in result["reasons"]),
|
||||
result["reasons"],
|
||||
)
|
||||
|
||||
def test_missing_next_suggested_pr_blocked(self):
|
||||
report = _clean_report(next="")
|
||||
result = assess_pr_queue_cleanup_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(
|
||||
any("Next suggested PR" in r for r in result["reasons"]),
|
||||
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")
|
||||
report = _clean_report(
|
||||
merge_result="Merge result: merged abc123abc123",
|
||||
)
|
||||
result = assess_pr_queue_cleanup_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(
|
||||
any("merge" in r.lower() for r in result["reasons"])
|
||||
any("authorization" in r.lower() for r in result["reasons"]),
|
||||
result["reasons"],
|
||||
)
|
||||
|
||||
def test_issue_mutations_forbidden(self):
|
||||
report = GOOD_REPORT.replace(
|
||||
"Issue mutations: none",
|
||||
"Issue mutations: gitea_mark_issue",
|
||||
)
|
||||
report = _clean_report(issue_mut="Issue mutations: gitea_mark_issue")
|
||||
result = assess_pr_queue_cleanup_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(
|
||||
any("issue mutations" in r.lower() for r in result["reasons"]),
|
||||
result["reasons"],
|
||||
)
|
||||
|
||||
def test_branch_mutations_forbidden(self):
|
||||
report = _clean_report(branch_mut="Branch mutations: created feat/x")
|
||||
result = assess_pr_queue_cleanup_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(
|
||||
any("branch mutations" in r.lower() for r in result["reasons"]),
|
||||
result["reasons"],
|
||||
)
|
||||
|
||||
def test_missing_workflow_citation_blocked(self):
|
||||
report = _clean_report(workflow="Workflow source: none")
|
||||
result = assess_pr_queue_cleanup_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(
|
||||
any("pr-queue-cleanup.md" in r for r in result["reasons"]),
|
||||
result["reasons"],
|
||||
)
|
||||
|
||||
def test_review_proofs_wrapper_matches_module(self):
|
||||
direct = assess_pr_queue_cleanup_report(GOOD_REPORT)
|
||||
wrapped = proofs_assess(GOOD_REPORT)
|
||||
direct = assess_pr_queue_cleanup_report(_clean_report())
|
||||
wrapped = proofs_assess(_clean_report())
|
||||
self.assertEqual(direct["proven"], wrapped["proven"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user