wehub-resource-sync
426 lines
15 KiB
Python
426 lines
15 KiB
Python
"""Tests for built-in tools."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from openharness.tools.bash_tool import BashTool, BashToolInput
|
|
from openharness.tools.base import ToolExecutionContext
|
|
from openharness.tools.brief_tool import BriefTool, BriefToolInput
|
|
from openharness.tools.cron_create_tool import CronCreateTool, CronCreateToolInput
|
|
from openharness.tools.cron_delete_tool import CronDeleteTool, CronDeleteToolInput
|
|
from openharness.tools.cron_list_tool import CronListTool, CronListToolInput
|
|
from openharness.tools.config_tool import ConfigTool, ConfigToolInput
|
|
from openharness.tools.enter_worktree_tool import EnterWorktreeTool, EnterWorktreeToolInput
|
|
from openharness.tools.exit_worktree_tool import ExitWorktreeTool, ExitWorktreeToolInput
|
|
from openharness.tools.file_edit_tool import FileEditTool, FileEditToolInput
|
|
from openharness.tools.file_read_tool import FileReadTool, FileReadToolInput
|
|
from openharness.tools.file_write_tool import FileWriteTool, FileWriteToolInput
|
|
from openharness.tools.glob_tool import GlobTool, GlobToolInput
|
|
from openharness.tools.grep_tool import GrepTool, GrepToolInput
|
|
from openharness.tools.lsp_tool import LspTool, LspToolInput
|
|
from openharness.tools.notebook_edit_tool import NotebookEditTool, NotebookEditToolInput
|
|
from openharness.tools.remote_trigger_tool import RemoteTriggerTool, RemoteTriggerToolInput
|
|
from openharness.tools.skill_tool import SkillTool, SkillToolInput
|
|
from openharness.tools.todo_write_tool import TodoWriteTool, TodoWriteToolInput
|
|
from openharness.tools.tool_search_tool import ToolSearchTool, ToolSearchToolInput
|
|
from openharness.tools import create_default_tool_registry
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_file_write_read_and_edit(tmp_path: Path):
|
|
context = ToolExecutionContext(cwd=tmp_path)
|
|
|
|
write_result = await FileWriteTool().execute(
|
|
FileWriteToolInput(path="notes.txt", content="one\ntwo\nthree\n"),
|
|
context,
|
|
)
|
|
assert write_result.is_error is False
|
|
assert (tmp_path / "notes.txt").exists()
|
|
|
|
read_result = await FileReadTool().execute(
|
|
FileReadToolInput(path="notes.txt", offset=1, limit=2),
|
|
context,
|
|
)
|
|
assert "2\ttwo" in read_result.output
|
|
assert "3\tthree" in read_result.output
|
|
|
|
edit_result = await FileEditTool().execute(
|
|
FileEditToolInput(path="notes.txt", old_str="two", new_str="TWO"),
|
|
context,
|
|
)
|
|
assert edit_result.is_error is False
|
|
assert "TWO" in (tmp_path / "notes.txt").read_text(encoding="utf-8")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_file_write_requests_edit_approval_and_reports_diff_stats(tmp_path: Path):
|
|
approvals: list[tuple[str, str, int, int]] = []
|
|
|
|
async def _approve(path: str, diff: str, added: int, removed: int) -> str:
|
|
approvals.append((path, diff, added, removed))
|
|
return "once"
|
|
|
|
result = await FileWriteTool().execute(
|
|
FileWriteToolInput(path="notes.txt", content="one\ntwo\n"),
|
|
ToolExecutionContext(cwd=tmp_path, metadata={"edit_approval_prompt": _approve}),
|
|
)
|
|
|
|
assert result.is_error is False
|
|
assert (tmp_path / "notes.txt").read_text(encoding="utf-8") == "one\ntwo\n"
|
|
assert len(approvals) == 1
|
|
path, diff, added, removed = approvals[0]
|
|
assert path == str(tmp_path / "notes.txt")
|
|
assert added == 2
|
|
assert removed == 0
|
|
assert "@@" in diff
|
|
assert "+one" in diff
|
|
assert "+two" in diff
|
|
assert "\033[32m+2\033[0m" in result.output
|
|
assert "\033[31m-0\033[0m" in result.output
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_file_write_rejection_does_not_create_parent_directories(tmp_path: Path):
|
|
async def _reject(path: str, diff: str, added: int, removed: int) -> str:
|
|
del path, diff, added, removed
|
|
return "reject"
|
|
|
|
result = await FileWriteTool().execute(
|
|
FileWriteToolInput(path="nested/notes.txt", content="draft\n"),
|
|
ToolExecutionContext(cwd=tmp_path, metadata={"edit_approval_prompt": _reject}),
|
|
)
|
|
|
|
assert result.is_error is True
|
|
assert "Write rejected by user" in result.output
|
|
assert not (tmp_path / "nested").exists()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_file_edit_rejects_when_edit_approval_denied(tmp_path: Path):
|
|
target = tmp_path / "notes.txt"
|
|
target.write_text("one\ntwo\n", encoding="utf-8")
|
|
approvals: list[tuple[str, str, int, int]] = []
|
|
|
|
async def _reject(path: str, diff: str, added: int, removed: int) -> str:
|
|
approvals.append((path, diff, added, removed))
|
|
return "reject"
|
|
|
|
result = await FileEditTool().execute(
|
|
FileEditToolInput(path="notes.txt", old_str="two", new_str="TWO"),
|
|
ToolExecutionContext(cwd=tmp_path, metadata={"edit_approval_prompt": _reject}),
|
|
)
|
|
|
|
assert result.is_error is True
|
|
assert "Edit rejected by user" in result.output
|
|
assert target.read_text(encoding="utf-8") == "one\ntwo\n"
|
|
assert len(approvals) == 1
|
|
path, diff, added, removed = approvals[0]
|
|
assert path == str(target)
|
|
assert added == 1
|
|
assert removed == 1
|
|
assert "-two" in diff
|
|
assert "+TWO" in diff
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_glob_and_grep(tmp_path: Path):
|
|
context = ToolExecutionContext(cwd=tmp_path)
|
|
(tmp_path / "a.py").write_text("def alpha():\n return 1\n", encoding="utf-8")
|
|
(tmp_path / "b.py").write_text("def beta():\n return 2\n", encoding="utf-8")
|
|
|
|
glob_result = await GlobTool().execute(GlobToolInput(pattern="*.py"), context)
|
|
assert glob_result.output.splitlines() == ["a.py", "b.py"]
|
|
|
|
aliased_input = GlobTool().input_model.model_validate({"path": "*.py"})
|
|
aliased_glob_result = await GlobTool().execute(aliased_input, context)
|
|
assert aliased_glob_result.output.splitlines() == ["a.py", "b.py"]
|
|
|
|
grep_result = await GrepTool().execute(
|
|
GrepToolInput(pattern=r"def\s+beta", file_glob="*.py"),
|
|
context,
|
|
)
|
|
assert "b.py:1:def beta():" in grep_result.output
|
|
|
|
file_root_result = await GrepTool().execute(
|
|
GrepToolInput(pattern=r"def\s+alpha", root="a.py"),
|
|
context,
|
|
)
|
|
assert "a.py:1:def alpha():" in file_root_result.output
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_glob_tool_accepts_absolute_patterns(tmp_path: Path, monkeypatch):
|
|
monkeypatch.setattr("openharness.tools.glob_tool.shutil.which", lambda _: None)
|
|
context = ToolExecutionContext(cwd=tmp_path.parent)
|
|
nested = tmp_path / "pkg"
|
|
nested.mkdir()
|
|
(nested / "a.py").write_text("print('a')\n", encoding="utf-8")
|
|
(nested / "b.txt").write_text("b\n", encoding="utf-8")
|
|
|
|
result = await GlobTool().execute(
|
|
GlobToolInput(pattern=str(tmp_path / "**" / "*.py")),
|
|
context,
|
|
)
|
|
|
|
assert result.is_error is False
|
|
assert result.output.replace("\\", "/").splitlines() == ["pkg/a.py"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bash_tool_runs_command(tmp_path: Path):
|
|
result = await BashTool().execute(
|
|
BashToolInput(command="printf 'hello'"),
|
|
ToolExecutionContext(cwd=tmp_path),
|
|
)
|
|
assert result.is_error is False
|
|
assert result.output == "hello"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tool_search_and_brief_tools(tmp_path: Path):
|
|
registry = create_default_tool_registry()
|
|
context = ToolExecutionContext(cwd=tmp_path, metadata={"tool_registry": registry})
|
|
|
|
search_result = await ToolSearchTool().execute(
|
|
ToolSearchToolInput(query="file"),
|
|
context,
|
|
)
|
|
assert "read_file" in search_result.output
|
|
|
|
brief_result = await BriefTool().execute(
|
|
BriefToolInput(text="abcdefghijklmnopqrstuvwxyz", max_chars=20),
|
|
ToolExecutionContext(cwd=tmp_path),
|
|
)
|
|
assert brief_result.output == "abcdefghijklmnopqrst..."
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_skill_todo_and_config_tools(tmp_path: Path, monkeypatch):
|
|
monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config"))
|
|
skills_dir = tmp_path / "config" / "skills"
|
|
skills_dir.mkdir(parents=True)
|
|
pytest_dir = skills_dir / "pytest"
|
|
pytest_dir.mkdir()
|
|
(pytest_dir / "SKILL.md").write_text("# Pytest\nHelpful pytest notes.\n", encoding="utf-8")
|
|
|
|
skill_result = await SkillTool().execute(
|
|
SkillToolInput(name="Pytest"),
|
|
ToolExecutionContext(cwd=tmp_path),
|
|
)
|
|
assert "Helpful pytest notes." in skill_result.output
|
|
|
|
todo_result = await TodoWriteTool().execute(
|
|
TodoWriteToolInput(item="wire commands"),
|
|
ToolExecutionContext(cwd=tmp_path),
|
|
)
|
|
assert todo_result.is_error is False
|
|
assert "wire commands" in (tmp_path / "TODO.md").read_text(encoding="utf-8")
|
|
|
|
config_result = await ConfigTool().execute(
|
|
ConfigToolInput(action="set", key="theme", value="solarized"),
|
|
ToolExecutionContext(cwd=tmp_path),
|
|
)
|
|
assert config_result.output == "Updated theme"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_skill_tool_rejects_user_only_skills(tmp_path: Path, monkeypatch):
|
|
monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config"))
|
|
skills_dir = tmp_path / "config" / "skills"
|
|
skills_dir.mkdir(parents=True)
|
|
deploy_dir = skills_dir / "deploy"
|
|
deploy_dir.mkdir()
|
|
(deploy_dir / "SKILL.md").write_text(
|
|
"---\n"
|
|
"description: User-only deploy workflow.\n"
|
|
"disable-model-invocation: true\n"
|
|
"---\n\n"
|
|
"# Deploy\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = await SkillTool().execute(
|
|
SkillToolInput(name="deploy"),
|
|
ToolExecutionContext(cwd=tmp_path),
|
|
)
|
|
|
|
assert result.is_error is True
|
|
assert "can only be invoked by the user as /deploy" in result.output
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_todo_write_upsert(tmp_path: Path):
|
|
tool = TodoWriteTool()
|
|
ctx = ToolExecutionContext(cwd=tmp_path)
|
|
|
|
await tool.execute(TodoWriteToolInput(item="task A"), ctx)
|
|
await tool.execute(TodoWriteToolInput(item="task B"), ctx)
|
|
|
|
# Marking done should update in-place, not append a duplicate
|
|
result = await tool.execute(TodoWriteToolInput(item="task A", checked=True), ctx)
|
|
assert result.is_error is False
|
|
|
|
content = (tmp_path / "TODO.md").read_text(encoding="utf-8")
|
|
assert content.count("task A") == 1
|
|
assert "- [x] task A" in content
|
|
assert "- [ ] task A" not in content
|
|
assert "- [ ] task B" in content
|
|
|
|
# Calling again with same state is a no-op
|
|
noop = await tool.execute(TodoWriteToolInput(item="task A", checked=True), ctx)
|
|
assert "No change" in noop.output
|
|
assert (tmp_path / "TODO.md").read_text(encoding="utf-8").count("task A") == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_notebook_edit_tool(tmp_path: Path):
|
|
result = await NotebookEditTool().execute(
|
|
NotebookEditToolInput(path="demo.ipynb", cell_index=0, new_source="print('nb ok')\n"),
|
|
ToolExecutionContext(cwd=tmp_path),
|
|
)
|
|
assert result.is_error is False
|
|
assert "demo.ipynb" in result.output
|
|
assert "nb ok" in (tmp_path / "demo.ipynb").read_text(encoding="utf-8")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_lsp_tool(tmp_path: Path):
|
|
(tmp_path / "pkg").mkdir()
|
|
(tmp_path / "pkg" / "utils.py").write_text(
|
|
'def greet(name):\n """Return a greeting."""\n return f"hi {name}"\n',
|
|
encoding="utf-8",
|
|
)
|
|
(tmp_path / "pkg" / "app.py").write_text(
|
|
"from pkg.utils import greet\n\nprint(greet('world'))\n",
|
|
encoding="utf-8",
|
|
)
|
|
context = ToolExecutionContext(cwd=tmp_path)
|
|
|
|
document_symbols = await LspTool().execute(
|
|
LspToolInput(operation="document_symbol", file_path="pkg/utils.py"),
|
|
context,
|
|
)
|
|
assert "function greet" in document_symbols.output
|
|
|
|
definition = await LspTool().execute(
|
|
LspToolInput(operation="go_to_definition", file_path="pkg/app.py", symbol="greet"),
|
|
context,
|
|
)
|
|
assert "pkg/utils.py:1:1" in definition.output.replace("\\", "/")
|
|
|
|
references = await LspTool().execute(
|
|
LspToolInput(operation="find_references", file_path="pkg/app.py", symbol="greet"),
|
|
context,
|
|
)
|
|
assert "pkg/app.py:1:from pkg.utils import greet" in references.output.replace("\\", "/")
|
|
|
|
hover = await LspTool().execute(
|
|
LspToolInput(operation="hover", file_path="pkg/app.py", symbol="greet"),
|
|
context,
|
|
)
|
|
assert "Return a greeting." in hover.output
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_worktree_tools(tmp_path: Path):
|
|
subprocess.run(["git", "init"], cwd=tmp_path, check=True, capture_output=True, text=True)
|
|
subprocess.run(
|
|
["git", "config", "user.email", "openharness@example.com"],
|
|
cwd=tmp_path,
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
subprocess.run(
|
|
["git", "config", "user.name", "OpenHarness Tests"],
|
|
cwd=tmp_path,
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
(tmp_path / "demo.txt").write_text("hello\n", encoding="utf-8")
|
|
subprocess.run(["git", "add", "-A"], cwd=tmp_path, check=True, capture_output=True, text=True)
|
|
subprocess.run(
|
|
["git", "commit", "-m", "init"],
|
|
cwd=tmp_path,
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
|
|
enter_result = await EnterWorktreeTool().execute(
|
|
EnterWorktreeToolInput(branch="feature/demo"),
|
|
ToolExecutionContext(cwd=tmp_path),
|
|
)
|
|
assert enter_result.is_error is False
|
|
worktree_path = Path(enter_result.output.split("Path: ", 1)[1].strip())
|
|
assert worktree_path.exists()
|
|
|
|
exit_result = await ExitWorktreeTool().execute(
|
|
ExitWorktreeToolInput(path=str(worktree_path)),
|
|
ToolExecutionContext(cwd=tmp_path),
|
|
)
|
|
assert exit_result.is_error is False
|
|
assert not worktree_path.exists()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cron_and_remote_trigger_tools(tmp_path: Path, monkeypatch):
|
|
monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data"))
|
|
context = ToolExecutionContext(cwd=tmp_path)
|
|
|
|
create_result = await CronCreateTool().execute(
|
|
CronCreateToolInput(
|
|
name="nightly",
|
|
schedule="0 0 * * *",
|
|
command="printf 'CRON_OK'",
|
|
notify={"type": "feishu_dm", "user_open_id": "ou_test"},
|
|
),
|
|
context,
|
|
)
|
|
assert create_result.is_error is False
|
|
|
|
list_result = await CronListTool().execute(CronListToolInput(), context)
|
|
assert "nightly" in list_result.output
|
|
assert "feishu_dm" in list_result.output
|
|
|
|
trigger_result = await RemoteTriggerTool().execute(
|
|
RemoteTriggerToolInput(name="nightly"),
|
|
context,
|
|
)
|
|
assert trigger_result.is_error is False
|
|
assert "CRON_OK" in trigger_result.output
|
|
|
|
delete_result = await CronDeleteTool().execute(
|
|
CronDeleteToolInput(name="nightly"),
|
|
context,
|
|
)
|
|
assert delete_result.is_error is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cron_create_agent_turn_payload(tmp_path: Path, monkeypatch):
|
|
monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data"))
|
|
context = ToolExecutionContext(cwd=tmp_path)
|
|
|
|
create_result = await CronCreateTool().execute(
|
|
CronCreateToolInput(
|
|
name="daily-summary",
|
|
schedule="0 18 * * *",
|
|
timezone="Asia/Hong_Kong",
|
|
message="check GitHub",
|
|
payload={"deliver": True, "channel": "feishu", "to": "ou_test"},
|
|
),
|
|
context,
|
|
)
|
|
assert create_result.is_error is False
|
|
|
|
list_result = await CronListTool().execute(CronListToolInput(), context)
|
|
assert "daily-summary" in list_result.output
|
|
assert "Asia/Hong_Kong" in list_result.output
|
|
assert "payload: agent_turn -> feishu:ou_test" in list_result.output
|