Octop/tests/unit/test_plugins.py

"""Unit tests for octop-harness plugin loader."""

from __future__ import annotations

from pathlib import Path
from typing import Any

import pytest
from langchain_core.tools import StructuredTool
from octop_harness.plugins import (
    PluginRegistry,
    build_plugin_tools,
    collect_plugin_tool_configs,
    load_plugin_dir,
)

from octop.infra.agents.plugins.plugin_tool_names import (
    sanitize_plugin_tool_name,
    sanitize_plugin_tool_names,
)

_FIXTURE = Path(__file__).resolve().parents[1] / "fixtures" / "plugins" / "echo-tool"


@pytest.fixture(autouse=True)
def _reset_registry() -> None:
    PluginRegistry.reset()
    yield
    PluginRegistry.reset()


def test_load_echo_tool_plugin() -> None:
    loaded = load_plugin_dir(_FIXTURE, install_deps=False)
    assert loaded.manifest.id == "echo-tool"
    assert len(loaded.tools) == 1
    assert loaded.tools[0].name == "echo_message"


def test_build_plugin_tools_respects_enabled_flag() -> None:
    load_plugin_dir(_FIXTURE, install_deps=False)
    disabled = build_plugin_tools(
        agent_plugins={"echo-tool": {"tools": {"echo_message": {"enabled": False}}}},
    )
    assert disabled == []
    enabled = build_plugin_tools(
        agent_plugins={"echo-tool": {"tools": {"echo_message": {"enabled": True}}}},
    )
    assert len(enabled) == 1
    assert isinstance(enabled[0], StructuredTool)
    assert enabled[0].name == "echo_message"


def test_expand_plugin_tools_default_on_without_agent_config() -> None:
    """Octop default-on expansion makes tools bind without an agent opt-in."""
    from octop.infra.agents.plugins.plugin_tool_defaults import expand_plugin_tools_default_on

    load_plugin_dir(_FIXTURE, install_deps=False)
    expanded = expand_plugin_tools_default_on(
        {},
        registered_tools=[("echo-tool", "echo_message")],
        global_plugins={},
    )
    tools = build_plugin_tools(agent_plugins=expanded, global_plugins={})
    assert len(tools) == 1
    assert tools[0].name == "echo_message"

    still_off = build_plugin_tools(
        agent_plugins=expand_plugin_tools_default_on(
            {},
            registered_tools=[("echo-tool", "echo_message")],
            global_plugins={"echo-tool": False},
        ),
        global_plugins={"echo-tool": False},
    )
    assert still_off == []

    # Explicit opt-out still wins.
    opted_out = expand_plugin_tools_default_on(
        {"echo-tool": {"tools": {"echo_message": {"enabled": False}}}},
        registered_tools=[("echo-tool", "echo_message")],
    )
    assert build_plugin_tools(agent_plugins=opted_out) == []


def test_collect_plugin_tool_configs() -> None:
    cfg = collect_plugin_tool_configs(
        {
            "echo-tool": {
                "tools": {
                    "echo_message": {
                        "enabled": True,
                        "config": {"prefix": ">> "},
                    },
                },
            },
        },
    )
    assert cfg == {"echo_message": {"prefix": ">> "}}


def _make_tool(name: str, description: str = "demo tool") -> StructuredTool:
    def _fn(message: str) -> str:
        return message

    return StructuredTool.from_function(func=_fn, name=name, description=description)


def test_sanitize_ascii_name_passthrough() -> None:
    tool = _make_tool("echo_message", "original description")
    result = sanitize_plugin_tool_names([tool])
    assert result[0].name == "echo_message"
    assert result[0].description == "original description"


def test_sanitize_chinese_name_transliterates_to_pinyin() -> None:
    pytest.importorskip("pypinyin")
    tool = _make_tool("天气查询", "查询指定城市的天气")
    result = sanitize_plugin_tool_names([tool])
    assert result[0].name == "tianqichaxun"
    assert result[0].description == "[原名: 天气查询] 查询指定城市的天气"


def test_sanitize_mixed_name_keeps_ascii_parts() -> None:
    pytest.importorskip("pypinyin")
    assert sanitize_plugin_tool_name("获取weather信息") == "huoquweatherxinxi"


