Gitea-Tools/tests/test_canonical_thread_handoff.py

"""Tests for Canonical Thread Handoff (CTH) protocol (#505)."""

from __future__ import annotations

import sys
import unittest
from pathlib import Path

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

import canonical_thread_handoff as cth  # noqa: E402

REPO_ROOT = Path(__file__).resolve().parent.parent
RUNBOOK = REPO_ROOT / "docs" / "llm-workflow-runbooks.md"
PROTOCOL_DOC = REPO_ROOT / "docs" / "canonical-thread-handoff.md"
TEMPLATE = REPO_ROOT / "skills" / "llm-project-workflow/templates" / "canonical-thread-handoff.md"
REVIEW_WORKFLOW = REPO_ROOT / "skills" / "llm-project-workflow/workflows" / "review-merge-pr.md"
WORK_WORKFLOW = REPO_ROOT / "skills" / "llm-project-workflow/workflows" / "work-issue.md"
REVIEW_TEMPLATE = REPO_ROOT / "skills" / "llm-project-workflow/templates" / "review-pr.md"
START_TEMPLATE = REPO_ROOT / "skills" / "llm-project-workflow/templates" / "start-issue.md"
MERGE_TEMPLATE = REPO_ROOT / "skills" / "llm-project-workflow/templates" / "merge-pr.md"


class TestCthModule(unittest.TestCase):
    def test_defines_cth_term_and_types(self):
        self.assertEqual(cth.CTH_ABBREV, "CTH")
        self.assertIn("State Handoff", cth.CTH_TYPES)
        self.assertIn("Reviewer Handoff", cth.CTH_TYPES)
        self.assertIn("Blocker", cth.CTH_TYPES)

    def test_format_and_parse_round_trip(self):
        body = cth.format_cth_body(
            cth_type="Author Handoff",
            status="implementation_complete",
            next_owner="reviewer",
            current_blocker="none",
            decision="PR opened",
            proof="pytest passed",
            next_action="review PR #9",
            ready_to_paste_prompt="Review PR #9 for issue #5",
        )
        parsed = cth.parse_cth_comment(body)
        self.assertIsNotNone(parsed)
        self.assertEqual(parsed["cth_type"], "Author Handoff")
        self.assertEqual(parsed["fields"]["status"], "implementation_complete")
        self.assertEqual(parsed["fields"]["next owner"], "reviewer")

    def test_valid_cth_passes_assessment(self):
        body = cth.format_cth_body(
            cth_type="Reviewer Handoff",
            status="approved",
            next_owner="merger",
            current_blocker="none",
            decision="APPROVE",
            proof="review id 42",
            next_action="merge if gates pass",
            ready_to_paste_prompt="Merge PR #12 with pinned head abcdef0123456789abcdef0123456789abcdef01",
        )
        result = cth.assess_cth_comment(body)
        self.assertFalse(result["block"])
        self.assertTrue(result["valid"])

    def test_missing_fields_fail_closed(self):
        body = "## CTH: Blocker\n\nStatus: blocked\n"
        result = cth.assess_cth_comment(body)
        self.assertTrue(result["block"])
        self.assertIn("missing required fields", result["reasons"][0])

    def test_find_latest_cth_prefers_newer_comment(self):
        older = cth.format_cth_body(
            cth_type="State Handoff",
            status="old",
            next_owner="author",
            current_blocker="none",
            decision="old",
            proof="old",
            next_action="old",
            ready_to_paste_prompt="old prompt text here",
        )
        newer = cth.format_cth_body(
            cth_type="Reviewer Handoff",
            status="new",
            next_owner="merger",
            current_blocker="none",
            decision="approve",
            proof="new",
            next_action="merge",
            ready_to_paste_prompt="merge PR #3 now with pinned head",
        )
        comments = [
            {"id": 1, "created_at": "2026-07-01T10:00:00Z", "body": older},
            {"id": 2, "created_at": "2026-07-02T10:00:00Z", "body": newer},
        ]
        latest = cth.find_latest_cth(comments)
        self.assertEqual(latest["comment_id"], 2)
        self.assertEqual(latest["cth_type"], "Reviewer Handoff")

    def test_cth_does_not_replace_formal_reviews_documented_in_protocol(self):
        text = PROTOCOL_DOC.read_text(encoding="utf-8")
        self.assertIn("do not replace", text.lower())
        self.assertIn("formal Gitea review", text)

    def test_examples_cover_required_scenarios(self):
        text = PROTOCOL_DOC.read_text(encoding="utf-8")
        expected_phrases = (
            "PR approved and ready for merger",
            "request-changes back to author",
            "superseded PR closure",
            "dirty root/worktree",
            "Stale head requiring fresh review",
            "Issue implementation handoff",
        )
        for phrase in expected_phrases:
            self.assertIn(phrase, text)


class TestCthDocumentationChecks(unittest.TestCase):
    def test_runbook_references_cth_discovery_and_posting(self):
        text = RUNBOOK.read_text(encoding="utf-8")
        self.assertIn("Canonical Thread Handoff", text)
        self.assertIn("Find the latest CTH comment", text)
        self.assertIn("Post a new CTH", text)
        self.assertIn("canonical-thread-handoff.md", text)

    def test_workflows_reference_cth_rules(self):
        for path in (REVIEW_WORKFLOW, WORK_WORKFLOW):
            text = path.read_text(encoding="utf-8")
            self.assertIn("latest CTH comment", text)
            self.assertIn("canonical-thread-handoff.md", text)

    def test_prompt_templates_reference_cth_rules(self):
        for path in (REVIEW_TEMPLATE, START_TEMPLATE, MERGE_TEMPLATE, TEMPLATE):
            text = path.read_text(encoding="utf-8")
            self.assertIn("CTH", text)
            self.assertIn("canonical-thread-handoff", text)


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