Gitea-Tools/tests/test_review_workflow_boundary.py

"""Tests for workflow-load session boundary tracking (#403)."""

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

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

import final_report_validator
import review_workflow_boundary
import review_workflow_load
import mcp_server


class TestPreReviewClassification(unittest.TestCase):
    def setUp(self):
        review_workflow_boundary.clear_pre_review_commands()
        review_workflow_load.clear_review_workflow_load()

    def test_inventory_command_allowed(self):
        result = review_workflow_boundary.classify_pre_review_command(
            "gitea_list_prs remote=prgs",
            cwd="/tmp",
            project_root="/repo/Gitea-Tools",
        )
        self.assertEqual(
            result["classification"],
            review_workflow_boundary.CLASSIFICATION_READ_ONLY_INVENTORY,
        )

    def test_main_checkout_pytest_is_boundary_violation(self):
        root = "/repo/Gitea-Tools"
        result = review_workflow_boundary.classify_pre_review_command(
            "python -m pytest tests/",
            cwd=root,
            project_root=root,
        )
        self.assertEqual(
            result["classification"],
            review_workflow_boundary.CLASSIFICATION_BOUNDARY_VIOLATION,
        )

    def test_profiles_json_inspection_is_boundary_violation(self):
        result = review_workflow_boundary.classify_pre_review_command(
            "cat profiles.json",
            cwd="/repo/Gitea-Tools",
            project_root="/repo/Gitea-Tools",
        )
        self.assertEqual(
            result["classification"],
            review_workflow_boundary.CLASSIFICATION_BOUNDARY_VIOLATION,
        )


class TestWorkflowLoadBoundaryGate(unittest.TestCase):
    def setUp(self):
        review_workflow_boundary.clear_pre_review_commands()
        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 _root(self) -> str:
        return str(__import__("pathlib").Path(__file__).resolve().parent.parent)

    def test_boundary_violation_blocks_mutation_after_load(self):
        root = self._root()
        review_workflow_boundary.record_pre_review_command(
            "python -m pytest tests/",
            cwd=root,
            project_root=root,
        )
        res = mcp_server.gitea_load_review_workflow()
        self.assertFalse(res["success"])
        self.assertEqual(res["boundary_status"], "violation")
        blocked = mcp_server.gitea_mark_final_review_decision(
            42, "approve", remote="prgs")
        self.assertFalse(blocked["marked_ready"])
        joined = " ".join(blocked["reasons"]).lower()
        self.assertTrue(
            "validation" in joined or "boundary" in joined or "workflow" in joined
        )

    def test_clean_inventory_then_load_passes(self):
        root = self._root()
        review_workflow_boundary.record_pre_review_command(
            "gitea_list_prs remote=prgs",
            cwd=root,
            project_root=root,
        )
        res = mcp_server.gitea_load_review_workflow()
        self.assertTrue(res["success"])
        self.assertEqual(res["boundary_status"], "clean")
        blockers = review_workflow_load.review_workflow_load_blockers(root)
        self.assertEqual(blockers, [])

    def test_file_view_narrative_fails_validator_without_helper(self):
        report = (
            "## Controller Handoff\n"
            "- Task: review-merge-pr\n"
            "- I read the canonical workflow review-merge-pr.md before review.\n"
        )
        findings = final_report_validator.assess_final_report_validator(
            report,
            task_kind="review_pr",
        )
        self.assertTrue(any(
            f["rule_id"] == "reviewer.workflow_load_boundary"
            for f in findings.get("findings") or []
        ))

    def test_helper_result_passes_validator(self):
        root = self._root()
        review_workflow_load.record_review_workflow_load(root)
        helper = review_workflow_boundary.workflow_load_helper_result(
            review_workflow_load._REVIEW_WORKFLOW_LOAD,
            root,
        )
        report = (
            "## Controller Handoff\n"
            "- Task: review-merge-pr\n"
            f"- Workflow-load helper result: workflow_hash: {helper['workflow_hash']}; "
            f"boundary_status: {helper['boundary_status']}\n"
        )
        findings = final_report_validator.assess_final_report_validator(
            report,
            task_kind="review_pr",
        )
        boundary_findings = [
            f for f in (findings.get("findings") or [])
            if f.get("rule_id") == "reviewer.workflow_load_boundary"
        ]
        self.assertEqual(boundary_findings, [])


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