422 lines
16 KiB
Python
422 lines
16 KiB
Python
"""Unit tests for StrReplaceEditorTool (U1, R1).
|
|
|
|
Covers happy path, edge cases, error/failure paths, path-security rejection,
|
|
and the integration contract that the tool is registered as a default core
|
|
tool in ReActEngine and exported from the tools package.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from agentkit.tools.str_replace_editor import StrReplaceEditorTool
|
|
|
|
|
|
# ── fixtures ──────────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.fixture
|
|
def workspace(tmp_path: Path) -> Path:
|
|
"""A clean workspace root directory for each test."""
|
|
return tmp_path
|
|
|
|
|
|
@pytest.fixture
|
|
def tool(workspace: Path) -> StrReplaceEditorTool:
|
|
return StrReplaceEditorTool(workspace_root=workspace)
|
|
|
|
|
|
# ── happy path ────────────────────────────────────────────────────────
|
|
|
|
|
|
async def test_create_writes_new_file(tool: StrReplaceEditorTool, workspace: Path) -> None:
|
|
result = await tool.execute(command="create", path="hello.py", file_text="print('hi')\n")
|
|
assert result["is_error"] is False
|
|
assert result["command"] == "create"
|
|
assert result["total_lines"] == 1
|
|
assert (workspace / "hello.py").read_text() == "print('hi')\n"
|
|
|
|
|
|
async def test_view_returns_content_with_line_numbers(
|
|
tool: StrReplaceEditorTool, workspace: Path
|
|
) -> None:
|
|
(workspace / "a.txt").write_text("alpha\nbeta\ngamma\n")
|
|
result = await tool.execute(command="view", path="a.txt")
|
|
assert result["is_error"] is False
|
|
assert result["total_lines"] == 3
|
|
assert result["start_line"] == 1
|
|
assert result["end_line"] == 3
|
|
# cat -n style: right-aligned number + tab.
|
|
assert result["content"] == " 1\talpha\n 2\tbeta\n 3\tgamma"
|
|
|
|
|
|
async def test_str_replace_replaces_unique_anchor(
|
|
tool: StrReplaceEditorTool, workspace: Path
|
|
) -> None:
|
|
(workspace / "f.txt").write_text("def foo():\n return 1\n")
|
|
result = await tool.execute(
|
|
command="str_replace",
|
|
path="f.txt",
|
|
old_str="return 1",
|
|
new_str="return 2",
|
|
)
|
|
assert result["is_error"] is False
|
|
assert (workspace / "f.txt").read_text() == "def foo():\n return 2\n"
|
|
|
|
|
|
async def test_insert_at_line_inserts_in_middle(
|
|
tool: StrReplaceEditorTool, workspace: Path
|
|
) -> None:
|
|
(workspace / "f.txt").write_text("line1\nline2\nline3\n")
|
|
result = await tool.execute(
|
|
command="insert_at_line", path="f.txt", insert_line=2, new_str="INSERTED"
|
|
)
|
|
assert result["is_error"] is False
|
|
assert (workspace / "f.txt").read_text() == "line1\nINSERTED\nline2\nline3\n"
|
|
|
|
|
|
# ── edge cases ────────────────────────────────────────────────────────
|
|
|
|
|
|
async def test_create_empty_file(tool: StrReplaceEditorTool, workspace: Path) -> None:
|
|
result = await tool.execute(command="create", path="empty.txt", file_text="")
|
|
assert result["is_error"] is False
|
|
assert result["total_lines"] == 0
|
|
assert (workspace / "empty.txt").read_text() == ""
|
|
# view of an empty file reports total_lines=0 with a note.
|
|
view = await tool.execute(command="view", path="empty.txt")
|
|
assert view["is_error"] is False
|
|
assert view["total_lines"] == 0
|
|
assert view["content"] == ""
|
|
assert view["note"] == "empty file"
|
|
|
|
|
|
async def test_str_replace_multiple_matches_is_error(
|
|
tool: StrReplaceEditorTool, workspace: Path
|
|
) -> None:
|
|
(workspace / "f.txt").write_text("x\nx\n")
|
|
result = await tool.execute(command="str_replace", path="f.txt", old_str="x", new_str="y")
|
|
assert result["is_error"] is True
|
|
assert "not unique" in result["error"]
|
|
# File is untouched on error.
|
|
assert (workspace / "f.txt").read_text() == "x\nx\n"
|
|
|
|
|
|
async def test_insert_at_line_zero_prepends(tool: StrReplaceEditorTool, workspace: Path) -> None:
|
|
(workspace / "f.txt").write_text("line1\nline2\n")
|
|
result = await tool.execute(
|
|
command="insert_at_line", path="f.txt", insert_line=0, new_str="TOP"
|
|
)
|
|
assert result["is_error"] is False
|
|
assert (workspace / "f.txt").read_text() == "TOP\nline1\nline2\n"
|
|
|
|
|
|
async def test_insert_at_line_beyond_eof_appends(
|
|
tool: StrReplaceEditorTool, workspace: Path
|
|
) -> None:
|
|
(workspace / "f.txt").write_text("line1\nline2\n")
|
|
result = await tool.execute(
|
|
command="insert_at_line", path="f.txt", insert_line=99, new_str="BOTTOM"
|
|
)
|
|
assert result["is_error"] is False
|
|
assert (workspace / "f.txt").read_text() == "line1\nline2\nBOTTOM\n"
|
|
|
|
|
|
async def test_insert_at_line_multiline_text(tool: StrReplaceEditorTool, workspace: Path) -> None:
|
|
(workspace / "f.txt").write_text("a\nb\n")
|
|
result = await tool.execute(
|
|
command="insert_at_line",
|
|
path="f.txt",
|
|
insert_line=2,
|
|
new_str="x\ny\nz",
|
|
)
|
|
assert result["is_error"] is False
|
|
assert (workspace / "f.txt").read_text() == "a\nx\ny\nz\nb\n"
|
|
|
|
|
|
async def test_view_with_line_range(tool: StrReplaceEditorTool, workspace: Path) -> None:
|
|
(workspace / "f.txt").write_text("one\ntwo\nthree\nfour\nfive\n")
|
|
result = await tool.execute(command="view", path="f.txt", start_line=2, end_line=4)
|
|
assert result["is_error"] is False
|
|
assert result["start_line"] == 2
|
|
assert result["end_line"] == 4
|
|
assert result["total_lines"] == 5
|
|
assert result["content"] == " 2\ttwo\n 3\tthree\n 4\tfour"
|
|
|
|
|
|
async def test_view_range_beyond_eof_returns_empty(
|
|
tool: StrReplaceEditorTool, workspace: Path
|
|
) -> None:
|
|
(workspace / "f.txt").write_text("only\n")
|
|
result = await tool.execute(command="view", path="f.txt", start_line=10, end_line=20)
|
|
assert result["is_error"] is False
|
|
assert result["content"] == ""
|
|
assert result["start_line"] == 10
|
|
|
|
|
|
# ── error and failure paths ───────────────────────────────────────────
|
|
|
|
|
|
async def test_create_refuses_overwrite(tool: StrReplaceEditorTool, workspace: Path) -> None:
|
|
(workspace / "f.txt").write_text("existing\n")
|
|
result = await tool.execute(command="create", path="f.txt", file_text="new\n")
|
|
assert result["is_error"] is True
|
|
assert "already exists" in result["error"]
|
|
# Original content preserved (data-loss guard).
|
|
assert (workspace / "f.txt").read_text() == "existing\n"
|
|
|
|
|
|
async def test_str_replace_anchor_not_found(tool: StrReplaceEditorTool, workspace: Path) -> None:
|
|
(workspace / "f.txt").write_text("hello world\n")
|
|
result = await tool.execute(
|
|
command="str_replace", path="f.txt", old_str="goodbye", new_str="hi"
|
|
)
|
|
assert result["is_error"] is True
|
|
assert "not found" in result["error"]
|
|
|
|
|
|
async def test_str_replace_empty_old_str_rejected(
|
|
tool: StrReplaceEditorTool, workspace: Path
|
|
) -> None:
|
|
(workspace / "f.txt").write_text("x\n")
|
|
result = await tool.execute(command="str_replace", path="f.txt", old_str="", new_str="y")
|
|
assert result["is_error"] is True
|
|
assert "old_str" in result["error"]
|
|
|
|
|
|
async def test_str_replace_on_missing_file(tool: StrReplaceEditorTool, workspace: Path) -> None:
|
|
result = await tool.execute(command="str_replace", path="nope.txt", old_str="a", new_str="b")
|
|
assert result["is_error"] is True
|
|
assert "not found" in result["error"].lower()
|
|
|
|
|
|
async def test_path_traversal_rejected(tool: StrReplaceEditorTool, workspace: Path) -> None:
|
|
result = await tool.execute(command="view", path="../../etc/passwd")
|
|
assert result["is_error"] is True
|
|
assert "rejected" in result["error"]
|
|
|
|
|
|
async def test_path_traversal_create_rejected(
|
|
tool: StrReplaceEditorTool, workspace: Path, tmp_path: Path
|
|
) -> None:
|
|
# Even if the target would resolve inside a sibling dir, `..` is rejected.
|
|
result = await tool.execute(command="create", path="../sibling.txt", file_text="x")
|
|
assert result["is_error"] is True
|
|
|
|
|
|
async def test_absolute_path_rejected(tool: StrReplaceEditorTool, workspace: Path) -> None:
|
|
# Absolute path to a real file outside the workspace.
|
|
result = await tool.execute(command="view", path="/etc/passwd")
|
|
assert result["is_error"] is True
|
|
assert "rejected" in result["error"]
|
|
|
|
|
|
async def test_absolute_path_inside_workspace_also_rejected(
|
|
tool: StrReplaceEditorTool, workspace: Path
|
|
) -> None:
|
|
# Absolute paths are rejected outright (force relative interpretation),
|
|
# even when the path would resolve inside the workspace.
|
|
target = workspace / "inside.txt"
|
|
target.write_text("ok\n")
|
|
result = await tool.execute(command="view", path=str(target))
|
|
assert result["is_error"] is True
|
|
assert "rejected" in result["error"]
|
|
|
|
|
|
async def test_symlink_escape_rejected(tmp_path: Path) -> None:
|
|
# Use a workspace SUBDIR of tmp_path so a file under tmp_path (but not
|
|
# under the workspace) counts as "outside the workspace".
|
|
workspace = tmp_path / "ws"
|
|
workspace.mkdir()
|
|
tool = StrReplaceEditorTool(workspace_root=workspace)
|
|
# Real secret file OUTSIDE the workspace (sibling, still under tmp_path).
|
|
outside = tmp_path / "secret.txt"
|
|
outside.write_text("top secret\n")
|
|
# Symlink inside the workspace pointing to the outside file.
|
|
link = workspace / "escape.txt"
|
|
os.symlink(outside, link)
|
|
# view through the symlink must be rejected (symlink escape).
|
|
result = await tool.execute(command="view", path="escape.txt")
|
|
assert result["is_error"] is True
|
|
assert "rejected" in result["error"]
|
|
# create through a symlink that escapes must also be rejected.
|
|
result2 = await tool.execute(
|
|
command="create", path="escape.txt", file_text="overwrite\n"
|
|
)
|
|
assert result2["is_error"] is True
|
|
# The outside file must NOT have been overwritten (data-loss guard).
|
|
assert outside.read_text() == "top secret\n"
|
|
|
|
|
|
async def test_symlink_to_inside_workspace_allowed(
|
|
tool: StrReplaceEditorTool, workspace: Path
|
|
) -> None:
|
|
# A symlink whose target is INSIDE the workspace is allowed (no escape).
|
|
real = workspace / "real.txt"
|
|
real.write_text("content\n")
|
|
link = workspace / "link.txt"
|
|
os.symlink(real, link)
|
|
result = await tool.execute(command="view", path="link.txt")
|
|
assert result["is_error"] is False
|
|
assert "content" in result["content"]
|
|
|
|
|
|
async def test_file_outside_workspace_rejected(tool: StrReplaceEditorTool, tmp_path: Path) -> None:
|
|
# A relative path that climbs out via `..` is rejected by the `..` rule,
|
|
# but also verify a nested traversal attempt is caught.
|
|
result = await tool.execute(command="view", path="sub/../../etc/passwd")
|
|
assert result["is_error"] is True
|
|
|
|
|
|
async def test_unknown_command_rejected(tool: StrReplaceEditorTool, workspace: Path) -> None:
|
|
result = await tool.execute(command="delete", path="f.txt")
|
|
assert result["is_error"] is True
|
|
assert "Unknown command" in result["error"]
|
|
|
|
|
|
async def test_missing_path_rejected(tool: StrReplaceEditorTool) -> None:
|
|
result = await tool.execute(command="view", path="")
|
|
assert result["is_error"] is True
|
|
assert "path" in result["error"].lower()
|
|
|
|
|
|
async def test_missing_file_text_rejected(tool: StrReplaceEditorTool, workspace: Path) -> None:
|
|
result = await tool.execute(command="create", path="f.txt")
|
|
assert result["is_error"] is True
|
|
assert "file_text" in result["error"]
|
|
|
|
|
|
async def test_missing_insert_line_rejected(tool: StrReplaceEditorTool, workspace: Path) -> None:
|
|
(workspace / "f.txt").write_text("a\n")
|
|
result = await tool.execute(command="insert_at_line", path="f.txt", new_str="b")
|
|
assert result["is_error"] is True
|
|
assert "insert_line" in result["error"]
|
|
|
|
|
|
async def test_insert_line_negative_rejected(tool: StrReplaceEditorTool, workspace: Path) -> None:
|
|
(workspace / "f.txt").write_text("a\n")
|
|
result = await tool.execute(command="insert_at_line", path="f.txt", insert_line=-1, new_str="b")
|
|
assert result["is_error"] is True
|
|
|
|
|
|
async def test_view_directory_rejected(tool: StrReplaceEditorTool, workspace: Path) -> None:
|
|
(workspace / "subdir").mkdir()
|
|
result = await tool.execute(command="view", path="subdir")
|
|
assert result["is_error"] is True
|
|
assert "directory" in result["error"].lower()
|
|
|
|
|
|
async def test_create_in_nested_subdir_creates_parents(
|
|
tool: StrReplaceEditorTool, workspace: Path
|
|
) -> None:
|
|
result = await tool.execute(
|
|
command="create",
|
|
path="nested/deep/file.txt",
|
|
file_text="deep\n",
|
|
)
|
|
assert result["is_error"] is False
|
|
assert (workspace / "nested" / "deep" / "file.txt").read_text() == "deep\n"
|
|
|
|
|
|
# ── integration contract ──────────────────────────────────────────────
|
|
|
|
|
|
def test_str_replace_editor_in_default_core_tools() -> None:
|
|
"""The tool must be a default core tool so its full description is
|
|
always injected into the LLM prompt (tiered injection)."""
|
|
from agentkit.core.react import ReActEngine
|
|
|
|
assert "str_replace_editor" in ReActEngine._DEFAULT_CORE_TOOLS
|
|
# The broken write_file placeholder must be gone.
|
|
assert "write_file" not in ReActEngine._DEFAULT_CORE_TOOLS
|
|
|
|
|
|
def test_tool_exported_from_tools_package() -> None:
|
|
from agentkit.tools import StrReplaceEditorTool as Exported
|
|
|
|
assert Exported is StrReplaceEditorTool
|
|
|
|
|
|
def test_tool_name_and_schema(tool: StrReplaceEditorTool) -> None:
|
|
assert tool.name == "str_replace_editor"
|
|
assert tool.input_schema is not None
|
|
props = tool.input_schema["properties"]
|
|
assert "command" in props
|
|
assert set(props["command"]["enum"]) == {
|
|
"create",
|
|
"str_replace",
|
|
"insert_at_line",
|
|
"view",
|
|
}
|
|
# Description mentions all four commands so the LLM knows what it can do.
|
|
assert "create" in tool.description
|
|
assert "str_replace" in tool.description
|
|
assert "insert_at_line" in tool.description
|
|
assert "view" in tool.description
|
|
|
|
|
|
def test_tool_appears_in_prompt_when_registered() -> None:
|
|
"""When a StrReplaceEditorTool is in the tool list and is a default core
|
|
tool, its full description (name + parameters) must appear in the
|
|
ReActEngine tool-use prompt (tiered injection contract)."""
|
|
from unittest.mock import MagicMock
|
|
|
|
from agentkit.core.react import ReActEngine
|
|
|
|
engine = ReActEngine(llm_gateway=MagicMock(), max_steps=1)
|
|
prompt = engine._build_tool_use_prompt([StrReplaceEditorTool()])
|
|
# Full description injected (core tool).
|
|
assert "str_replace_editor" in prompt
|
|
assert "create" in prompt and "str_replace" in prompt
|
|
assert "insert_at_line" in prompt and "view" in prompt
|
|
|
|
|
|
# ── end-to-end workflow ───────────────────────────────────────────────
|
|
|
|
|
|
async def test_create_view_str_replace_insert_workflow(
|
|
tool: StrReplaceEditorTool, workspace: Path
|
|
) -> None:
|
|
# 1. create
|
|
created = await tool.execute(
|
|
command="create",
|
|
path="app.py",
|
|
file_text="def main():\n pass\n",
|
|
)
|
|
assert created["is_error"] is False
|
|
|
|
# 2. view (get exact anchors / line numbers)
|
|
viewed = await tool.execute(command="view", path="app.py")
|
|
assert viewed["is_error"] is False
|
|
assert " 1\tdef main():" in viewed["content"]
|
|
|
|
# 3. str_replace
|
|
replaced = await tool.execute(
|
|
command="str_replace",
|
|
path="app.py",
|
|
old_str=" pass",
|
|
new_str=" return 42",
|
|
)
|
|
assert replaced["is_error"] is False
|
|
|
|
# 4. insert_at_line (add a docstring at the top)
|
|
inserted = await tool.execute(
|
|
command="insert_at_line",
|
|
path="app.py",
|
|
insert_line=0,
|
|
new_str='"""Module doc."""',
|
|
)
|
|
assert inserted["is_error"] is False
|
|
|
|
final = (workspace / "app.py").read_text()
|
|
assert final == '"""Module doc."""\ndef main():\n return 42\n'
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# Allow direct execution for a quick smoke check without pytest.
|
|
sys.exit(pytest.main([__file__, "-x", "-q"]))
|