"""Tests for gitea_resolve_task_capability tool.

Covers Issue #141 requirements and #145 regression tests for permission boundaries,
self blocks, and structured guidance.
"""
import os
import sys
import json
import tempfile
import unittest
from unittest.mock import patch, MagicMock

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

import gitea_config
import gitea_auth
import mcp_server

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

class TestResolveTaskCapability(unittest.TestCase):
    def setUp(self):
        self._remotes_patch = patch.dict(mcp_server.REMOTES, {
            "dadeschools": {"host": "gitea.example.com", "org": "Example-Org", "repo": "Example-Repo"},
            "prgs": {"host": "gitea.example.com", "org": "Example-Org", "repo": "Example-Repo"}
        })
        self._remotes_patch.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")
        self._write_config(CONFIG_RESOLVER)

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

    def _write_config(self, obj):
        with open(self.config_path, "w", encoding="utf-8") as fh:
            fh.write(json.dumps(obj))

    def _env(self, profile="author-profile"):
        return {
            "GITEA_MCP_CONFIG": self.config_path,
            "GITEA_MCP_PROFILE": profile,
            "GITEA_TOKEN_AUTHOR": "author-pass",
            "GITEA_TOKEN_REVIEWER": "reviewer-pass",
            "GITEA_TOKEN_MERGER": "merger-pass",
        }

    def test_diagnose_terminal_success_has_no_error(self):
        res = mcp_server.gitea_diagnose_terminal(
            command=[sys.executable, "-c", "raise SystemExit(0)"]
        )
        self.assertTrue(res["healthy"])
        self.assertIsNone(res["error_type"])
        self.assertEqual(res["error_msg"], "")

    @patch("mcp_server.api_request", return_value={"login": "author-user"})
    @patch("mcp_server.get_auth_header", return_value="token author-pass")
    def test_resolve_author_capable_task(self, _auth, _api):
        with patch.dict(os.environ, self._env("author-profile")):
            res = mcp_server.gitea_resolve_task_capability(task="create_issue", remote="prgs")
            self.assertEqual(res["requested_task"], "create_issue")
            self.assertEqual(res["required_operation_permission"], "gitea.issue.create")
            self.assertEqual(res["required_role_kind"], "author")
            self.assertEqual(res["active_profile"], "author-profile")
            self.assertTrue(res["allowed_in_current_session"])
            self.assertIn("author-profile", res["matching_configured_profile"])
            self.assertFalse(res["different_mcp_namespace_required"])
            self.assertIn("ready for operations", res["exact_safe_next_action"])

    @patch("mcp_server.native_mcp_preference.probe_terminal_spawn")
    @patch("mcp_server.api_request", return_value={"login": "author-user"})
    @patch("mcp_server.get_auth_header", return_value="token author-pass")
    def test_resolve_mutation_task_blocks_when_terminal_launcher_unhealthy(
        self,
        _auth,
        _api,
        mock_probe,
    ):
        mock_probe.return_value = {
            "healthy": False,
            "error_type": "missing cwd",
            "error_msg": "Current working directory does not exist: /missing",
            "cwd": "/missing",
            "command": ["git", "--version"],
            "diagnostics": {"cwd_exists": False},
        }
        env = self._env("author-profile")
        # Probe is skipped under pytest unless forced (mirrors production path).
        env["GITEA_FORCE_TERMINAL_PROBE"] = "1"
        with patch.dict(os.environ, env):
            res = mcp_server.gitea_resolve_task_capability(task="create_issue", remote="prgs")

        self.assertFalse(res["allowed_in_current_session"])
        self.assertTrue(res["stop_required"])
        self.assertTrue(res["infra_stop"])
        self.assertTrue(res["terminal_launcher_unhealthy"])
        self.assertTrue(res["blocked"])
        self.assertEqual(res["blocked_reason"], "BLOCKED + DIAGNOSE")
        self.assertEqual(res["terminal_launcher_diagnostics"]["error_type"], "missing cwd")
        self.assertIn("terminal-launcher-failure", res["exact_safe_next_action"])
        self.assertIn("BLOCKED + DIAGNOSE", res["exact_safe_next_action"])

    @patch("mcp_server.api_request", return_value={"login": "author-user"})
    @patch("mcp_server.get_auth_header", return_value="token author-pass")
    def test_resolve_reviewer_capable_task_blocked(self, _auth, _api):
        with patch.dict(os.environ, self._env("author-profile")):
            res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
            self.assertEqual(res["requested_task"], "review_pr")
            self.assertEqual(res["required_operation_permission"], "gitea.pr.review")
            self.assertEqual(res["required_role_kind"], "reviewer")
            self.assertEqual(res["active_profile"], "author-profile")
            self.assertFalse(res["allowed_in_current_session"])
            self.assertIn("reviewer-profile", res["matching_configured_profile"])
            self.assertTrue(res["different_mcp_namespace_required"])
            self.assertIn("gitea-reviewer", res["exact_safe_next_action"])

    @patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
    def test_resolve_reviewer_capable_task_allowed(self, _auth, _api):
        with patch.dict(os.environ, self._env("reviewer-profile")):
            res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
            self.assertTrue(res["allowed_in_current_session"])
            self.assertFalse(res["different_mcp_namespace_required"])

    @patch("mcp_server.api_request", return_value={"login": "author-user"})
    @patch("mcp_server.get_auth_header", return_value="token author-pass")
    def test_resolve_work_issue_author_profile_allowed(self, _auth, _api):
        with patch.dict(os.environ, self._env("author-profile")):
            for task in ("work_issue", "work-issue"):
                res = mcp_server.gitea_resolve_task_capability(
                    task=task, remote="prgs"
                )
                self.assertEqual(res["requested_task"], task)
                self.assertEqual(
                    res["required_operation_permission"], "gitea.pr.create"
                )
                self.assertEqual(res["required_role_kind"], "author")
                self.assertTrue(res["allowed_in_current_session"])

    @patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
    def test_resolve_work_issue_reviewer_profile_blocked(self, _auth, _api):
        with patch.dict(os.environ, self._env("reviewer-profile")):
            res = mcp_server.gitea_resolve_task_capability(
                task="work_issue", remote="prgs"
            )
            self.assertFalse(res["allowed_in_current_session"])
            self.assertEqual(res["required_role_kind"], "author")
            self.assertTrue(res["different_mcp_namespace_required"])

    @patch("mcp_server.api_request", return_value={"login": "author-user"})
    @patch("mcp_server.get_auth_header", return_value="token author-pass")
    def test_resolve_issue_comment_task(self, _auth, _api):
        with patch.dict(os.environ, self._env("author-profile")):
            res = mcp_server.gitea_resolve_task_capability(task="comment_issue", remote="prgs")
            self.assertTrue(res["allowed_in_current_session"])
            self.assertIn("author-profile", res["matching_configured_profile"])
            self.assertIn("reviewer-profile", res["matching_configured_profile"])

    @patch("mcp_server.api_request", return_value={"login": "author-user"})
    @patch("mcp_server.get_auth_header", return_value="token author-pass")
    def test_resolve_mark_issue_author_profile_allowed(self, _auth, _api):
        # #183: mark_issue must map to gitea.issue.comment (prgs-author allowed op).
        with patch.dict(os.environ, self._env("author-profile")):
            res = mcp_server.gitea_resolve_task_capability(task="mark_issue", remote="prgs")
            self.assertEqual(res["requested_task"], "mark_issue")
            self.assertEqual(res["required_operation_permission"], "gitea.issue.comment")
            self.assertTrue(res["allowed_in_current_session"])
            self.assertIn("gitea.issue.comment", res["active_profile_allowed_operations"])

    @patch("mcp_server.api_request", return_value={"login": "author-user"})
    @patch("mcp_server.get_auth_header", return_value="token author-pass")
    def test_resolve_lock_issue_author_profile_allowed(self, _auth, _api):
        # #275: lock_issue must be an exact resolver task for claim/lock gates.
        with patch.dict(os.environ, self._env("author-profile")):
            res = mcp_server.gitea_resolve_task_capability(task="lock_issue", remote="prgs")
            self.assertEqual(res["requested_task"], "lock_issue")
            self.assertEqual(res["required_operation_permission"], "gitea.issue.comment")
            self.assertTrue(res["allowed_in_current_session"])
            self.assertIn("gitea.issue.comment", res["active_profile_allowed_operations"])

    def test_resolve_unknown_task_fails_closed(self):
        # #723: unknown tasks return structured unknown_task (no ValueError escape).
        with patch.dict(os.environ, self._env("author-profile")):
            res = mcp_server.gitea_resolve_task_capability(
                task="invalid_task_xyz", remote="prgs"
            )
            self.assertIsInstance(res, dict)
            self.assertEqual(res.get("reason_code"), "unknown_task", res)
            self.assertFalse(res.get("allowed_in_current_session"))
            self.assertTrue(res.get("stop_required"))
            self.assertIs(res.get("mutation_performed"), False)

    @patch("mcp_server.api_request", return_value={"login": "author-user"})
    @patch("mcp_server.get_auth_header", return_value="token author-pass")
    def test_denied_role_exclusive_resolution_does_not_stamp_role(
        self, _auth, _api
    ):
        with patch.dict(os.environ, self._env("author-profile")):
            with patch.object(
                mcp_server,
                "record_preflight_check",
                wraps=mcp_server.record_preflight_check,
            ) as record:
                result = mcp_server.gitea_resolve_task_capability(
                    task="review_pr", remote="prgs"
                )

        self.assertFalse(result["allowed_in_current_session"], result)
        self.assertFalse(
            any(
                len(call.args) > 1 and call.args[1] == "reviewer"
                for call in record.call_args_list
            ),
            "denied reviewer resolution must never record a reviewer stamp",
        )
        self.assertIsNone(mcp_server._preflight_resolved_role)
        self.assertIsNone(mcp_server._preflight_resolved_task)

    # Additional regression tests per #145 for permission boundaries and structured guidance
    def test_issue_comment_does_not_imply_close(self):
        # Author profile has issue.comment but not issue.close
        with patch.dict(os.environ, self._env("author-profile")):
            res_comment = mcp_server.gitea_resolve_task_capability(task="comment_issue", remote="prgs")
            res_close = mcp_server.gitea_resolve_task_capability(task="close_issue", remote="prgs")
            self.assertTrue(res_comment["allowed_in_current_session"])
            self.assertIn("gitea.issue.comment", res_comment["active_profile_allowed_operations"])
            self.assertFalse(res_close["allowed_in_current_session"])
            self.assertIn("gitea.issue.close", res_close.get("required_operation_permission", ""))

    def test_pr_comment_does_not_imply_issue_comment(self):
        # Create a profile that has pr.comment but not issue.comment
        config = dict(CONFIG_RESOLVER)
        config["profiles"]["pr-only"] = {
            "enabled": True,
            "context": "ctx",
            "role": "author",
            "username": "pr-user",
            "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
            "allowed_operations": ["gitea.read", "gitea.pr.comment"],
            "forbidden_operations": ["gitea.issue.comment"],
            "execution_profile": "pr-only"
        }
        self._write_config(config)
        with patch.dict(os.environ, {**self._env("pr-only"), "GITEA_MCP_PROFILE": "pr-only"}):
            res_pr = mcp_server.gitea_resolve_task_capability(task="comment_pr", remote="prgs")
            res_issue = mcp_server.gitea_resolve_task_capability(task="comment_issue", remote="prgs")
            self.assertTrue(res_pr["allowed_in_current_session"])
            self.assertFalse(res_issue["allowed_in_current_session"])
            self.assertIn("gitea.pr.comment", res_pr["active_profile_allowed_operations"])
            self.assertNotIn("gitea.issue.comment", res_pr["active_profile_allowed_operations"])

    @patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
    def test_reviewer_cannot_push_branch(self, _auth, _api):
        with patch.dict(os.environ, self._env("reviewer-profile")):
            res = mcp_server.gitea_resolve_task_capability(task="push_branch", remote="prgs")
            self.assertFalse(res["allowed_in_current_session"])
            self.assertIn("author-profile", res["matching_configured_profile"])
            self.assertTrue(res["different_mcp_namespace_required"])

    @patch("mcp_server.api_request", return_value={"login": "author-user"})
    @patch("mcp_server.get_auth_header", return_value="token author-pass")
    def test_structured_guidance_for_missing_permission(self, _auth, _api):
        with patch.dict(os.environ, self._env("author-profile")):
            res = mcp_server.gitea_resolve_task_capability(task="merge_pr", remote="prgs")
            self.assertFalse(res["allowed_in_current_session"])
            self.assertIn("merger", res["exact_safe_next_action"].lower())
            self.assertEqual(res["required_operation_permission"], "gitea.pr.merge")
            self.assertEqual(res["required_role_kind"], "merger")

    @patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
    def test_reviewer_role_with_merge_permission_still_cannot_merge(self, _auth, _api):
        # #539: exact permission alone is insufficient. The active profile
        # role must also be merger for merge_pr.
        config = json.loads(json.dumps(CONFIG_RESOLVER))
        config["profiles"]["reviewer-with-merge"] = {
            "enabled": True,
            "context": "ctx",
            "role": "reviewer",
            "username": "reviewer-user",
            "auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"},
            "allowed_operations": [
                "gitea.read", "gitea.pr.review", "gitea.pr.approve",
                "gitea.pr.merge",
            ],
            "forbidden_operations": ["gitea.pr.create", "gitea.branch.push"],
            "execution_profile": "reviewer-with-merge",
        }
        self._write_config(config)
        with patch.dict(os.environ, self._env("reviewer-with-merge")):
            res = mcp_server.gitea_resolve_task_capability(
                task="merge_pr", remote="prgs"
            )
        self.assertEqual(res["required_role_kind"], "merger")
        self.assertEqual(res["active_role_kind"], "reviewer")
        self.assertTrue(res["active_profile_permission_allowed"])
        self.assertFalse(res["allowed_in_current_session"])
        self.assertTrue(res["stop_required"])
        self.assertIn("cannot perform merger task", " ".join(res["task_role_guidance"]))

    @patch("mcp_server.api_request", return_value={"login": "merger-user"})
    @patch("mcp_server.get_auth_header", return_value="token merger-pass")
    def test_merger_profile_can_resolve_merge_task(self, _auth, _api):
        with patch.dict(os.environ, self._env("merger-profile")):
            res = mcp_server.gitea_resolve_task_capability(
                task="merge_pr", remote="prgs"
            )
        self.assertEqual(res["required_role_kind"], "merger")
        self.assertEqual(res["active_role_kind"], "merger")
        self.assertTrue(res["allowed_in_current_session"])
        self.assertFalse(res["stop_required"])

    @patch("mcp_server.api_request")
    def test_self_review_blocked_structured(self, mock_api):
        # Test that self-approve (self-review gate) is blocked and returns structured reasons
        def side_api(method, url, auth):
            if "/user" in url:
                return {"login": "author-user"}
            if "/pulls/123" in url:
                return {"user": {"login": "author-user"}, "state": "open", "head": {"sha": "abc"}, "mergeable": True}
            return {}
        mock_api.side_effect = side_api
        with patch.dict(os.environ, self._env("author-profile")):
            res = mcp_server.gitea_check_pr_eligibility(pr_number=123, action="approve", remote="prgs")
            self.assertFalse(res.get("eligible", True))
            reasons = res.get("reasons", [])
            self.assertTrue(any("author" in str(r).lower() for r in reasons) or len(reasons) > 0)

    # Issue #216: close_pr is a first-class resolver task gated on gitea.pr.close.
    @patch("mcp_server.api_request", return_value={"login": "author-user"})
    @patch("mcp_server.get_auth_header", return_value="token author-pass")
    def test_resolve_close_pr_author_profile_without_close_blocked(self, _auth, _api):
        # Default author profile has PR create/comment but not close: the
        # close capability must never be implied by broader author ops.
        with patch.dict(os.environ, self._env("author-profile")):
            res = mcp_server.gitea_resolve_task_capability(task="close_pr", remote="prgs")
            self.assertEqual(res["requested_task"], "close_pr")
            self.assertEqual(res["required_operation_permission"], "gitea.pr.close")
            self.assertEqual(res["required_role_kind"], "author")
            self.assertFalse(res["allowed_in_current_session"])
            self.assertTrue(res["stop_required"])
            self.assertEqual(res["matching_configured_profile"], [])
            self.assertTrue(res["different_mcp_namespace_required"])

    @patch("mcp_server.api_request", return_value={"login": "author-user"})
    @patch("mcp_server.get_auth_header", return_value="token author-pass")
    def test_resolve_close_pr_author_profile_with_close_allowed(self, _auth, _api):
        # Operator-granted close capability resolves cleanly under an author profile.
        config = json.loads(json.dumps(CONFIG_RESOLVER))
        config["profiles"]["author-closer"] = {
            "enabled": True,
            "context": "ctx",
            "role": "author",
            "username": "author-user",
            "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
            "allowed_operations": ["gitea.read", "gitea.pr.close"],
            "forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"],
            "execution_profile": "author-closer",
        }
        self._write_config(config)
        with patch.dict(os.environ, self._env("author-closer")):
            res = mcp_server.gitea_resolve_task_capability(task="close_pr", remote="prgs")
            self.assertTrue(res["allowed_in_current_session"])
            self.assertFalse(res["stop_required"])
            self.assertIn("author-closer", res["matching_configured_profile"])
            self.assertIn("ready for operations", res["exact_safe_next_action"])

    @patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
    def test_resolve_close_pr_reviewer_profile_blocked(self, _auth, _api):
        # close_pr is author-side: a reviewer profile must be told to stop.
        with patch.dict(os.environ, self._env("reviewer-profile")):
            res = mcp_server.gitea_resolve_task_capability(task="close_pr", remote="prgs")
            self.assertFalse(res["allowed_in_current_session"])
            self.assertTrue(res["stop_required"])
            self.assertEqual(res["required_role_kind"], "author")
            self.assertIn("author", res["exact_safe_next_action"].lower())

    @patch("mcp_server.api_request", return_value={"login": "author-user"})
    @patch("mcp_server.get_auth_header", return_value="token author-pass")
    def test_resolve_reconcile_already_landed_pr_requires_close(self, _auth, _api):
        with patch.dict(os.environ, self._env("author-profile")):
            res = mcp_server.gitea_resolve_task_capability(
                task="reconcile_already_landed_pr", remote="prgs"
            )
            self.assertEqual(res["requested_task"], "reconcile_already_landed_pr")
            self.assertEqual(res["required_operation_permission"], "gitea.pr.close")
            self.assertEqual(res["required_role_kind"], "reconciler")
            self.assertFalse(res["allowed_in_current_session"])

    @patch("mcp_server.api_request", return_value={"login": "author-user"})
    @patch("mcp_server.get_auth_header", return_value="token author-pass")
    def test_resolve_pr_queue_cleanup_author_profile_blocked(self, _auth, _api):
        with patch.dict(os.environ, self._env("author-profile")):
            res = mcp_server.gitea_resolve_task_capability(
                task="pr_queue_cleanup", remote="prgs"
            )
            self.assertEqual(res["required_role_kind"], "reviewer")
            self.assertFalse(res["allowed_in_current_session"])
            self.assertTrue(res["stop_required"])

    @patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
    def test_resolve_pr_queue_cleanup_reviewer_profile_allowed(self, _auth, _api):
        with patch.dict(os.environ, self._env("reviewer-profile")):
            res = mcp_server.gitea_resolve_task_capability(
                task="pr_queue_cleanup", remote="prgs"
            )
            self.assertEqual(res["required_operation_permission"], "gitea.pr.review")
            self.assertTrue(res["allowed_in_current_session"])
            self.assertFalse(res["stop_required"])

    @patch("mcp_server.api_request", return_value={"login": "author-user"})
    @patch("mcp_server.get_auth_header", return_value="token author-pass")
    def test_close_pr_known_and_lookalike_tasks_still_fail_closed(self, _auth, _api):
        # close_pr resolves (no Unknown-task error); near-miss spellings keep
        # failing closed so no untracked close fallback can be rationalized.
        with patch.dict(os.environ, self._env("author-profile")):
            res = mcp_server.gitea_resolve_task_capability(task="close_pr", remote="prgs")
            self.assertEqual(res["required_operation_permission"], "gitea.pr.close")
            for unknown in ("close_pull_request", "close", "pr_close"):
                unk = mcp_server.gitea_resolve_task_capability(
                    task=unknown, remote="prgs"
                )
                self.assertEqual(unk.get("reason_code"), "unknown_task", unk)
                self.assertFalse(unk.get("allowed_in_current_session"))

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