def test_sanitize_collision_gets_suffix() -> None:
    first = sanitize_plugin_tool_name("天气查询")
    second = sanitize_plugin_tool_name("天气查询", used={first})
    assert second != first
    assert second == f"{first}_2"
    # A legal name that is already reserved also gets deduped.
    assert sanitize_plugin_tool_name("echo_message", used={"echo_message"}) == "echo_message_2"


def test_sanitize_truncates_overlong_names() -> None:
    long_name = "很" * 80
    sanitized = sanitize_plugin_tool_name(long_name)
    assert len(sanitized) <= 64


def test_sanitize_clamps_prefixed_mcp_tool_name() -> None:
    # Remote MCP prefixes `{server}_{tool}`; tencent-docs + ULID + this tool is 66.
    name = "tencent-docs__01ARZ3NDEKTSV4RRFFQ69G5FAV_create_smartcanvas_by_mdx"
    assert len(name) > 64
    assert len("create_smartcanvas_by_mdx") <= 64
    sanitized = sanitize_plugin_tool_name(name)
    assert len(sanitized) <= 64
    assert sanitized.isascii()
    assert all(ch.isalnum() or ch in "_-" for ch in sanitized)
    tool = _make_tool(name, "create a smart canvas")
    sanitize_plugin_tool_names([tool])
    assert len(tool.name) <= 64
    assert "[原名: " in tool.description


def test_sanitize_dedupe_keeps_legal_names_within_limit() -> None:
    base = "a" * 64
    first = sanitize_plugin_tool_name(base)
    second = sanitize_plugin_tool_name(base, used={first})
    assert first == base
    assert second != first
    assert len(second) <= 64
    assert second.endswith("_2")


def test_sanitize_falls_back_to_underscores_without_pypinyin(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    import builtins

    real_import = builtins.__import__

    def _blocked(name: str, *args: Any, **kwargs: Any) -> Any:
        if name == "pypinyin" or name.startswith("pypinyin."):
            raise ImportError(name)
        return real_import(name, *args, **kwargs)

    monkeypatch.setattr(builtins, "__import__", _blocked)
    sanitized = sanitize_plugin_tool_name("天气查询")
    assert sanitized
    assert sanitized.isascii()
    assert all(ch.isalnum() or ch in "_-" for ch in sanitized)


def test_sanitize_plugin_tools_reserved_names() -> None:
    tool = _make_tool("echo_message")
    sanitize_plugin_tool_names([tool], reserved={"echo_message"})
    assert tool.name == "echo_message_2"


def test_build_plugin_tools_then_sanitize_keeps_config_keys_original() -> None:
    pytest.importorskip("pypinyin")
    from octop_harness.plugins.manifest import PluginManifest
    from octop_harness.plugins.registry import LoadedPlugin, ToolRegistration

    manifest = PluginManifest(id="demo", version="1.0.0", name="Demo", kind="tool", entry="main.py")
    PluginRegistry().register(
        LoadedPlugin(
            manifest=manifest,
            source_path=Path("."),
            tools=[
                ToolRegistration(
                    plugin_id="demo",
                    name="发送邮件",
                    fn=lambda to: to,
                    description="发送一封邮件",
                ),
                ToolRegistration(
                    plugin_id="demo",
                    name="echo_message",
                    fn=lambda text: text,
                    description="echo",
                ),
            ],
        ),
    )
    tools = build_plugin_tools(
        agent_plugins={
            "demo": {
                "tools": {
                    "发送邮件": {"enabled": True},
                    "echo_message": {"enabled": True},
                },
            },
        },
    )
    sanitized = sanitize_plugin_tool_names(tools)
    names = {t.name for t in sanitized}
    assert "fasongyoujian" in names  # pinyin of 发送邮件
    assert "echo_message" in names
    descriptions = {t.name: t.description for t in sanitized}
    assert descriptions["fasongyoujian"].startswith("[原名: 发送邮件]")
    # Config lookup still keyed by the original plugin-side name.
    assert collect_plugin_tool_configs(
        {"demo": {"tools": {"发送邮件": {"enabled": True, "config": {"a": 1}}}}},
    ) == {"发送邮件": {"a": 1}}