#
#  Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
#
#  Licensed under the Apache License, Version 2.0 (the "License");
#  you may not use this file except in compliance with the License.
#  You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
#  limitations under the License.
#
import asyncio
import importlib.util
import inspect
import json
import sys
from contextlib import nullcontext
from functools import wraps
from pathlib import Path
from types import ModuleType, SimpleNamespace

import pytest


class _DummyManager:
    def route(self, *_args, **_kwargs):
        def decorator(func):
            return func

        return decorator


class _Args(dict):
    def getlist(self, key):
        value = self.get(key, [])
        if isinstance(value, list):
            return value
        return [value]


class _Field:
    def __init__(self, name):
        self.name = name

    def __eq__(self, other):
        return (self.name, other)


class _DummyMCPServer:
    id = _Field("id")
    tenant_id = _Field("tenant_id")

    def __init__(self, **kwargs):
        self.id = kwargs.get("id", "")
        self.name = kwargs.get("name", "")
        self.url = kwargs.get("url", "")
        self.server_type = kwargs.get("server_type", "sse")
        self.tenant_id = kwargs.get("tenant_id", "tenant_1")
        self.variables = kwargs.get("variables", {})
        self.headers = kwargs.get("headers", {})

    def to_dict(self):
        return {
            "id": self.id,
            "name": self.name,
            "url": self.url,
            "server_type": self.server_type,
            "tenant_id": self.tenant_id,
            "variables": self.variables,
            "headers": self.headers,
        }


class _DummyMCPServerService:
    @staticmethod
    def get_servers(*_args, **_kwargs):
        return []

    @staticmethod
    def get_or_none(*_args, **_kwargs):
        return None

    @staticmethod
    def get_by_id(*_args, **_kwargs):
        return False, None

    @staticmethod
    def get_by_name_and_tenant(*_args, **_kwargs):
        return False, None

    @staticmethod
    def insert(**_kwargs):
        return True

    @staticmethod
    def filter_update(*_args, **_kwargs):
        return True

    @staticmethod
    def delete_by_ids(*_args, **_kwargs):
        return True


class _DummyTenantService:
    @staticmethod
    def get_by_id(*_args, **_kwargs):
        return True, SimpleNamespace(id="tenant_1")


class _DummyTool:
    def __init__(self, name):
        self._name = name

    def model_dump(self):
        return {"name": self._name}


class _DummyMCPToolCallSession:
    def __init__(self, _mcp_server, _variables):
        self._tools = [_DummyTool("tool_a"), _DummyTool("tool_b")]

    def get_tools(self, _timeout):
        return self._tools

    def tool_call(self, _name, _arguments, _timeout):
        return "ok"


def _run(coro):
    return asyncio.run(coro)


def _set_request_json(monkeypatch, module, payload):
    async def _request_json():
        return payload

    monkeypatch.setattr(module, "get_request_json", _request_json)


def _stub_url_safety(monkeypatch, module, unsafe_urls=None):
    unsafe_urls = set(unsafe_urls or [])

    def _assert_url_is_safe(url):
        if url in unsafe_urls:
            raise ValueError("blocked unsafe url")
        return "safe.example", "93.184.216.34"

    monkeypatch.setattr(module, "assert_url_is_safe", _assert_url_is_safe)
    monkeypatch.setattr(module, "pin_dns_global", lambda *_args, **_kwargs: nullcontext())


@pytest.fixture(scope="session")
def auth():
    return "unit-auth"


@pytest.fixture(scope="session", autouse=True)
def set_tenant_info():
    return None


