diff --git a/tests/test_pr_queue_cleanup.py b/tests/test_pr_queue_cleanup.py index 3cec75a..e26a026 100644 --- a/tests/test_pr_queue_cleanup.py +++ b/tests/test_pr_queue_cleanup.py @@ -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() \ No newline at end of file + unittest.main()