feat: validate Selected PR: none in cleanup mode when queue is empty #576
@@ -132,6 +132,13 @@ def check_cleanup_task_allowed(task: str) -> tuple[bool, list[str]]:
|
|||||||
_SELECTED_PR_RE = re.compile(
|
_SELECTED_PR_RE = re.compile(
|
||||||
r"^\s*[-*]?\s*selected pr\s*:\s*#?(\d+)", re.IGNORECASE | re.MULTILINE
|
r"^\s*[-*]?\s*selected pr\s*:\s*#?(\d+)", re.IGNORECASE | re.MULTILINE
|
||||||
)
|
)
|
||||||
|
_SELECTED_PR_NONE_RE = re.compile(
|
||||||
|
r"^\s*[-*]?\s*selected pr\s*:\s*(?:none|n/a)\b", re.IGNORECASE | re.MULTILINE
|
||||||
|
)
|
||||||
|
_EMPTY_QUEUE_RE = re.compile(
|
||||||
|
r"\b0 open pr|\bno open pr|\bno eligible pr|\bempty (?:review )?queue|\binventory empty|\btrusted_empty\b",
|
||||||
|
re.IGNORECASE,
|
||||||
|
)
|
||||||
_NEXT_SUGGESTED_RE = re.compile(
|
_NEXT_SUGGESTED_RE = re.compile(
|
||||||
r"^\s*[-*]?\s*next suggested pr\s*:", re.IGNORECASE | re.MULTILINE
|
r"^\s*[-*]?\s*next suggested pr\s*:", re.IGNORECASE | re.MULTILINE
|
||||||
)
|
)
|
||||||
@@ -176,6 +183,10 @@ def assess_pr_queue_cleanup_report(report_text: str) -> dict:
|
|||||||
|
|
||||||
selected = _SELECTED_PR_RE.findall(text)
|
selected = _SELECTED_PR_RE.findall(text)
|
||||||
if len(selected) == 0:
|
if len(selected) == 0:
|
||||||
|
if _SELECTED_PR_NONE_RE.search(text):
|
||||||
|
if not _EMPTY_QUEUE_RE.search(text):
|
||||||
|
reasons.append("Selected PR is 'none' but no valid empty-queue claim found")
|
||||||
|
else:
|
||||||
reasons.append("report must name exactly one Selected PR")
|
reasons.append("report must name exactly one Selected PR")
|
||||||
elif len(set(selected)) > 1:
|
elif len(set(selected)) > 1:
|
||||||
reasons.append(
|
reasons.append(
|
||||||
@@ -184,6 +195,9 @@ def assess_pr_queue_cleanup_report(report_text: str) -> dict:
|
|||||||
"PR per run"
|
"PR per run"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
if len(selected) > 0 and _SELECTED_PR_NONE_RE.search(text):
|
||||||
|
reasons.append("report contains both a selected PR and a claim of none selected")
|
||||||
|
|
||||||
terminal = _TERMINAL_MUTATION_RE.findall(text)
|
terminal = _TERMINAL_MUTATION_RE.findall(text)
|
||||||
if len(terminal) > 1:
|
if len(terminal) > 1:
|
||||||
reasons.append(
|
reasons.append(
|
||||||
|
|||||||
@@ -265,5 +265,38 @@ class TestCleanupReportVerifier(unittest.TestCase):
|
|||||||
self.assertEqual(direct["proven"], wrapped["proven"])
|
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__":
|
if __name__ == "__main__":
|
||||||
unittest.main()
|
unittest.main()
|
||||||
|
|||||||
Reference in New Issue
Block a user