def _load_mcp_api(monkeypatch):
    repo_root = Path(__file__).resolve().parents[3]

    quart_mod = ModuleType("quart")
    quart_mod.Response = object
    quart_mod.request = SimpleNamespace(args=_Args({}))
    monkeypatch.setitem(sys.modules, "quart", quart_mod)

    common_pkg = ModuleType("common")
    common_pkg.__path__ = [str(repo_root / "common")]
    monkeypatch.setitem(sys.modules, "common", common_pkg)

    constants_mod = ModuleType("common.constants")
    constants_mod.VALID_MCP_SERVER_TYPES = {"sse", "streamable-http"}
    monkeypatch.setitem(sys.modules, "common.constants", constants_mod)

    apps_mod = ModuleType("api.apps")
    apps_mod.current_user = SimpleNamespace(id="tenant_1")
    apps_mod.login_required = lambda func: func
    monkeypatch.setitem(sys.modules, "api.apps", apps_mod)

    db_models_mod = ModuleType("api.db.db_models")
    db_models_mod.MCPServer = _DummyMCPServer
    monkeypatch.setitem(sys.modules, "api.db.db_models", db_models_mod)

    mcp_service_mod = ModuleType("api.db.services.mcp_server_service")
    mcp_service_mod.MCPServerService = _DummyMCPServerService
    monkeypatch.setitem(sys.modules, "api.db.services.mcp_server_service", mcp_service_mod)

    user_service_mod = ModuleType("api.db.services.user_service")
    user_service_mod.TenantService = _DummyTenantService
    monkeypatch.setitem(sys.modules, "api.db.services.user_service", user_service_mod)

    mcp_conn_mod = ModuleType("common.mcp_tool_call_conn")
    mcp_conn_mod.MCPToolCallSession = _DummyMCPToolCallSession
    mcp_conn_mod.close_multiple_mcp_toolcall_sessions = lambda _sessions: None
    monkeypatch.setitem(sys.modules, "common.mcp_tool_call_conn", mcp_conn_mod)

    api_utils_mod = ModuleType("api.utils.api_utils")

    async def _default_request_json():
        return {}

    def _get_json_result(code=0, message="success", data=None):
        return {"code": code, "message": message, "data": data}

    def _get_data_error_result(code=102, message="Sorry! Data missing!"):
        return {"code": code, "message": message}

    def _server_error_response(error):
        return {"code": 100, "message": repr(error)}

    async def _get_mcp_tools(*_args, **_kwargs):
        return {}

    def _validate_request(*_args, **_kwargs):
        def _decorator(func):
            @wraps(func)
            async def _wrapped(*func_args, **func_kwargs):
                if inspect.iscoroutinefunction(func):
                    return await func(*func_args, **func_kwargs)
                return func(*func_args, **func_kwargs)

            return _wrapped

        return _decorator

    api_utils_mod.get_request_json = _default_request_json
    api_utils_mod.get_json_result = _get_json_result
    api_utils_mod.get_data_error_result = _get_data_error_result
    api_utils_mod.server_error_response = _server_error_response
    api_utils_mod.validate_request = _validate_request
    api_utils_mod.get_mcp_tools = _get_mcp_tools
    monkeypatch.setitem(sys.modules, "api.utils.api_utils", api_utils_mod)

    web_utils_mod = ModuleType("api.utils.web_utils")

    def _get_float(data, key, default):
        try:
            return float(data.get(key, default))
        except (TypeError, ValueError):
            return default

    def _safe_json_parse(value):
        if isinstance(value, (dict, list)):
            return value
        if value in (None, ""):
            return {}
        try:
            return json.loads(value)
        except (TypeError, ValueError):
            return {}

    web_utils_mod.get_float = _get_float
    web_utils_mod.safe_json_parse = _safe_json_parse
    monkeypatch.setitem(sys.modules, "api.utils.web_utils", web_utils_mod)

    module_name = "test_mcp_api_unit_module"
    module_path = repo_root / "api" / "apps" / "restful_apis" / "mcp_api.py"
    spec = importlib.util.spec_from_file_location(module_name, module_path)
    module = importlib.util.module_from_spec(spec)
    module.manager = _DummyManager()
    monkeypatch.setitem(sys.modules, module_name, module)
    spec.loader.exec_module(module)
    return module


@pytest.mark.p2
def test_list_mcp_desc_pagination_and_exception(monkeypatch):
    module = _load_mcp_api(monkeypatch)

    monkeypatch.setattr(
        module,
        "request",
        SimpleNamespace(args=_Args({"keywords": "k", "page": "2", "page_size": "1", "orderby": "create_time", "desc": "false"})),
    )
    _set_request_json(monkeypatch, module, {"mcp_ids": []})
    monkeypatch.setattr(module.MCPServerService, "get_servers", lambda *_args, **_kwargs: [{"id": "a"}, {"id": "b"}])

    res = _run(module.list_mcp())
    assert res["code"] == 0
    assert res["data"]["total"] == 2
    assert res["data"]["mcp_servers"] == [{"id": "b"}]

    monkeypatch.setattr(module, "request", SimpleNamespace(args=_Args({})))
    _set_request_json(monkeypatch, module, {"mcp_ids": []})

    def _raise_list(*_args, **_kwargs):
        raise RuntimeError("list explode")

    monkeypatch.setattr(module.MCPServerService, "get_servers", _raise_list)
    res = _run(module.list_mcp())
    assert res["code"] == 100
    assert "list explode" in res["message"]


