"""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_builtin_powershell_helpers_hide_the_console_window(monkeypatch): calls: list[dict[str, object]] = [] class Result: stdout = "done" stderr = "" returncode = 0 def fake_run(*_args, **kwargs): calls.append(kwargs) return Result() monkeypatch.setattr(builtin_tools.subprocess, "run", fake_run) monkeypatch.setattr(builtin_tools.subprocess, "CREATE_NO_WINDOW", 42, raising=False) assert builtin_tools._run_powershell("Write-Output done") == "done" assert calls[0]["creationflags"] == 42 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"])