Merge pull request 'feat: require canonical next-action state comments (Closes #495)' (#499) from feat/issue-495-canonical-next-action-comments into master

tests/test_canonical_state_comments.py

@@ -0,0 +1,143 @@
+"""Tests for canonical state comment validation (#495)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from canonical_state_comments import (  # noqa: E402
+    validate_canonical_state_comment,
+    validate_final_report_state_update,
+)
+
+
+GOOD_ISSUE_STATE = """## Canonical Issue State
+
+STATE:
+ready-for-author
+
+WHO_IS_NEXT:
+author
+
+NEXT_ACTION:
+Implement the focused validator and open a PR.
+
+NEXT_PROMPT:
+Find issue #495, load the work-issue workflow, implement the validator, and open a PR.
+
+WHAT_HAPPENED:
+Controller selected the canonical tracking issue.
+
+WHY:
+The duplicate issue points here as canonical.
+
+RELATED_PRS:
+none
+
+BRANCH:
+none
+
+HEAD_SHA:
+none
+
+VALIDATION:
+none
+
+BLOCKERS:
+none
+"""
+
+
+class TestCanonicalStateComments(unittest.TestCase):
+    def test_good_issue_state_validates(self):
+        result = validate_canonical_state_comment(GOOD_ISSUE_STATE)
+        self.assertTrue(result["valid"])
+        self.assertEqual(result["reasons"], [])
+
+    def test_missing_next_actor_is_rejected(self):
+        result = validate_canonical_state_comment(
+            GOOD_ISSUE_STATE.replace("WHO_IS_NEXT:\nauthor", "WHO_IS_NEXT:\n")
+        )
+        self.assertFalse(result["valid"])
+        self.assertIn("WHO_IS_NEXT", " ".join(result["reasons"]))
+
+    def test_missing_next_prompt_is_rejected(self):
+        result = validate_canonical_state_comment(
+            GOOD_ISSUE_STATE.replace(
+                "NEXT_PROMPT:\nFind issue #495, load the work-issue workflow, implement the validator, and open a PR.",
+                "NEXT_PROMPT:\n",
+            )
+        )
+        self.assertFalse(result["valid"])
+        self.assertIn("NEXT_PROMPT", " ".join(result["reasons"]))
+
+    def test_vague_next_action_is_rejected(self):
+        result = validate_canonical_state_comment(
+            GOOD_ISSUE_STATE.replace(
+                "NEXT_ACTION:\nImplement the focused validator and open a PR.",
+                "NEXT_ACTION:\ncontinue",
+            )
+        )
+        self.assertFalse(result["valid"])
+        self.assertIn("too vague", " ".join(result["reasons"]))
+
+    def test_ready_to_merge_requires_approval_and_head_sha(self):
+        text = """## Canonical PR State
+
+STATE:
+ready-to-merge
+
+WHO_IS_NEXT:
+merger
+
+NEXT_ACTION:
+Merge PR #12 after checking the current head.
+
+NEXT_PROMPT:
+Load the merge workflow and merge PR #12 if gates pass.
+
+ISSUE:
+#11
+
+HEAD_SHA:
+abc123
+
+REVIEW_STATUS:
+none
+
+MERGE_READY:
+yes
+"""
+        result = validate_canonical_state_comment(text)
+        self.assertFalse(result["valid"])
+        reasons = " ".join(result["reasons"])
+        self.assertIn("approved REVIEW_STATUS", reasons)
+        self.assertIn("full HEAD_SHA", reasons)
+
+    def test_superseded_requires_canonical_item(self):
+        text = GOOD_ISSUE_STATE.replace("STATE:\nready-for-author", "STATE:\nsuperseded")
+        result = validate_canonical_state_comment(text)
+        self.assertFalse(result["valid"])
+        self.assertIn("canonical item", " ".join(result["reasons"]))
+
+    def test_blocked_requires_unblock_condition(self):
+        text = GOOD_ISSUE_STATE.replace("STATE:\nready-for-author", "STATE:\nblocked")
+        text = text.replace("BLOCKERS:\nnone", "BLOCKERS:\n")
+        result = validate_canonical_state_comment(text)
+        self.assertFalse(result["valid"])
+        self.assertIn("BLOCKERS", " ".join(result["reasons"]))
+
+    def test_final_report_claim_without_block_is_rejected(self):
+        report = "Posted canonical state comment on issue #495."
+        result = validate_final_report_state_update(report)
+        self.assertTrue(result["applicable"])
+        self.assertFalse(result["valid"])
+
+    def test_final_report_without_state_claim_not_applicable(self):
+        result = validate_final_report_state_update("ordinary final report")
+        self.assertFalse(result["applicable"])
+        self.assertTrue(result["valid"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_final_report_validator.py

@@ -351,6 +351,92 @@ class TestReconciliationRules(unittest.TestCase):
         )
 
 
+class TestCanonicalStateUpdateRules(unittest.TestCase):
+    def test_claimed_state_comment_without_block_blocks(self):
+        report = _reconcile_handoff() + "\nPosted canonical state comment on PR #99."
+        result = assess_final_report_validator(report, "reconcile_already_landed")
+        self.assertTrue(result["blocked"])
+        self.assertTrue(
+            any(f["rule_id"] == "shared.canonical_state_update" for f in result["findings"])
+        )
+
+    def test_state_comment_missing_next_prompt_blocks(self):
+        report = (
+            _reconcile_handoff()
+            + """
+
+## Canonical PR State
+
+STATE:
+needs-review
+
+WHO_IS_NEXT:
+reviewer
+
+NEXT_ACTION:
+Review PR #99.
+
+NEXT_PROMPT:
+
+ISSUE:
+#98
+
+HEAD_SHA:
+0fdc8f582026b72a229d59a172c0a63ac4aaeaf9
+
+REVIEW_STATUS:
+none
+
+MERGE_READY:
+no
+"""
+        )
+        result = assess_final_report_validator(report, "reconcile_already_landed")
+        self.assertTrue(result["blocked"])
+        reasons = " ".join(f["reason"] for f in result["findings"])
+        self.assertIn("NEXT_PROMPT", reasons)
+
+    def test_valid_state_comment_claim_passes_shared_rule(self):
+        report = (
+            _reconcile_handoff()
+            + """
+
+## Canonical PR State
+
+STATE:
+needs-review
+
+WHO_IS_NEXT:
+reviewer
+
+NEXT_ACTION:
+Review PR #99 against the pinned head.
+
+NEXT_PROMPT:
+Load the review workflow and review PR #99.
+
+ISSUE:
+#98
+
+HEAD_SHA:
+0fdc8f582026b72a229d59a172c0a63ac4aaeaf9
+
+REVIEW_STATUS:
+none
+
+MERGE_READY:
+no
+
+BLOCKERS:
+none
+"""
+        )
+        result = assess_final_report_validator(report, "reconcile_already_landed")
+        self.assertFalse(
+            any(f["rule_id"] == "shared.canonical_state_update" for f in result["findings"])
+        )
+
+
 class TestCanonicalReconcileSchema(unittest.TestCase):
     """Issue #307: reconciliation workflows emit only the canonical schema."""