@pytest.mark.p2
def test_detail_not_found_success_and_exception(monkeypatch):
    module = _load_mcp_api(monkeypatch)
    monkeypatch.setattr(module, "request", SimpleNamespace(args=_Args({})))

    monkeypatch.setattr(module.MCPServerService, "get_or_none", lambda **_kwargs: None)
    res = module.detail("mcp-1")
    assert res["code"] == 102
    assert "Cannot find MCP server mcp-1 for user tenant_1" in res["message"]

    monkeypatch.setattr(
        module.MCPServerService,
        "get_or_none",
        lambda **_kwargs: _DummyMCPServer(id="mcp-1", name="srv", url="http://a", server_type="sse", tenant_id="tenant_1"),
    )
    res = module.detail("mcp-1")
    assert res["code"] == 0
    assert res["data"]["id"] == "mcp-1"

    def _raise_detail(**_kwargs):
        raise RuntimeError("detail explode")

    monkeypatch.setattr(module.MCPServerService, "get_or_none", _raise_detail)
    res = module.detail("mcp-1")
    assert res["code"] == 100
    assert "detail explode" in res["message"]


@pytest.mark.p2
def test_create_validation_guards(monkeypatch):
    module = _load_mcp_api(monkeypatch)

    monkeypatch.setattr(module.MCPServerService, "get_by_name_and_tenant", lambda **_kwargs: (False, None))

    _set_request_json(monkeypatch, module, {"name": "srv", "url": "http://a", "server_type": "invalid"})
    res = _run(module.create.__wrapped__())
    assert "Unsupported MCP server type" in res["message"]

    _set_request_json(monkeypatch, module, {"name": "", "url": "http://a", "server_type": "sse"})
    res = _run(module.create.__wrapped__())
    assert "Invalid MCP name" in res["message"]

    monkeypatch.setattr(module.MCPServerService, "get_by_name_and_tenant", lambda **_kwargs: (True, object()))
    _set_request_json(monkeypatch, module, {"name": "srv", "url": "http://a", "server_type": "sse"})
    res = _run(module.create.__wrapped__())
    assert "Duplicated MCP server name" in res["message"]

    monkeypatch.setattr(module.MCPServerService, "get_by_name_and_tenant", lambda **_kwargs: (False, None))
    _set_request_json(monkeypatch, module, {"name": "srv", "url": "", "server_type": "sse"})
    res = _run(module.create.__wrapped__())
    assert "Invalid url" in res["message"]

    _set_request_json(monkeypatch, module, {"name": "srv", "url": 123, "server_type": "sse"})
    res = _run(module.create.__wrapped__())
    assert "Invalid url" in res["message"]

    _set_request_json(monkeypatch, module, {"name": "srv", "url": "http://unsafe", "server_type": "sse"})
    _stub_url_safety(monkeypatch, module, {"http://unsafe"})
    res = _run(module.create.__wrapped__())
    assert "blocked unsafe url" in res["message"]


