Gitea-Tools/tests/test_capability_stop_terminal.py

"""Tests for capability stop terminal mode (#197)."""
import json
import os
import sys
import tempfile
import unittest
from unittest.mock import patch

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

import capability_stop_terminal
import gitea_config
import mcp_server
from review_proofs import assess_capability_stop_terminal_report

CONFIG = {
    "version": 2,
    "contexts": {
        "ctx": {
            "enabled": True,
            "gitea": {"enabled": True, "base_url": "https://gitea.example.com"},
        }
    },
    "profiles": {
        "prgs-author": {
            "enabled": True,
            "context": "ctx",
            "role": "author",
            "username": "jcwalker3",
            "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
            "allowed_operations": [
                "gitea.read", "gitea.issue.create", "gitea.issue.comment",
                "gitea.pr.create", "gitea.branch.push",
            ],
            "forbidden_operations": [
                "gitea.pr.approve", "gitea.pr.merge", "gitea.pr.review",
            ],
            "execution_profile": "prgs-author",
        },
    },
    "rules": {"allow_runtime_switching": False},
}


class TestCapabilityStopTerminal(unittest.TestCase):
    def setUp(self):
        capability_stop_terminal.clear()
        self._remotes = patch.dict(mcp_server.REMOTES, {
            "prgs": {
                "host": "gitea.example.com",
                "org": "Scaled-Tech-Consulting",
                "repo": "Gitea-Tools",
            },
        })
        self._remotes.start()
        mcp_server._IDENTITY_CACHE.clear()
        gitea_config._active_profile_override = None
        self._dir = tempfile.TemporaryDirectory()
        self.config_path = os.path.join(self._dir.name, "profiles.json")
        with open(self.config_path, "w", encoding="utf-8") as fh:
            fh.write(json.dumps(CONFIG))

    def tearDown(self):
        self._remotes.stop()
        capability_stop_terminal.clear()
        mcp_server._IDENTITY_CACHE.clear()
        gitea_config._active_profile_override = None
        self._dir.cleanup()

    def _env(self):
        return {
            "GITEA_MCP_CONFIG": self.config_path,
            "GITEA_MCP_PROFILE": "prgs-author",
            "GITEA_TOKEN_AUTHOR": "author-pass",
        }

    @patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
    @patch("mcp_server.get_auth_header", return_value="token author-pass")
    def test_review_pr_stop_enters_terminal_mode(self, _auth, _api):
        with patch.dict(os.environ, self._env()):
            res = mcp_server.gitea_resolve_task_capability(
                task="review_pr", remote="prgs"
            )
        self.assertTrue(res["stop_required"])
        self.assertTrue(res.get("terminal_mode"))
        self.assertTrue(capability_stop_terminal.is_active())
        self.assertIn(
            capability_stop_terminal.TERMINAL_REPORT_HEADING,
            res["terminal_report"]["heading"],
        )

    @patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
    @patch("mcp_server.get_auth_header", return_value="token author-pass")
    def test_list_prs_blocked_after_capability_stop(self, _auth, _api):
        with patch.dict(os.environ, self._env()):
            mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
            with self.assertRaises(RuntimeError) as ctx:
                mcp_server.gitea_list_prs(remote="prgs")
        self.assertIn("Cannot perform reviewer task", str(ctx.exception))
        self.assertIn("review_pr", str(ctx.exception))

    @patch("mcp_server.api_fetch_page")
    @patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
    @patch("mcp_server.get_auth_header", return_value="token author-pass")
    def test_list_prs_allowed_after_author_task_clears_stale_denial(
        self, _auth, _api, mock_fetch,
    ):
        mock_fetch.return_value = ([], {
            "page": 1, "per_page": 50, "returned_count": 0,
            "has_more": False, "next_page": None, "is_final_page": True,
        })
        with patch.dict(os.environ, self._env()):
            denied = mcp_server.gitea_resolve_task_capability(
                task="review_pr", remote="prgs"
            )
            self.assertTrue(denied["stop_required"])
            self.assertTrue(capability_stop_terminal.is_active())

            cleared = mcp_server.gitea_resolve_task_capability(
                task="claim_issue", remote="prgs"
            )
            self.assertTrue(cleared["allowed_in_current_session"])
            self.assertTrue(cleared.get("cleared_stale_denial"))
            self.assertFalse(capability_stop_terminal.is_active())

            prs = mcp_server.gitea_list_prs(remote="prgs")
        self.assertEqual(prs["prs"], [])

    @patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
    @patch("mcp_server.get_auth_header", return_value="token author-pass")
    def test_eligibility_check_blocked_after_stop(self, _auth, _api):
        with patch.dict(os.environ, self._env()):
            mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
            res = mcp_server.gitea_check_pr_eligibility(
                pr_number=193, action="review", remote="prgs"
            )
        self.assertFalse(res["eligible"])
        self.assertTrue(res.get("terminal_mode"))

    def test_report_with_pr_selection_impure(self):
        report = (
            "Cannot perform reviewer task under current profile. "
            "No reviewer mutations performed.\n"
            "Selected PR #193 for review anyway."
        )
        result = assess_capability_stop_terminal_report(report)
        self.assertFalse(result["pure"])

    def test_session_eligibility_wording_blocked(self):
        ok, violations = capability_stop_terminal.validate_eligibility_wording(
            "PR 193 is not authored by this session so it is eligible."
        )
        self.assertFalse(ok)
        self.assertTrue(violations)

    def test_rebase_fallback_blocked_in_report(self):
        report = (
            "Cannot perform reviewer task under current profile. "
            "No reviewer mutations performed.\n"
            "Or have me rebase conflicted PR 193."
        )
        result = assess_capability_stop_terminal_report(report)
        self.assertFalse(result["pure"])
        self.assertTrue(
            any("author fallback" in v for v in result["violations"])
        )

    def test_empty_queue_without_trusted_empty_blocked(self):
        report = (
            "Cannot perform reviewer task under current profile. "
            "No reviewer mutations performed.\n"
            "The repo has 0 open PRs."
        )
        result = assess_capability_stop_terminal_report(
            report, trust_gate_status="untrusted_empty"
        )
        self.assertFalse(result["pure"])

    def test_empty_queue_with_parsed_trusted_status_passes_gate_check(self):
        report = (
            "Cannot perform reviewer task under current profile. "
            "No reviewer mutations performed.\n"
            "Repository: Scaled-Tech-Consulting/Gitea-Tools\n"
            "pr_inventory_trust_gate.status: trusted_empty\n"
            "pr_inventory_trust_gate.corroborated: true\n"
            "Inventory profile: prgs-reviewer\n"
            "No open PRs in queue."
        )
        result = assess_capability_stop_terminal_report(report)
        self.assertTrue(result["pure"])

    def test_pure_terminal_report_passes(self):
        report = (
            "Cannot perform reviewer task under current profile. "
            "No reviewer mutations performed.\n"
            "Identity: jcwalker3 / prgs-author.\n"
            "Required: prgs-reviewer.\n"
            "No mutations performed."
        )
        result = assess_capability_stop_terminal_report(report)
        self.assertTrue(result["pure"])


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