games-development-ai/cheap-worker/tests/test_planning_team_tools_patch.py

"""planning_team_tools_patch 单测:补丁后工具面无 Planning/Team/Schedule 三组、幂等、其余工具/skills 不受伤。
跑:cheap-worker/.venv/bin/python -m pytest cheap-worker/tests/test_planning_team_tools_patch.py -v

【为何不直接跑真 get_toolkit】get_toolkit / Team 工具类 / _chat 全在 agentscope.app 命名空间下,import 会牵
apscheduler([service] extra);本机 venv 无 apscheduler(真门只在 mini-desktop)。故核心三测用**真 Toolkit +
真 Planning 工具类 + Team/Schedule 替身(按 2.0.2 固定类名造 .name)**验过滤逻辑——走的是与生产同一份
_filter_toolkit / _install_get_toolkit_wrapper 代码路径;真 _chat.get_toolkit 的包装装配另有一枚 importorskip
守卫的集成测试(本机跳过、mini-desktop 全跑)。
"""

import asyncio
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))  # → cheap-worker/
import _bootstrap  # noqa: E402,F401  仅加 sys.path(import 时不取 key、不触网)
from agentscope.tool import (  # noqa: E402
    TaskCreate,
    TaskGet,
    TaskList,
    TaskUpdate,
    ToolGroup,
    Toolkit,
)
from agentscope.tool._base import ToolBase  # noqa: E402
from planning_team_tools_patch import (  # noqa: E402
    _PLANNING_TOOL_NAMES,
    _SCHEDULE_TOOL_NAMES,
    _TEAM_TOOL_NAMES,
    _filter_toolkit,
    _install_get_toolkit_wrapper,
)

# 便宜档六工具与后台控制工具的代表名(不在收窄范围,必须原样保留)。
_KEEP_NAMES = ("read_source", "write_source", "check_and_build", "run_gates", "read_verdict", "finish", "ToolStop")


class _FakeTool(ToolBase):
    """测试替身工具:过滤逻辑只看 .name,故替身只需一个 name(Team/Schedule/六工具在此仅取名验过滤)。"""

    def __init__(self, name: str) -> None:
        self.name = name
        self.description = f"fake {name}"
        self.input_schema = {"type": "object", "properties": {}}
        self.is_concurrency_safe = True
        self.is_read_only = True

    async def check_permissions(self, tool_input, context):  # noqa: ANN001 —— 替身不会被真调用
        raise NotImplementedError


def _build_framework_like_toolkit() -> Toolkit:
    """仿 2.0.2 get_toolkit 产物结构:basic 组 = 真 Planning[4] + Team 替身[4] + ToolStop 替身 + 六工具替身;
    外挂一个 schedule_tools 组(Schedule 替身[4])。"""
    planning = [TaskCreate(), TaskList(), TaskGet(), TaskUpdate()]  # 真 Planning 类:坐实 .name 契约
    team = [_FakeTool(n) for n in _TEAM_TOOL_NAMES]
    background = [_FakeTool("ToolStop")]  # 后台控制,不在收窄范围
    cheap_six = [_FakeTool(n) for n in ("read_source", "write_source", "check_and_build", "run_gates", "read_verdict", "finish")]
    schedule_group = ToolGroup(
        name="schedule_tools",
        description="cron schedule tools",
        tools=[_FakeTool(n) for n in _SCHEDULE_TOOL_NAMES],
    )
    return Toolkit(tools=planning + team + background + cheap_six, tool_groups=[schedule_group])


def _surface_names(tk: Toolkit) -> set:
    """取工具面全部工具名(遍历所有组)。"""
    return {t.name for g in tk.tool_groups for t in g.tools}