@pytest.mark.p2
def test_create_service_paths(monkeypatch):
    module = _load_mcp_api(monkeypatch)
    _stub_url_safety(monkeypatch, module)

    base_payload = {
        "name": "srv",
        "url": "http://server",
        "server_type": "sse",
        "headers": '{"Authorization": "x"}',
        "variables": '{"tools": {"old": 1}, "token": "abc"}',
        "timeout": "2.5",
    }

    monkeypatch.setattr(module, "get_uuid", lambda: "uuid-create")
    monkeypatch.setattr(module.MCPServerService, "get_by_name_and_tenant", lambda **_kwargs: (False, None))

    _set_request_json(monkeypatch, module, dict(base_payload))
    monkeypatch.setattr(module.TenantService, "get_by_id", lambda *_args, **_kwargs: (False, None))
    res = _run(module.create.__wrapped__())
    assert "Tenant not found" in res["message"]

    _set_request_json(monkeypatch, module, dict(base_payload))
    monkeypatch.setattr(module.TenantService, "get_by_id", lambda *_args, **_kwargs: (True, object()))

    async def _thread_pool_tools_error(_func, _servers, _timeout):
        return None, "tools error"

    monkeypatch.setattr(module, "thread_pool_exec", _thread_pool_tools_error)
    res = _run(module.create.__wrapped__())
    assert res["code"] == 102
    assert "tools error" in res["message"]

    _set_request_json(monkeypatch, module, dict(base_payload))

    async def _thread_pool_ok(_func, servers, _timeout):
        return {servers[0].name: [{"name": "tool_a"}, {"invalid": True}]}, None

    monkeypatch.setattr(module, "thread_pool_exec", _thread_pool_ok)
    monkeypatch.setattr(module.MCPServerService, "insert", lambda **_kwargs: False)
    res = _run(module.create.__wrapped__())
    assert res["code"] == 102
    assert "Failed to create MCP server" in res["message"]

    _set_request_json(monkeypatch, module, dict(base_payload))
    monkeypatch.setattr(module.MCPServerService, "insert", lambda **_kwargs: True)
    res = _run(module.create.__wrapped__())
    assert res["code"] == 0
    assert res["data"]["id"] == "uuid-create"
    assert res["data"]["tenant_id"] == "tenant_1"
    assert res["data"]["variables"]["tools"] == {"tool_a": {"name": "tool_a"}}

    _set_request_json(monkeypatch, module, dict(base_payload))

    async def _thread_pool_raises(_func, _servers, _timeout):
        raise RuntimeError("create explode")

    monkeypatch.setattr(module, "thread_pool_exec", _thread_pool_raises)
    res = _run(module.create.__wrapped__())
    assert res["code"] == 100
    assert "create explode" in res["message"]


@pytest.mark.p2
def test_update_validation_guards(monkeypatch):
    module = _load_mcp_api(monkeypatch)

    existing = _DummyMCPServer(id="mcp-1", name="srv", url="http://server", server_type="sse", tenant_id="tenant_1", variables={}, headers={})

    _set_request_json(monkeypatch, module, {"mcp_id": "mcp-1"})
    monkeypatch.setattr(module.MCPServerService, "get_by_id", lambda _mcp_id: (False, None))
    res = _run(module.update("mcp-1"))
    assert "Cannot find MCP server" in res["message"]

    _set_request_json(monkeypatch, module, {"mcp_id": "mcp-1"})
    monkeypatch.setattr(
        module.MCPServerService,
        "get_by_id",
        lambda _mcp_id: (True, _DummyMCPServer(id="mcp-1", name="srv", url="http://server", server_type="sse", tenant_id="other", variables={}, headers={})),
    )
    res = _run(module.update("mcp-1"))
    assert "Cannot find MCP server" in res["message"]

    _set_request_json(monkeypatch, module, {"mcp_id": "mcp-1", "server_type": "invalid"})
    monkeypatch.setattr(module.MCPServerService, "get_by_id", lambda _mcp_id: (True, existing))
    res = _run(module.update("mcp-1"))
    assert "Unsupported MCP server type" in res["message"]

    _set_request_json(monkeypatch, module, {"mcp_id": "mcp-1", "name": "a" * 256})
    res = _run(module.update("mcp-1"))
    assert "Invalid MCP name" in res["message"]

    _set_request_json(monkeypatch, module, {"mcp_id": "mcp-1", "url": ""})
    res = _run(module.update("mcp-1"))
    assert "Invalid url" in res["message"]

    _set_request_json(monkeypatch, module, {"mcp_id": "mcp-1", "url": {"raw": "http://a"}})
    res = _run(module.update("mcp-1"))
    assert "Invalid url" in res["message"]

    _set_request_json(monkeypatch, module, {"mcp_id": "mcp-1", "url": "http://unsafe"})
    _stub_url_safety(monkeypatch, module, {"http://unsafe"})
    res = _run(module.update("mcp-1"))
    assert "blocked unsafe url" in res["message"]


