Merge pull request 'feat: add dedicated reconciliation controller-handoff schema (Closes #303)' (#373) from feat/issue-303-reconciliation-handoff into master

tests/test_reconciliation_handoff.py

@@ -0,0 +1,162 @@
+"""Tests for the dedicated reconciliation controller-handoff schema (#303).
+
+Already-landed PR reconciliation runs must emit the reconciliation
+schema, not stale author/reviewer handoff fields; observed git ref
+mutations (fetch) must be reported, and legacy fields fail validation.
+"""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from review_proofs import assess_reconciliation_handoff  # noqa: E402
+
+
+def _good_handoff(**overrides):
+    fields = {
+        "Task": "Reconcile already-landed open PRs",
+        "Repo": "prgs/Scaled-Tech-Consulting/Gitea-Tools",
+        "Role/profile": "prgs-reconciler",
+        "Identity": "jcwalker3",
+        "Selected PR": "278",
+        "PR live state": "open",
+        "Candidate head SHA": "2a544b7",
+        "Target branch": "master",
+        "Target branch SHA": "fa0cb12",
+        "Ancestor proof": "candidate head is ancestor of target branch SHA",
+        "Linked issue": "263",
+        "Linked issue live status": "open (fetched live this session)",
+        "Eligibility class": "ALREADY_LANDED_RECONCILE_REQUIRED",
+        "Capabilities proven": "gitea.pr.comment",
+        "Missing capabilities": "gitea.pr.close",
+        "PR comments posted": "1 (PR #278 reconciliation notice)",
+        "Issue comments posted": "none",
+        "PRs closed": "none",
+        "Issues closed": "none",
+        "Git ref mutations": "git fetch prgs master",
+        "Worktree mutations": "none",
+        "MCP/Gitea mutations": "PR comment on #278",
+        "External-state mutations": "none",
+        "Read-only diagnostics": "gitea_view_pr 278, gitea_view_issue 263",
+        "Blocker": "PR close capability missing",
+        "Safe next action": "operator closes PR #278 or grants close capability",
+        "No review/merge confirmation": "no review or merge mutations performed",
+    }
+    fields.update(overrides)
+    lines = ["Controller Handoff"]
+    lines += [f"- {k}: {v}" for k, v in fields.items() if v is not None]
+    return "\n".join(lines) + "\n"
+
+
+class TestReconciliationSchemaPasses(unittest.TestCase):
+    def test_blocked_reconciliation_passes(self):
+        result = assess_reconciliation_handoff(
+            _good_handoff(),
+            observed_commands=["git fetch prgs master"],
+        )
+        self.assertTrue(result["complete"])
+        self.assertEqual(result["reasons"], [])
+
+    def test_successful_pr_close_passes(self):
+        report = _good_handoff(**{
+            "Missing capabilities": "none",
+            "PRs closed": "PR #278",
+            "Blocker": "none",
+            "Safe next action": "none; reconciliation complete",
+        })
+        result = assess_reconciliation_handoff(
+            report, observed_commands=["git fetch prgs master"]
+        )
+        self.assertTrue(result["complete"])
+
+    def test_comment_only_reconciliation_passes(self):
+        report = _good_handoff(**{
+            "Git ref mutations": "none",
+        })
+        result = assess_reconciliation_handoff(report, observed_commands=[])
+        self.assertTrue(result["complete"])
+
+    def test_noop_already_closed_passes(self):
+        report = _good_handoff(**{
+            "PR live state": "closed",
+            "PR comments posted": "none",
+            "MCP/Gitea mutations": "none",
+            "Git ref mutations": "none",
+            "Blocker": "none",
+            "Safe next action": "none; PR already closed",
+        })
+        result = assess_reconciliation_handoff(report, observed_commands=[])
+        self.assertTrue(result["complete"])
+
+
+class TestSchemaViolationsFail(unittest.TestCase):
+    def test_missing_linked_issue_proof_fails(self):
+        report = _good_handoff(**{"Linked issue live status": None})
+        result = assess_reconciliation_handoff(report, observed_commands=[])
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("linked issue live status" in r.lower()
+                for r in result["reasons"])
+        )
+
+    def test_wrong_eligibility_class_fails(self):
+        report = _good_handoff(**{"Eligibility class": "REVIEW_ELIGIBLE"})
+        result = assess_reconciliation_handoff(report, observed_commands=[])
+        self.assertFalse(result["complete"])
+
+    def test_missing_handoff_section_fails(self):
+        result = assess_reconciliation_handoff(
+            "no handoff here", observed_commands=[]
+        )
+        self.assertFalse(result["complete"])
+
+
+class TestStaleFieldsFail(unittest.TestCase):
+    def test_pr_number_opened_fails(self):
+        report = _good_handoff() + "- PR number opened: 278\n"
+        result = assess_reconciliation_handoff(report, observed_commands=[])
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("pr number opened" in r.lower() for r in result["reasons"])
+        )
+
+    def test_pinned_reviewed_head_fails(self):
+        report = _good_handoff() + "- Pinned reviewed head: 2a544b7\n"
+        result = assess_reconciliation_handoff(report, observed_commands=[])
+        self.assertFalse(result["complete"])
+
+    def test_scratch_worktree_used_fails(self):
+        report = _good_handoff() + "- Scratch worktree used: False\n"
+        result = assess_reconciliation_handoff(report, observed_commands=[])
+        self.assertFalse(result["complete"])
+
+    def test_workspace_mutations_fails(self):
+        report = _good_handoff() + "- Workspace mutations: None\n"
+        result = assess_reconciliation_handoff(report, observed_commands=[])
+        self.assertFalse(result["complete"])
+
+    def test_author_lock_fields_fail(self):
+        report = (
+            _good_handoff()
+            + "- Issue lock proof: locked before diff\n"
+            + "- Claim/comment status: claimed\n"
+        )
+        result = assess_reconciliation_handoff(report, observed_commands=[])
+        self.assertFalse(result["complete"])
+
+
+class TestObservedFetchMustBeReported(unittest.TestCase):
+    def test_fetch_with_ref_mutations_none_fails(self):
+        report = _good_handoff(**{"Git ref mutations": "none"})
+        result = assess_reconciliation_handoff(
+            report, observed_commands=["git fetch prgs master"]
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("git ref mutation" in r.lower() for r in result["reasons"])
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()