Gitea-Tools/tests/test_reconciliation_handoff.py

"""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()