@pytest.mark.p2
def test_update_service_paths(monkeypatch):
    module = _load_mcp_api(monkeypatch)
    _stub_url_safety(monkeypatch, module)

    existing = _DummyMCPServer(
        id="mcp-1",
        name="srv",
        url="http://server",
        server_type="sse",
        tenant_id="tenant_1",
        variables={"tools": {"old": {"enabled": True}}, "token": "abc"},
        headers={"Authorization": "old"},
    )
    updated = _DummyMCPServer(
        id="mcp-1",
        name="srv-new",
        url="http://server-new",
        server_type="sse",
        tenant_id="tenant_1",
        variables={"tools": {"tool_a": {"name": "tool_a"}}},
        headers={"Authorization": "new"},
    )

    base_payload = {
        "mcp_id": "mcp-1",
        "name": "srv-new",
        "url": "http://server-new",
        "server_type": "sse",
        "headers": '{"Authorization": "new"}',
        "variables": '{"tools": {"ignore": 1}, "token": "new"}',
        "timeout": "3.0",
    }

    _set_request_json(monkeypatch, module, dict(base_payload))
    monkeypatch.setattr(module.MCPServerService, "get_by_id", lambda _mcp_id: (True, existing))

    async def _thread_pool_tools_error(_func, _servers, _timeout):
        return None, "update tools error"

    monkeypatch.setattr(module, "thread_pool_exec", _thread_pool_tools_error)
    res = _run(module.update("mcp-1"))
    assert res["code"] == 102
    assert "update tools error" in res["message"]

    _set_request_json(monkeypatch, module, dict(base_payload))

    async def _thread_pool_ok(_func, servers, _timeout):
        return {servers[0].name: [{"name": "tool_a"}, {"bad": True}]}, None

    monkeypatch.setattr(module, "thread_pool_exec", _thread_pool_ok)
    monkeypatch.setattr(module.MCPServerService, "filter_update", lambda *_args, **_kwargs: False)
    res = _run(module.update("mcp-1"))
    assert "Failed to updated MCP server" in res["message"]

    _set_request_json(monkeypatch, module, dict(base_payload))
    monkeypatch.setattr(module.MCPServerService, "filter_update", lambda *_args, **_kwargs: True)

    def _get_by_id_fetch_fail(_mcp_id):
        if _get_by_id_fetch_fail.calls == 0:
            _get_by_id_fetch_fail.calls += 1
            return True, existing
        return False, None

    _get_by_id_fetch_fail.calls = 0
    monkeypatch.setattr(module.MCPServerService, "get_by_id", _get_by_id_fetch_fail)
    res = _run(module.update("mcp-1"))
    assert "Failed to fetch updated MCP server" in res["message"]

    _set_request_json(monkeypatch, module, dict(base_payload))

    def _get_by_id_success(_mcp_id):
        if _get_by_id_success.calls == 0:
            _get_by_id_success.calls += 1
            return True, existing
        return True, updated

    _get_by_id_success.calls = 0
    monkeypatch.setattr(module.MCPServerService, "get_by_id", _get_by_id_success)
    res = _run(module.update("mcp-1"))
    assert res["code"] == 0
    assert res["data"]["id"] == "mcp-1"

    _set_request_json(monkeypatch, module, dict(base_payload))
    monkeypatch.setattr(module.MCPServerService, "get_by_id", lambda _mcp_id: (True, existing))

    async def _thread_pool_raises(_func, _servers, _timeout):
        raise RuntimeError("update explode")

    monkeypatch.setattr(module, "thread_pool_exec", _thread_pool_raises)
    res = _run(module.update("mcp-1"))
    assert res["code"] == 100
    assert "update explode" in res["message"]


@pytest.mark.p2
def test_rm_failure_success_and_exception(monkeypatch):
    module = _load_mcp_api(monkeypatch)
    server = _DummyMCPServer(id="id1", name="srv", url="http://a", server_type="sse", tenant_id="tenant_1", variables={})
    monkeypatch.setattr(module.MCPServerService, "get_by_id", lambda _mcp_id: (True, server))

    _set_request_json(monkeypatch, module, {"mcp_ids": ["a", "b"]})
    monkeypatch.setattr(module.MCPServerService, "delete_by_ids", lambda _ids: False)
    res = _run(module.rm("id1"))
    assert "Failed to delete MCP servers" in res["message"]

    _set_request_json(monkeypatch, module, {"mcp_ids": ["a", "b"]})
    monkeypatch.setattr(module.MCPServerService, "delete_by_ids", lambda _ids: True)
    res = _run(module.rm("id1"))
    assert res["code"] == 0
    assert res["data"] is True

    _set_request_json(monkeypatch, module, {"mcp_ids": ["a", "b"]})

    def _raise_rm(_ids):
        raise RuntimeError("rm explode")

    monkeypatch.setattr(module.MCPServerService, "delete_by_ids", _raise_rm)
    res = _run(module.rm("id1"))
    assert res["code"] == 100
    assert "rm explode" in res["message"]


