MoviePilot/tests/test_mcp_plugin_tools.py

"""
验证 MCP 插件工具动态发现、移除和调用准入边界。
"""

import asyncio
import json
from types import SimpleNamespace
from typing import Iterator
from unittest.mock import patch

import pytest

from app.agent.tools.base import MoviePilotTool
from app.agent.tools.catalog import ToolCatalogSnapshot
from app.agent.tools.factory import MoviePilotToolFactory
from app.agent.tools.manager import MoviePilotToolsManager
from app.api.endpoints import mcp
from app.foundation.singleton import Singleton
from app.runtime.extensions.plugin.manager import PluginManager


class DemoPluginTool(MoviePilotTool):
    """
    测试用插件 MCP 工具。
    """

    name: str = "demo_plugin_tool"
    description: str = "测试插件动态注册的 MCP 工具"

    async def run(self, **kwargs) -> str:
        """
        返回固定测试结果。
        """
        return "plugin-ok"


@pytest.fixture
def plugin_manager() -> Iterator[PluginManager]:
    """
    构造隔离的插件管理器并在测试后恢复原单例。
    """
    singleton_key = (PluginManager, (), frozenset())
    previous_instance = Singleton._instances.pop(singleton_key, None)
    manager = PluginManager()
    yield manager
    Singleton._instances.pop(singleton_key, None)
    if previous_instance is not None:
        Singleton._instances[singleton_key] = previous_instance


def _build_plugin() -> SimpleNamespace:
    """
    构造声明一个 Agent 工具的已启用插件。
    """
    return SimpleNamespace(
        plugin_name="Demo Plugin",
        get_state=lambda: True,
        get_agent_tools=lambda: [DemoPluginTool],
    )


def test_mcp_refreshes_tools_after_plugin_lifecycle_change(
    plugin_manager: PluginManager,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """
    MCP 管理器应发现初始化后新增的插件工具，并在插件移除后停止暴露。
    """
    with patch.object(
        MoviePilotToolFactory,
        "_get_builtin_tool_classes",
        return_value=[],
    ), patch.object(MoviePilotToolFactory, "EXTERNAL_SERVICE_TOOL_CLASSES", ()):
        tool_manager = MoviePilotToolsManager(
            session_id="mcp-plugin-test",
            user_id="api_user",
        )
        monkeypatch.setattr(mcp, "moviepilot_tool_manager", tool_manager)

        assert asyncio.run(mcp.handle_tools_list()) == {"tools": []}

        plugin_manager.running_plugins["DemoPlugin"] = _build_plugin()
        plugin_manager.clear_plugin_agent_tools_cache()

        listed_tools = asyncio.run(mcp.handle_tools_list())["tools"]
        assert [tool["name"] for tool in listed_tools] == ["demo_plugin_tool"]

        call_result = asyncio.run(
            mcp.handle_tools_call(
                {
                    "name": "demo_plugin_tool",
                    "arguments": {},
                }
            )
        )
        assert call_result == {"content": [{"type": "text", "text": "plugin-ok"}]}

        plugin_manager.running_plugins.pop("DemoPlugin")
        plugin_manager.clear_plugin_agent_tools_cache()

        assert asyncio.run(mcp.handle_tools_list()) == {"tools": []}
        missing_result = asyncio.run(
            mcp.handle_tools_call(
                {
                    "name": "demo_plugin_tool",
                    "arguments": {},
                }
            )
        )
        missing_payload = json.loads(missing_result["content"][0]["text"])
        assert "未找到" in missing_payload["error"]


def test_direct_manager_rejects_ambiguous_tool_identity() -> None:
    """
    HTTP/MCP direct 调用不得保留 first-wins 执行路径。
    """
    first = DemoPluginTool(session_id="session", user_id="user")
    second = DemoPluginTool(session_id="session", user_id="user")
    manager = MoviePilotToolsManager(session_id="session", user_id="user")
    manager.tools = [first, second]
    manager.catalog = ToolCatalogSnapshot.from_tools(
        manager.tools,
        plugin_revision=manager._plugin_agent_tools_revision,
        factory_revision=MoviePilotToolFactory.catalog_factory_revision(),
    )

    with pytest.raises(RuntimeError, match="TOOL_IDENTITY_AMBIGUOUS"):
        manager.list_tools()
    with pytest.raises(RuntimeError, match="TOOL_IDENTITY_AMBIGUOUS"):
        manager.get_tool("demo_plugin_tool")
    with pytest.raises(RuntimeError, match="TOOL_IDENTITY_AMBIGUOUS"):
        manager.get_strict_tool("demo_plugin_tool")
    with pytest.raises(RuntimeError, match="TOOL_IDENTITY_AMBIGUOUS"):
        asyncio.run(manager.call_tool("demo_plugin_tool", {}))


def test_read_skill_is_hidden_from_mcp_calls() -> None:
    """
    read_skill 只属于内置 Agent，中间件工具不得通过 MCP 直接调用。
    """
    builtin_names = {
        MoviePilotToolFactory._tool_class_name(tool_class)
        for tool_class in MoviePilotToolFactory.BUILTIN_TOOL_CLASSES
    }

    assert "read_skill" in mcp.MCP_HIDDEN_TOOLS
    assert "read_skill" not in builtin_names

    result = asyncio.run(
        mcp.handle_tools_call(
            {
                "name": "read_skill",
                "arguments": {"name": "moviepilot-api"},
            }
        )
    )

    assert result["isError"] is True
    assert "未找到" in result["content"][0]["text"]