def test_filter_removes_planning_team_schedule():
    # 补丁后工具面无 Planning/Team/Schedule 三组(既无其工具名,schedule 更是整组被移除)。
    tk = _filter_toolkit(_build_framework_like_toolkit())
    names = _surface_names(tk)
    assert not (names & _PLANNING_TOOL_NAMES), f"Planning 应被剔除,残留:{names & _PLANNING_TOOL_NAMES}"
    assert not (names & _TEAM_TOOL_NAMES), f"Team 应被剔除,残留:{names & _TEAM_TOOL_NAMES}"
    assert not (names & _SCHEDULE_TOOL_NAMES), f"Schedule 应被剔除,残留:{names & _SCHEDULE_TOOL_NAMES}"
    assert not any(g.name == "schedule_tools" for g in tk.tool_groups), "schedule_tools 组应整组移除"


def test_filter_preserves_other_tools_and_skills():
    # 工具面其余成员不受伤:六工具 + 后台 ToolStop 全在席,basic 组结构仍在。
    tk = _filter_toolkit(_build_framework_like_toolkit())
    names = _surface_names(tk)
    for n in _KEEP_NAMES:
        assert n in names, f"非收窄工具 {n} 不应被误伤,实际工具面:{sorted(names)}"
    assert any(g.name == "basic" for g in tk.tool_groups), "重建后应仍保留 basic 组"


def test_filter_idempotent():
    # 幂等:对已收窄的 Toolkit 再过滤一次,工具面稳定不变(重复 apply 无害的过滤层证据)。
    once = _filter_toolkit(_build_framework_like_toolkit())
    twice = _filter_toolkit(once)
    assert _surface_names(once) == _surface_names(twice), "对已收窄 Toolkit 再过滤应稳定不变"
    assert not (_surface_names(twice) & (_PLANNING_TOOL_NAMES | _TEAM_TOOL_NAMES | _SCHEDULE_TOOL_NAMES))


def test_install_wrapper_filters_and_is_idempotent():
    # 用假 chat 模块验:包装真的「调原 → 过滤」,且重复 install 不再包层(装配层幂等,与 ws patch 同款)。
    class _FakeChatMod:
        pass

    fake = _FakeChatMod()
    built = _build_framework_like_toolkit()

    async def _orig_get_toolkit(**kwargs):  # 仿 get_toolkit:async、返回未过滤 Toolkit
        return built

    fake.get_toolkit = _orig_get_toolkit
    _install_get_toolkit_wrapper(fake)
    wrapped = fake.get_toolkit
    assert wrapped is not _orig_get_toolkit, "install 后 get_toolkit 应被替换为包装函数"
    assert getattr(wrapped, "_cheap_ptl_patched", False), "包装函数应带幂等标记"

    # 包装确实过滤了三组
    out = asyncio.run(fake.get_toolkit())
    assert not (_surface_names(out) & (_PLANNING_TOOL_NAMES | _TEAM_TOOL_NAMES | _SCHEDULE_TOOL_NAMES)), "包装应过滤三组"
    assert "finish" in _surface_names(out), "六工具应保留"

    # 幂等:第二次 install 短路,不再包一层
    _install_get_toolkit_wrapper(fake)
    assert fake.get_toolkit is wrapped, "幂等:重复 install 不应替换包装函数"


def test_apply_on_real_chat_module_if_available():
    # 集成:真 _chat.get_toolkit 被包装 + 幂等。需 apscheduler(agentscope.app 牵),本机无则跳过(真跑在 mini-desktop)。
    import pytest  # noqa: PLC0415

    pytest.importorskip("apscheduler", reason="agentscope.app 需 [service] extra(apscheduler);真跑在 mini-desktop")
    from agentscope.app._service import _chat as chat_mod  # noqa: PLC0415

    from planning_team_tools_patch import apply_planning_team_tools_patch  # noqa: PLC0415

    apply_planning_team_tools_patch()
    assert getattr(chat_mod.get_toolkit, "_cheap_ptl_patched", False), "真 _chat.get_toolkit 应被包装"
    wrapped = chat_mod.get_toolkit
    apply_planning_team_tools_patch()  # 再来一次
    assert chat_mod.get_toolkit is wrapped, "幂等:重复 apply 不换包装"