@pytest.mark.p2
def test_import_multiple_missing_servers_and_exception(monkeypatch):
    module = _load_mcp_api(monkeypatch)
    _stub_url_safety(monkeypatch, module)

    _set_request_json(monkeypatch, module, {"mcpServers": {}})
    res = _run(module.import_multiple.__wrapped__())
    assert "No MCP servers provided" in res["message"]

    _set_request_json(monkeypatch, module, {"mcpServers": {"srv": {"type": "sse", "url": "http://x"}}, "timeout": "1"})

    def _raise_import(**_kwargs):
        raise RuntimeError("import explode")

    monkeypatch.setattr(module.MCPServerService, "get_by_name_and_tenant", _raise_import)
    res = _run(module.import_multiple.__wrapped__())
    assert res["code"] == 100
    assert "import explode" in res["message"]


@pytest.mark.p2
def test_import_multiple_mixed_results(monkeypatch):
    module = _load_mcp_api(monkeypatch)
    _stub_url_safety(monkeypatch, module, {"http://unsafe"})

    payload = {
        "mcpServers": {
            "missing_fields": {"type": "sse"},
            "": {"type": "sse", "url": "http://empty"},
            "invalid_type": {"type": "invalid", "url": "http://invalid"},
            "non_string_url": {"type": "sse", "url": True},
            "unsafe": {"type": "sse", "url": "http://unsafe"},
            "dup": {"type": "sse", "url": "http://dup", "authorization_token": "dup-token"},
            "tool_err": {"type": "sse", "url": "http://err"},
            "insert_fail": {"type": "sse", "url": "http://fail"},
        },
        "timeout": "3",
    }
    _set_request_json(monkeypatch, module, payload)

    monkeypatch.setattr(module, "get_uuid", lambda: "uuid-import")

    def _get_by_name_and_tenant(name, tenant_id):
        if name == "dup" and not _get_by_name_and_tenant.first_dup_seen:
            _get_by_name_and_tenant.first_dup_seen = True
            return True, object()
        return False, None

    _get_by_name_and_tenant.first_dup_seen = False
    monkeypatch.setattr(module.MCPServerService, "get_by_name_and_tenant", _get_by_name_and_tenant)

    async def _thread_pool_exec(func, servers, _timeout):
        mcp_server = servers[0]
        if mcp_server.name == "tool_err":
            return None, "tool call failed"
        return {mcp_server.name: [{"name": "tool_a"}, {"invalid": True}]}, None

    monkeypatch.setattr(module, "thread_pool_exec", _thread_pool_exec)

    def _insert(**kwargs):
        return kwargs["name"] != "insert_fail"

    monkeypatch.setattr(module.MCPServerService, "insert", _insert)

    res = _run(module.import_multiple.__wrapped__())
    assert res["code"] == 0

    results = {item["server"]: item for item in res["data"]["results"]}
    assert results["missing_fields"]["success"] is False
    assert "Missing required fields" in results["missing_fields"]["message"]
    assert results[""]["success"] is False
    assert "Invalid MCP name" in results[""]["message"]
    assert results["invalid_type"]["success"] is False
    assert "Unsupported MCP server type" in results["invalid_type"]["message"]
    assert results["non_string_url"]["success"] is False
    assert "Invalid url" in results["non_string_url"]["message"]
    assert results["unsafe"]["success"] is False
    assert "blocked unsafe url" in results["unsafe"]["message"]
    assert results["tool_err"]["success"] is False
    assert "tool call failed" in results["tool_err"]["message"]
    assert results["insert_fail"]["success"] is False
    assert "Failed to create MCP server" in results["insert_fail"]["message"]
    assert results["dup"]["success"] is True
    assert results["dup"]["new_name"] == "dup_0"
    assert "Renamed from 'dup' to 'dup_0' avoid duplication" == results["dup"]["message"]


