Merge pull request 'feat: classify already-landed PRs as reconciliation-only (Closes #295)' (#382) from feat/issue-295-already-landed-classification into master

tests/test_final_report_validator.py

@@ -194,6 +194,20 @@ class TestReviewPrRules(unittest.TestCase):
             )
         )
 
+    def test_already_landed_oldest_eligible_pr_blocks(self):
+        report = (
+            _review_handoff()
+            + "\nAlready landed.\nOldest eligible PR: PR #278."
+        )
+        result = assess_final_report_validator(report, "review_pr")
+        self.assertTrue(result["blocked"])
+        self.assertTrue(
+            any(
+                f["rule_id"] == "reviewer.already_landed_eligible_wording"
+                for f in result["findings"]
+            )
+        )
+
     def test_already_landed_gate_blocks_review_terminal_states(self):
         cases = {
             "APPROVED": ("- Review decision: request_changes", "- Review decision: APPROVED"),
@@ -242,7 +256,6 @@ class TestReviewPrRules(unittest.TestCase):
         )
         result = assess_final_report_validator(report, "review_pr")
         self.assertEqual(result["grade"], "A")
-
     def test_git_fetch_must_not_be_readonly_only(self):
         report = (
             _review_handoff()

tests/test_llm_workflow_split.py

@@ -243,6 +243,12 @@ def test_worktree_ownership_verifier_exported():
     assert callable(assess_worktree_ownership_report)
 
 
+def test_already_landed_classification_verifier_exported():
+    from review_proofs import assess_already_landed_classification_report
+
+    assert callable(assess_already_landed_classification_report)
+
+
 def test_pr_queue_cleanup_verifier_exported():
     from review_proofs import assess_pr_queue_cleanup_report
 

tests/test_reviewer_already_landed_classification.py

@@ -0,0 +1,81 @@
+import unittest
+
+from reviewer_already_landed_classification import (  # noqa: E402
+    ALREADY_LANDED_ELIGIBILITY_CLASS,
+    assess_already_landed_classification_report,
+)
+
+
+class TestAlreadyLandedClassification(unittest.TestCase):
+    def test_non_landed_report_passes(self):
+        result = assess_already_landed_classification_report(
+            "Selected the next eligible PR: PR #286."
+        )
+        self.assertTrue(result["proven"])
+        self.assertFalse(result["landed_context"])
+
+    def test_oldest_eligible_pr_on_landed_blocks(self):
+        result = assess_already_landed_classification_report(
+            "PR is already landed.\n"
+            "Oldest eligible PR: PR #278."
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["block"])
+
+    def test_next_eligible_pr_on_landed_blocks(self):
+        result = assess_already_landed_classification_report(
+            "Ancestor proof: passed\n"
+            "Selected the next eligible PR: PR #278."
+        )
+        self.assertFalse(result["proven"])
+
+    def test_eligible_for_review_on_landed_blocks(self):
+        result = assess_already_landed_classification_report(
+            "Already landed on master and eligible for review next."
+        )
+        self.assertFalse(result["proven"])
+
+    def test_proper_landed_wording_passes(self):
+        result = assess_already_landed_classification_report(
+            "Oldest open PR requiring action: PR #278\n"
+            f"Eligibility class: {ALREADY_LANDED_ELIGIBILITY_CLASS}\n"
+            "Candidate head SHA: abc123\n"
+            "Reviewed head SHA: none\n"
+            "Queue blocked by already-landed PR requiring reconciliation",
+            selected_pr_already_landed=True,
+        )
+        self.assertTrue(result["proven"])
+
+    def test_pinned_reviewed_head_without_validation_blocks(self):
+        result = assess_already_landed_classification_report(
+            "Already landed.\n"
+            "Pinned reviewed head: abc123def456",
+            selected_pr_already_landed=True,
+        )
+        self.assertFalse(result["proven"])
+
+    def test_wrong_eligibility_class_blocks(self):
+        result = assess_already_landed_classification_report(
+            "Already landed.\n"
+            "Eligibility class: REVIEW_ELIGIBLE",
+            selected_pr_already_landed=True,
+        )
+        self.assertFalse(result["proven"])
+
+    def test_missing_queue_blocker_blocks(self):
+        result = assess_already_landed_classification_report(
+            f"Eligibility class: {ALREADY_LANDED_ELIGIBILITY_CLASS}\n"
+            "Candidate head SHA: abc123\n"
+            "Reviewed head SHA: none",
+            selected_pr_already_landed=True,
+        )
+        self.assertFalse(result["proven"])
+
+    def test_exported_from_review_proofs(self):
+        from review_proofs import assess_already_landed_classification_report as exported
+
+        self.assertTrue(callable(exported))
+
+
+if __name__ == "__main__":
+    unittest.main()