Gitea-Tools/tests/test_reviewer_handoff_consistency.py

"""Tests for reviewer handoff consistency validation (#501)."""
import sys
import unittest
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

import reviewer_handoff_consistency  # noqa: E402
from final_report_validator import assess_final_report_validator  # noqa: E402


def _base_handoff(**overrides):
    lines = [
        "## Controller Handoff",
        "",
        "- Task: review PR #472",
        "- Review decision: request_changes",
        "- Review mutations: submitted request_changes review",
        "- Merge mutations: none",
        "- MCP/Gitea mutations: reviewer lease acquired",
    ]
    text = "\n".join(lines)
    for key, value in overrides.items():
        token = f"- {key}:"
        if token in text:
            before, _, after = text.partition(token)
            rest = after.split("\n", 1)
            suffix = rest[1] if len(rest) > 1 else ""
            text = f"{before}{token} {value}" + (f"\n{suffix}" if suffix else "")
        else:
            text += f"\n- {key}: {value}"
    return text


class TestReviewerHandoffConsistency(unittest.TestCase):
    def test_consistent_handoff_passes(self):
        result = reviewer_handoff_consistency.assess_reviewer_handoff_consistency(
            _base_handoff()
        )
        self.assertTrue(result["proven"], result["reasons"])

    def test_claimed_merge_missing_from_ledger(self):
        report = _base_handoff(
            **{
                "Current status": "Merged PR #411 earlier; reviewing PR #472",
                "Merge mutations": "none",
                "MCP/Gitea mutations": "reviewer lease acquired",
            }
        )
        result = reviewer_handoff_consistency.assess_reviewer_handoff_consistency(report)
        self.assertFalse(result["proven"])
        self.assertTrue(any("merge" in r.lower() for r in result["reasons"]))

    def test_terminal_budget_consumed_without_merge_in_ledger(self):
        report = _base_handoff(
            **{
                "Current status": "Terminal mutation budget already consumed by merge",
                "Merge mutations": "none",
                "MCP/Gitea mutations": "reviewer lease acquired",
            }
        )
        result = reviewer_handoff_consistency.assess_reviewer_handoff_consistency(report)
        self.assertFalse(result["proven"])
        self.assertTrue(any("terminal mutation budget" in r for r in result["reasons"]))

    def test_review_blocked_but_final_decision_marked(self):
        report = _base_handoff(
            **{
                "Current status": "gitea_submit_pr_review blocked",
                "Review mutations": "none",
            }
        )
        report += "\nServer-side final decision marked via gitea_mark_final_review_decision."
        result = reviewer_handoff_consistency.assess_reviewer_handoff_consistency(report)
        self.assertFalse(result["proven"])
        self.assertTrue(any("blocked" in r and "final decision" in r for r in result["reasons"]))

    def test_lease_without_review_decision(self):
        report = _base_handoff(**{"Review decision": "none"})
        report += "\nReviewer lease acquired for PR #472."
        result = reviewer_handoff_consistency.assess_reviewer_handoff_consistency(report)
        self.assertFalse(result["proven"])
        self.assertTrue(any("Review decision" in r for r in result["reasons"]))

    def test_rejected_mutation_without_proof(self):
        report = _base_handoff(
            **{
                "Review mutations": "none",
                "MCP/Gitea mutations": "none",
            }
        )
        report += "\nReview submission blocked before mutation."
        result = reviewer_handoff_consistency.assess_reviewer_handoff_consistency(report)
        self.assertFalse(result["proven"])
        self.assertTrue(any("proof fields" in r for r in result["reasons"]))

    def test_blocked_template_includes_required_fields(self):
        template = reviewer_handoff_consistency.render_blocked_review_handoff_template()
        for field in reviewer_handoff_consistency._BLOCKED_REVIEW_PROOF_FIELDS:
            self.assertIn(field, template.lower())

    def test_final_report_validator_integration(self):
        report = _base_handoff(
            **{
                "Current status": "Merged PR #411; terminal mutation budget consumed",
                "Merge mutations": "none",
            }
        )
        result = assess_final_report_validator(report, "review_pr")
        self.assertTrue(
            any(
                f["rule_id"] == "reviewer.handoff_consistency"
                for f in result["findings"]
            ),
            result["findings"],
        )


if __name__ == "__main__":
    unittest.main()