@pytest.mark.p2
def test_detail_download_success_and_exception(monkeypatch):
    module = _load_mcp_api(monkeypatch)
    monkeypatch.setattr(module, "request", SimpleNamespace(args=_Args({"mode": "download"})))

    monkeypatch.setattr(
        module.MCPServerService,
        "get_by_id",
        lambda _mcp_id: (
            True,
            _DummyMCPServer(
                id="id1",
                name="srv-one",
                url="http://one",
                server_type="sse",
                tenant_id="tenant_1",
                variables={"authorization_token": "tok", "tools": {"tool_a": {"enabled": True}}},
            ),
        ),
    )
    res = module.detail("id1")
    assert res["code"] == 0
    assert list(res["data"]["mcpServers"].keys()) == ["srv-one"]

    monkeypatch.setattr(module.MCPServerService, "get_by_id", lambda _mcp_id: (False, None))
    res = module.detail("missing")
    assert res["code"] == 102
    assert "Cannot find MCP server missing for user tenant_1" in res["message"]

    monkeypatch.setattr(
        module.MCPServerService,
        "get_by_id",
        lambda _mcp_id: (
            True,
            _DummyMCPServer(
                id="id2",
                name="srv-two",
                url="http://two",
                server_type="sse",
                tenant_id="other",
                variables={},
            ),
        ),
    )
    res = module.detail("id2")
    assert res["code"] == 102
    assert "Cannot find MCP server id2 for user tenant_1" in res["message"]

    def _raise_export(_mcp_id):
        raise RuntimeError("export explode")

    monkeypatch.setattr(module.MCPServerService, "get_by_id", _raise_export)
    res = module.detail("id1")
    assert res["code"] == 100
    assert "export explode" in res["message"]


@pytest.mark.p2
def test_test_mcp_route_matrix_unit(monkeypatch):
    module = _load_mcp_api(monkeypatch)
    _stub_url_safety(monkeypatch, module)

    _set_request_json(monkeypatch, module, {"url": "", "server_type": "sse"})
    res = _run(module.test_mcp("mcp-1"))
    assert "Invalid MCP url" in res["message"]

    _set_request_json(monkeypatch, module, {"url": ["http://a"], "server_type": "sse"})
    res = _run(module.test_mcp("mcp-1"))
    assert "Invalid MCP url" in res["message"]

    _set_request_json(monkeypatch, module, {"url": "http://a", "server_type": "invalid"})
    res = _run(module.test_mcp("mcp-1"))
    assert "Unsupported MCP server type" in res["message"]

    close_calls = []

    async def _thread_pool_exec_inner_error(func, *args):
        if func is module.close_multiple_mcp_toolcall_sessions:
            close_calls.append(args[0])
            return None
        if getattr(func, "__name__", "") == "get_tools":
            raise RuntimeError("get tools explode")
        return func(*args)

    monkeypatch.setattr(module, "thread_pool_exec", _thread_pool_exec_inner_error)
    _set_request_json(monkeypatch, module, {"url": "http://a", "server_type": "sse"})
    res = _run(module.test_mcp("mcp-1"))
    assert res["code"] == 102
    assert "Test MCP error: get tools explode" in res["message"]
    assert close_calls and len(close_calls[-1]) == 1

    close_calls_success = []

    async def _thread_pool_exec_success(func, *args):
        if func is module.close_multiple_mcp_toolcall_sessions:
            close_calls_success.append(args[0])
            return None
        return func(*args)

    monkeypatch.setattr(module, "thread_pool_exec", _thread_pool_exec_success)
    _set_request_json(monkeypatch, module, {"url": "http://a", "server_type": "sse"})
    res = _run(module.test_mcp("mcp-1"))
    assert res["code"] == 0
    assert res["data"][0]["name"] == "tool_a"
    assert all(tool["enabled"] is True for tool in res["data"])
    assert close_calls_success and len(close_calls_success[-1]) == 1

    def _raise_session(*_args, **_kwargs):
        raise RuntimeError("session explode")

    monkeypatch.setattr(module, "MCPToolCallSession", _raise_session)
    _set_request_json(monkeypatch, module, {"url": "http://a", "server_type": "sse"})
    res = _run(module.test_mcp("mcp-1"))
    assert res["code"] == 100
    assert "session explode" in res["message"]
