Gitea-Tools/tests/test_review_workflow_load.py

"""Tests for canonical review workflow load proof (#389)."""

import os
import sys
import unittest
from unittest.mock import patch

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

import review_workflow_load
import mcp_server


class TestReviewWorkflowLoadModule(unittest.TestCase):
    def setUp(self):
        review_workflow_load.clear_review_workflow_load()
        mcp_server._save_review_decision_lock(None)

    def test_load_records_hash_and_schema(self):
        root = str(__import__("pathlib").Path(__file__).resolve().parent.parent)
        recorded = review_workflow_load.record_review_workflow_load(root)
        self.assertEqual(
            recorded["workflow_source"],
            review_workflow_load.WORKFLOW_REL_PATH,
        )
        self.assertEqual(recorded["task_mode"], "review-merge-pr")
        self.assertRegex(recorded["workflow_hash"], r"^[0-9a-f]{12}$")
        self.assertEqual(
            recorded["final_report_schema_path"],
            review_workflow_load.SCHEMA_REL_PATH,
        )
        status = review_workflow_load.workflow_load_status(root)
        self.assertTrue(status["workflow_load_proof_present"])
        self.assertTrue(status["workflow_load_valid"])

    def test_stale_session_pid_blocks(self):
        root = str(__import__("pathlib").Path(__file__).resolve().parent.parent)
        review_workflow_load.record_review_workflow_load(root)
        review_workflow_load._REVIEW_WORKFLOW_LOAD["session_pid"] = 0
        blockers = review_workflow_load.review_workflow_load_blockers(root)
        self.assertTrue(any("different process" in b for b in blockers))

    def test_prompt_conflict_detected(self):
        conflict, reasons = review_workflow_load.assess_prompt_conflict(
            "Run work-issue author implementation only")
        self.assertTrue(conflict)
        self.assertTrue(reasons)


class TestReviewWorkflowLoadGates(unittest.TestCase):
    def setUp(self):
        review_workflow_load.clear_review_workflow_load()
        mcp_server.init_review_decision_lock("prgs", "review_pr")
        mcp_server.record_preflight_check("whoami")
        mcp_server.record_preflight_check("capability", "reviewer")

    def _load_workflow(self):
        return mcp_server.gitea_load_review_workflow()

    def test_mcp_helper_returns_required_fields(self):
        res = self._load_workflow()
        self.assertTrue(res["success"])
        self.assertTrue(res["loaded"])
        self.assertIn("workflow_source", res)
        self.assertIn("workflow_hash", res)
        self.assertIn("final_report_schema_path", res)
        self.assertIn("final_report_schema_hash", res)

    def test_mark_final_blocked_without_load(self):
        res = mcp_server.gitea_mark_final_review_decision(
            42, "approve", remote="prgs")
        self.assertFalse(res["marked_ready"])
        self.assertTrue(any(
            "gitea_load_review_workflow" in r for r in res["reasons"]))
        self.assertTrue(any(
            "approve/merge replay" in r.lower() or "Do not call" in r
            for r in res["reasons"]))

    def test_submit_review_blocked_without_load(self):
        with patch("mcp_server.gitea_check_pr_eligibility") as elig:
            elig.return_value = {
                "eligible": True,
                "authenticated_user": "rev",
                "profile_name": "prgs-reviewer",
                "pr_author": "author",
                "head_sha": "abc123",
                "reasons": [],
            }
            res = mcp_server.gitea_submit_pr_review(
                42,
                "approve",
                remote="prgs",
                final_review_decision_ready=True,
            )
        self.assertFalse(res["performed"])
        self.assertTrue(any(
            "gitea_load_review_workflow" in r for r in res["reasons"]))

    def test_merge_blocked_without_load(self):
        res = mcp_server.gitea_merge_pr(
            42,
            confirmation="MERGE PR 42",
            remote="prgs",
        )
        self.assertFalse(res["performed"])
        self.assertTrue(any(
            "gitea_load_review_workflow" in r for r in res["reasons"]))

    def test_resolve_capability_reports_missing_load(self):
        with patch.object(mcp_server, "_ensure_matching_profile"):
            with patch.object(
                    mcp_server.gitea_config, "is_runtime_switching_enabled",
                    return_value=False):
                with patch.object(
                        mcp_server, "_authenticated_username",
                        return_value="rev"):
                    res = mcp_server.gitea_resolve_task_capability(
                        "review_pr", remote="prgs")
        proof = res.get("workflow_load_proof") or {}
        self.assertFalse(proof.get("workflow_load_valid"))
        self.assertTrue(any(
            "gitea_load_review_workflow" in g
            for g in res.get("task_role_guidance") or []))

    def test_init_review_lock_clears_prior_load(self):
        self._load_workflow()
        mcp_server.init_review_decision_lock("prgs", "review_pr")
        blockers = review_workflow_load.review_workflow_load_blockers(
            str(__import__("pathlib").Path(__file__).resolve().parent.parent))
        self.assertTrue(blockers)

    def test_dry_run_allowed_without_load(self):
        with patch("mcp_server.gitea_check_pr_eligibility") as elig:
            elig.return_value = {
                "eligible": True,
                "authenticated_user": "rev",
                "profile_name": "prgs-reviewer",
                "pr_author": "author",
                "head_sha": "abc123",
                "reasons": [],
            }
            res = mcp_server.gitea_dry_run_pr_review(
                42, "approve", remote="prgs")
        self.assertNotIn(
            "gitea_load_review_workflow",
            " ".join(res.get("reasons") or []),
        )


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