303 lines
11 KiB
Python
303 lines
11 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_FORBIDDEN_TASKS,
|
|
CLEANUP_WORKFLOW_PATH,
|
|
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
|
|
|
|
|
|
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)
|
|
|
|
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["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, 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_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 = _clean_report() + "\nSelected PR: #403\n"
|
|
result = assess_pr_queue_cleanup_report(report)
|
|
self.assertFalse(result["proven"])
|
|
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 = _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"]),
|
|
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 = _clean_report(
|
|
merge_result="Merge result: merged abc123abc123",
|
|
)
|
|
result = assess_pr_queue_cleanup_report(report)
|
|
self.assertFalse(result["proven"])
|
|
self.assertTrue(
|
|
any("authorization" in r.lower() for r in result["reasons"]),
|
|
result["reasons"],
|
|
)
|
|
|
|
def test_issue_mutations_forbidden(self):
|
|
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(_clean_report())
|
|
wrapped = proofs_assess(_clean_report())
|
|
self.assertEqual(direct["proven"], wrapped["proven"])
|
|
|
|
|
|
class TestCleanupEmptyQueue(unittest.TestCase):
|
|
def test_empty_queue_report_passes(self):
|
|
report = _clean_report(
|
|
selected="Selected PR: none",
|
|
decision="Review decision: comment",
|
|
next="Next suggested PR: approvals (queue empty)",
|
|
pagination="PR inventory pagination proof: inventory_complete=true, total_count 0",
|
|
)
|
|
report += "\npr_inventory_trust_gate.status: trusted_empty"
|
|
result = assess_pr_queue_cleanup_report(report)
|
|
self.assertTrue(result["proven"], result["reasons"])
|
|
|
|
def test_empty_queue_without_evidence_fails(self):
|
|
report = _clean_report(
|
|
selected="Selected PR: none",
|
|
)
|
|
result = assess_pr_queue_cleanup_report(report)
|
|
self.assertFalse(result["proven"])
|
|
self.assertTrue(
|
|
any("no valid empty-queue claim" in r for r in result["reasons"]),
|
|
result["reasons"]
|
|
)
|
|
|
|
def test_contradictory_selected_pr_fails(self):
|
|
report = _clean_report() + "\nSelected PR: none"
|
|
result = assess_pr_queue_cleanup_report(report)
|
|
self.assertFalse(result["proven"])
|
|
self.assertTrue(
|
|
any("contains both a selected PR and a claim of none selected" in r for r in result["reasons"]),
|
|
result["reasons"]
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|