mulit-agent/mulit_agent/tests/test_builtin_tools.py

"""The preset computer tools must work on a freshly installed Agent."""

from datetime import UTC, datetime, timedelta, timezone
from threading import Event
from zoneinfo import ZoneInfo

import pytest

from mulit_agent_app.application.task_executor import TaskExecutor
from mulit_agent_app.infrastructure import builtin_tools
from mulit_agent_app.infrastructure.builtin_tools import builtin_tools as specs
from mulit_agent_app.infrastructure.database import AgentDatabase
from mulit_agent_app.infrastructure.tool_registry import Toolbox

_HANDLERS = {spec.name: spec.handler for spec in specs()}


def test_preset_tool_set_covers_file_open_time_process_screenshot_and_clipboard():
    names = set(_HANDLERS)
    assert {
        "computer_list_directory",
        "computer_read_text_file",
        "computer_write_text_file",
        "computer_create_directory",
        "computer_move_path",
        "computer_delete_path",
        "computer_open",
        "computer_time",
        "computer_list_processes",
        "computer_screenshot",
        "computer_read_clipboard",
        "computer_write_clipboard",
    } <= names
    for spec in specs():
        assert spec.description and spec.parameters["type"] == "object"


def test_file_tools_round_trip_inside_a_folder(tmp_path):
    folder = tmp_path / "work"
    assert "已创建文件夹" in _HANDLERS["computer_create_directory"]({"path": str(folder)})
    assert "已写入" in _HANDLERS["computer_write_text_file"]({"path": str(folder / "note.txt"), "content": "第一行\n第二行"})

    assert "第一行" in _HANDLERS["computer_read_text_file"]({"path": str(folder / "note.txt")})
    listing = _HANDLERS["computer_list_directory"]({"path": str(folder)})
    assert "note.txt" in listing and "[文件]" in listing

    moved = folder / "moved.txt"
    assert "已移动" in _HANDLERS["computer_move_path"]({"source": str(folder / "note.txt"), "destination": str(moved)})
    assert _HANDLERS["computer_delete_path"]({"path": str(moved)}).startswith("已删除")


def test_file_tools_refuse_dangerous_or_accidental_overwrites(tmp_path):
    existing = tmp_path / "keep.txt"
    existing.write_text("原内容", encoding="utf-8")

    with pytest.raises(ValueError, match="overwrite=true"):
        _HANDLERS["computer_write_text_file"]({"path": str(existing), "content": "新内容"})
    assert existing.read_text(encoding="utf-8") == "原内容"
    _HANDLERS["computer_write_text_file"]({"path": str(existing), "content": "新内容", "overwrite": True})
    assert existing.read_text(encoding="utf-8") == "新内容"

    with pytest.raises(ValueError, match="不是可读取的文件"):
        _HANDLERS["computer_read_text_file"]({"path": str(tmp_path / "missing.txt")})

    folder = tmp_path / "nested"
    (folder / "child").mkdir(parents=True)
    (folder / "child" / "file.txt").write_text("x", encoding="utf-8")
    with pytest.raises(ValueError, match="recursive=true"):
        _HANDLERS["computer_delete_path"]({"path": str(folder)})
    assert _HANDLERS["computer_delete_path"]({"path": str(folder), "recursive": True}).startswith("已删除")

    with pytest.raises(ValueError, match="拒绝删除"):
        _HANDLERS["computer_delete_path"]({"path": "C:\\Windows", "recursive": True})


def test_time_tool_accepts_local_offsets_and_iana_zones():
    assert "星期" in _HANDLERS["computer_time"]({})

    for requested in ("Asia/Tokyo", "UTC", "UTC+9"):
        before = datetime.now(_zone(requested)).strftime("%Y-%m-%d %H")
        reported = _HANDLERS["computer_time"]({"timezone": requested})
        after = datetime.now(_zone(requested)).strftime("%Y-%m-%d %H")
        assert before <= reported[:13] <= after, f"{requested} 返回了错误时间：{reported}"

    with pytest.raises(ValueError, match="无法识别时区"):
        _HANDLERS["computer_time"]({"timezone": "Mars/Olympus"})


def _zone(requested: str):
    if requested == "UTC":
        return UTC
    if requested == "UTC+9":
        return timezone(timedelta(hours=9))
    return ZoneInfo(requested)


def test_open_tool_opens_urls_and_refuses_unknown_paths(monkeypatch):
    opened: list[str] = []
    monkeypatch.setattr(builtin_tools.webbrowser, "open", lambda target: opened.append(target) or True)

    assert "已用系统默认程序打开网址" in _HANDLERS["computer_open"]({"target": "https://example.com"})
    assert opened == ["https://example.com"]

    with pytest.raises(ValueError, match="只支持 http/https"):
        _HANDLERS["computer_open"]({"target": "ftp://example.com"})


def test_process_screenshot_and_clipboard_use_the_local_windows_session(tmp_path):
    processes = _HANDLERS["computer_list_processes"]({"limit": 5})
    assert "Name" in processes or "名称" in processes

    shot = _HANDLERS["computer_screenshot"]({"path": str(tmp_path / "shot.png")})
    assert "已保存屏幕截图" in shot
    assert (tmp_path / "shot.png").read_bytes()[:8] == b"\x89PNG\r\n\x1a\n"

    assert "已写入剪贴板" in _HANDLERS["computer_write_clipboard"]({"text": "MultiClaw 剪贴板测试"})
    assert "MultiClaw 剪贴板测试" in _HANDLERS["computer_read_clipboard"]({})


def test_toolbox_exposes_preset_tools_and_reports_builtin_execution(tmp_path):
    toolbox = Toolbox("task-1", None)
    names = {spec.name for spec in toolbox.specs()}
    assert "run_powershell" in names and "computer_time" in names

    assert "星期" in toolbox.call("computer_time", {"timezone": "UTC"})
    assert toolbox.used_builtin is True and toolbox.used_any_tool is True
    assert toolbox.call("computer_unknown", {}).startswith("工具调用被拒绝")
    assert toolbox.call("computer_read_text_file", {"path": "C:\\definitely\\missing.txt"}).startswith("工具调用失败")


def test_task_executor_marks_preset_tool_usage(tmp_path):
    database = AgentDatabase(tmp_path / "agent.sqlite3")

    def runner(_instruction, _attachments, toolbox):
        return f"本机时间：{toolbox.call('computer_time', {'timezone': 'UTC'})}"

    executor = TaskExecutor(database, runner)
    completed = Event()
    events: list[dict[str, object]] = []
    executor.task_event.connect(
        lambda event: (events.append(event), completed.set())
        if isinstance(event, dict) and event.get("status") == "completed"
        else None
    )

    executor.submit("preset-task", "看看现在几点")
    assert completed.wait(10)
    assert events[-1]["execution"] == "builtin"
    assert "本机时间：" in str(events[-1]["output"])