Gitea-Tools/tests/test_thread_state_ledger_validator.py

"""Tests for two-comment workflow validation (#507)."""
import sys
import unittest
from pathlib import Path

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

from thread_state_ledger_validator import (  # noqa: E402
    BLOCKER_CLASSIFICATIONS,
    assess_controller_handoff_comment,
    assess_thread_state_ledger_comment,
    assess_two_comment_workflow,
    assess_two_comment_pair,
)


def _minimal_handoff(**overrides):
    lines = [
        "[CONTROLLER HANDOFF] PR #203 / Issue #182 — review complete",
        "",
        "Identity/profile:",
        "- Active profile: prgs-reviewer",
        "- Authenticated identity: sysadmin",
        "",
        "Server-side mutation ledger:",
        "- gitea_submit_pr_review → APPROVED review posted to Gitea",
        "",
        "Blockers:",
        "- none",
    ]
    text = "\n".join(lines)
    for key, value in overrides.items():
        if f"{key}:" in text:
            text = text.replace(f"{key}:", f"{key}: {value}", 1)
    return text


def _minimal_ledger(**overrides):
    lines = [
        "[THREAD STATE LEDGER] PR #203 — APPROVED review posted to Gitea",
        "",
        "What is true now:",
        "- PR state: open",
        "- Current head SHA: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
        "- Server-side decision state: APPROVED review posted to Gitea",
        "- Local verdict/state: APPROVE verdict prepared locally",
        "- Latest known validation: tests passed locally",
        "",
        "What changed:",
        "- APPROVED review posted to Gitea at pinned head",
        "",
        "What is blocked:",
        "- Blocker classification: no blocker",
        "",
        "Who/what acts next:",
        "- Next actor: merger",
        "- Required action: merge on explicit operator command",
        "- Do not do: re-post APPROVE",
        "- Resume from: PR #203 review feedback",
    ]
    text = "\n".join(lines)
    for key, value in overrides.items():
        text = text.replace(f"- {key}:", f"- {key}: {value}")
    return text


class TestTwoCommentWorkflow(unittest.TestCase):
    def test_valid_combined_report_passes(self):
        report = _minimal_handoff() + "\n\n" + _minimal_ledger()
        result = assess_two_comment_workflow(report)
        self.assertTrue(result["proven"])
        self.assertFalse(result["block"])

    def test_missing_ledger_after_tagged_handoff_blocks(self):
        result = assess_two_comment_workflow(_minimal_handoff())
        self.assertTrue(result["block"])
        self.assertTrue(
            any("THREAD STATE LEDGER" in r for r in result["reasons"])
        )

    def test_legacy_controller_handoff_section_not_required_ledger(self):
        legacy = "## Controller Handoff\n\n- Task: work issue\n- Next: continue\n"
        result = assess_two_comment_workflow(legacy)
        self.assertFalse(result["block"])

    def test_ambiguous_approved_blocks(self):
        handoff = _minimal_handoff().replace(
            "APPROVED review posted to Gitea",
            "approved",
        )
        report = handoff + "\n\n" + _minimal_ledger()
        result = assess_two_comment_workflow(report)
        self.assertTrue(result["block"])
        self.assertTrue(any("approved" in r.lower() for r in result["reasons"]))

    def test_ambiguous_merged_blocks(self):
        handoff = _minimal_handoff() + "\n\n- Narrative: merged successfully"
        ledger = _minimal_ledger()
        report = handoff + "\n\n" + ledger
        result = assess_two_comment_workflow(report)
        self.assertTrue(result["block"])
        self.assertTrue(any("merge" in r.lower() for r in result["reasons"]))

    def test_blocked_without_gate_blocks(self):
        handoff = _minimal_handoff().replace("- none", "- blocked")
        ledger = _minimal_ledger().replace(
            "no blocker", "blocked but no gate named"
        )
        report = handoff + "\n\n" + ledger
        result = assess_two_comment_workflow(report)
        self.assertTrue(result["block"])

    def test_mutation_ledger_contradiction_blocks(self):
        handoff = _minimal_handoff().replace(
            "APPROVED review posted to Gitea",
            "none — no server-side state changed",
        )
        handoff += "\n- Narrative: APPROVED review posted to Gitea"
        report = handoff + "\n\n" + _minimal_ledger()
        result = assess_two_comment_workflow(report)
        self.assertTrue(result["block"])
        self.assertTrue(any("contradict" in r.lower() for r in result["reasons"]))

    def test_local_verdict_as_server_side_blocks(self):
        ledger = _minimal_ledger(
            **{
                "Server-side decision state": "APPROVE verdict prepared locally",
                "Local verdict/state": "none",
            }
        )
        report = _minimal_handoff() + "\n\n" + ledger
        result = assess_two_comment_workflow(report)
        self.assertTrue(result["block"])


class TestHandoffComment(unittest.TestCase):
    def test_handoff_comment_valid(self):
        result = assess_controller_handoff_comment(_minimal_handoff())
        self.assertTrue(result["proven"])

    def test_handoff_without_marker_skips(self):
        result = assess_controller_handoff_comment("casual comment")
        self.assertTrue(result["proven"])
        self.assertFalse(result["performed"])


class TestLedgerComment(unittest.TestCase):
    def test_ledger_requires_blocker_classification(self):
        ledger = _minimal_ledger().replace(
            "Blocker classification: no blocker",
            "something unclear",
        )
        result = assess_thread_state_ledger_comment(ledger)
        self.assertTrue(result["block"])

    def test_all_blocker_classifications_recognized(self):
        self.assertIn("no blocker", BLOCKER_CLASSIFICATIONS)
        self.assertIn("environment/tooling blocker", BLOCKER_CLASSIFICATIONS)


class TestPairedComments(unittest.TestCase):
    def test_pair_passes(self):
        result = assess_two_comment_pair(_minimal_handoff(), _minimal_ledger())
        self.assertTrue(result["proven"])

    def test_pair_missing_ledger_blocks(self):
        result = assess_two_comment_pair(_minimal_handoff(), "")
        self.assertTrue(result["block"])


class TestExamples(unittest.TestCase):
    """Smoke-test bundled examples from docs."""

    def test_examples_module_loads(self):
        from thread_state_ledger_examples import EXAMPLES  # noqa: E402

        self.assertGreaterEqual(len(EXAMPLES), 8)
        for name, handoff, ledger in EXAMPLES:
            result = assess_two_comment_pair(handoff, ledger)
            self.assertTrue(
                result["proven"],
                f"example '{name}' failed: {result['reasons']}",
            )


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