jpmschweitzerandClaude Opus 4.5
d0fa5b38a7
New tools for code modification: - EditFileTool: find-and-replace with safety checks (unique match required) - WriteFileTool: create/overwrite files with path validation - BashTool: full bash with controlled write access Security controls on BashTool: - Allowed: mkdir, touch, cp, mv, rm (single files), git, pip, pytest - Forbidden: sudo, curl, wget, ssh, rm -rf, chmod 777 Includes 39 new tests (78 total now passing). Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
482 lines
16 KiB
Python
482 lines
16 KiB
Python
"""
|
|
Tests for coding tools (edit, write, bash full).
|
|
"""
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from src.domains.tools.file.edit import EditFileTool
|
|
from src.domains.tools.file.write import WriteFileTool
|
|
from src.domains.tools.shell.bash_full import BashTool
|
|
|
|
|
|
class TestEditFileTool:
|
|
"""Tests for EditFileTool."""
|
|
|
|
@pytest.fixture
|
|
def tool(self):
|
|
return EditFileTool()
|
|
|
|
@pytest.fixture
|
|
def temp_file(self):
|
|
"""Create a temporary file with content."""
|
|
with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
|
|
f.write("def hello():\n return 'Hello'\n\ndef world():\n return 'World'\n")
|
|
f.flush()
|
|
yield Path(f.name)
|
|
Path(f.name).unlink(missing_ok=True)
|
|
|
|
@pytest.mark.anyio
|
|
async def test_edit_file_success(self, tool, temp_file):
|
|
"""Test successful single replacement."""
|
|
result = await tool.execute(
|
|
file_path=str(temp_file),
|
|
old_string="return 'Hello'",
|
|
new_string="return 'Hi'"
|
|
)
|
|
assert result.success
|
|
assert "Hi" in Path(temp_file).read_text()
|
|
assert "Hello" not in Path(temp_file).read_text()
|
|
|
|
@pytest.mark.anyio
|
|
async def test_edit_file_not_found(self, tool):
|
|
"""Test editing non-existent file."""
|
|
result = await tool.execute(
|
|
file_path="/nonexistent/file.py",
|
|
old_string="old",
|
|
new_string="new"
|
|
)
|
|
assert not result.success
|
|
assert "not found" in result.error.lower()
|
|
|
|
@pytest.mark.anyio
|
|
async def test_edit_file_old_string_not_found(self, tool, temp_file):
|
|
"""Test when old_string doesn't exist in file."""
|
|
result = await tool.execute(
|
|
file_path=str(temp_file),
|
|
old_string="nonexistent text",
|
|
new_string="replacement"
|
|
)
|
|
assert not result.success
|
|
assert "not found" in result.error.lower()
|
|
|
|
@pytest.mark.anyio
|
|
async def test_edit_file_multiple_matches_error(self, tool, temp_file):
|
|
"""Test error when old_string has multiple matches and replace_all=False."""
|
|
result = await tool.execute(
|
|
file_path=str(temp_file),
|
|
old_string="return",
|
|
new_string="yield"
|
|
)
|
|
assert not result.success
|
|
assert "2" in result.error # Should mention count
|
|
|
|
@pytest.mark.anyio
|
|
async def test_edit_file_replace_all(self, tool, temp_file):
|
|
"""Test replace_all=True replaces all occurrences."""
|
|
result = await tool.execute(
|
|
file_path=str(temp_file),
|
|
old_string="return",
|
|
new_string="yield",
|
|
replace_all=True
|
|
)
|
|
assert result.success
|
|
content = Path(temp_file).read_text()
|
|
assert "return" not in content
|
|
assert content.count("yield") == 2
|
|
|
|
@pytest.mark.anyio
|
|
async def test_edit_file_path_restriction(self, temp_file):
|
|
"""Test path restriction enforcement."""
|
|
tool = EditFileTool(allowed_paths=["/some/other/path"])
|
|
result = await tool.execute(
|
|
file_path=str(temp_file),
|
|
old_string="Hello",
|
|
new_string="Hi"
|
|
)
|
|
assert not result.success
|
|
assert "not in allowed" in result.error.lower()
|
|
|
|
@pytest.mark.anyio
|
|
async def test_edit_file_empty_old_string(self, tool, temp_file):
|
|
"""Test that empty old_string is rejected."""
|
|
result = await tool.execute(
|
|
file_path=str(temp_file),
|
|
old_string="",
|
|
new_string="new"
|
|
)
|
|
assert not result.success
|
|
assert "empty" in result.error.lower()
|
|
|
|
@pytest.mark.anyio
|
|
async def test_edit_file_same_string(self, tool, temp_file):
|
|
"""Test that identical old/new strings are rejected."""
|
|
result = await tool.execute(
|
|
file_path=str(temp_file),
|
|
old_string="Hello",
|
|
new_string="Hello"
|
|
)
|
|
assert not result.success
|
|
assert "identical" in result.error.lower()
|
|
|
|
|
|
class TestWriteFileTool:
|
|
"""Tests for WriteFileTool."""
|
|
|
|
@pytest.fixture
|
|
def tool(self):
|
|
return WriteFileTool()
|
|
|
|
@pytest.fixture
|
|
def temp_dir(self):
|
|
"""Create a temporary directory."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
yield Path(tmpdir)
|
|
|
|
@pytest.mark.anyio
|
|
async def test_write_new_file(self, tool, temp_dir):
|
|
"""Test creating a new file."""
|
|
file_path = temp_dir / "new_file.py"
|
|
result = await tool.execute(
|
|
file_path=str(file_path),
|
|
content="# New file\nprint('hello')"
|
|
)
|
|
assert result.success
|
|
assert file_path.exists()
|
|
assert "hello" in file_path.read_text()
|
|
assert result.metadata.get("overwritten") is False
|
|
|
|
@pytest.mark.anyio
|
|
async def test_write_overwrite_existing(self, tool, temp_dir):
|
|
"""Test overwriting an existing file."""
|
|
file_path = temp_dir / "existing.txt"
|
|
file_path.write_text("old content")
|
|
|
|
result = await tool.execute(
|
|
file_path=str(file_path),
|
|
content="new content"
|
|
)
|
|
assert result.success
|
|
assert file_path.read_text() == "new content"
|
|
assert result.metadata.get("overwritten") is True
|
|
|
|
@pytest.mark.anyio
|
|
async def test_write_file_path_restriction(self, temp_dir):
|
|
"""Test path restriction enforcement."""
|
|
tool = WriteFileTool(allowed_paths=["/some/other/path"])
|
|
result = await tool.execute(
|
|
file_path=str(temp_dir / "file.txt"),
|
|
content="content"
|
|
)
|
|
assert not result.success
|
|
assert "not in allowed" in result.error.lower()
|
|
|
|
@pytest.mark.anyio
|
|
async def test_write_file_parent_not_exists(self, tool, temp_dir):
|
|
"""Test writing to path where parent directory doesn't exist."""
|
|
file_path = temp_dir / "nonexistent_dir" / "file.txt"
|
|
result = await tool.execute(
|
|
file_path=str(file_path),
|
|
content="content"
|
|
)
|
|
assert not result.success
|
|
assert "parent" in result.error.lower() or "directory" in result.error.lower()
|
|
|
|
@pytest.mark.anyio
|
|
async def test_write_file_content_size_limit(self, temp_dir):
|
|
"""Test content size limit enforcement."""
|
|
tool = WriteFileTool(max_content_size=100)
|
|
result = await tool.execute(
|
|
file_path=str(temp_dir / "large.txt"),
|
|
content="x" * 200
|
|
)
|
|
assert not result.success
|
|
assert "large" in result.error.lower() or "size" in result.error.lower()
|
|
|
|
@pytest.mark.anyio
|
|
async def test_write_file_returns_metadata(self, tool, temp_dir):
|
|
"""Test that metadata is returned correctly."""
|
|
file_path = temp_dir / "meta.txt"
|
|
content = "line1\nline2\nline3"
|
|
result = await tool.execute(
|
|
file_path=str(file_path),
|
|
content=content
|
|
)
|
|
assert result.success
|
|
assert result.metadata.get("lines") == 3
|
|
assert result.metadata.get("file_size") == len(content.encode('utf-8'))
|
|
|
|
|
|
class TestBashTool:
|
|
"""Tests for BashTool (full write capabilities)."""
|
|
|
|
@pytest.fixture
|
|
def tool(self):
|
|
return BashTool()
|
|
|
|
@pytest.fixture
|
|
def temp_dir(self):
|
|
"""Create a temporary directory."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
yield Path(tmpdir)
|
|
|
|
# === Allowed commands ===
|
|
|
|
@pytest.mark.anyio
|
|
async def test_ls_command(self, tool, temp_dir):
|
|
"""Test ls is allowed."""
|
|
result = await tool.execute(command="ls", cwd=str(temp_dir))
|
|
assert result.success
|
|
|
|
@pytest.mark.anyio
|
|
async def test_mkdir_command(self, tool, temp_dir):
|
|
"""Test mkdir is allowed."""
|
|
result = await tool.execute(
|
|
command=f"mkdir {temp_dir}/new_dir",
|
|
cwd=str(temp_dir)
|
|
)
|
|
assert result.success
|
|
assert (temp_dir / "new_dir").exists()
|
|
|
|
@pytest.mark.anyio
|
|
async def test_touch_command(self, tool, temp_dir):
|
|
"""Test touch is allowed."""
|
|
result = await tool.execute(
|
|
command=f"touch {temp_dir}/new_file.txt",
|
|
cwd=str(temp_dir)
|
|
)
|
|
assert result.success
|
|
assert (temp_dir / "new_file.txt").exists()
|
|
|
|
@pytest.mark.anyio
|
|
async def test_cp_command(self, tool, temp_dir):
|
|
"""Test cp within allowed paths."""
|
|
(temp_dir / "source.txt").write_text("content")
|
|
result = await tool.execute(
|
|
command=f"cp {temp_dir}/source.txt {temp_dir}/dest.txt",
|
|
cwd=str(temp_dir)
|
|
)
|
|
assert result.success
|
|
assert (temp_dir / "dest.txt").exists()
|
|
|
|
@pytest.mark.anyio
|
|
async def test_mv_command(self, tool, temp_dir):
|
|
"""Test mv within allowed paths."""
|
|
(temp_dir / "source.txt").write_text("content")
|
|
result = await tool.execute(
|
|
command=f"mv {temp_dir}/source.txt {temp_dir}/moved.txt",
|
|
cwd=str(temp_dir)
|
|
)
|
|
assert result.success
|
|
assert (temp_dir / "moved.txt").exists()
|
|
assert not (temp_dir / "source.txt").exists()
|
|
|
|
@pytest.mark.anyio
|
|
async def test_command_chaining_and(self, tool, temp_dir):
|
|
"""Test && chaining is allowed."""
|
|
result = await tool.execute(
|
|
command=f"mkdir {temp_dir}/dir1 && touch {temp_dir}/dir1/file.txt",
|
|
cwd=str(temp_dir)
|
|
)
|
|
assert result.success
|
|
assert (temp_dir / "dir1" / "file.txt").exists()
|
|
|
|
@pytest.mark.anyio
|
|
async def test_command_chaining_or(self, tool, temp_dir):
|
|
"""Test || chaining is allowed."""
|
|
result = await tool.execute(
|
|
command=f"ls {temp_dir}/nonexistent || echo 'fallback'",
|
|
cwd=str(temp_dir)
|
|
)
|
|
# Either succeeds or falls back
|
|
assert result.success or "fallback" in (result.data or "")
|
|
|
|
@pytest.mark.anyio
|
|
async def test_echo_command(self, tool, temp_dir):
|
|
"""Test echo command."""
|
|
result = await tool.execute(
|
|
command="echo 'hello world'",
|
|
cwd=str(temp_dir)
|
|
)
|
|
assert result.success
|
|
assert "hello world" in result.data
|
|
|
|
@pytest.mark.anyio
|
|
async def test_git_status(self, tool, temp_dir):
|
|
"""Test git status is allowed."""
|
|
# Initialize a git repo first
|
|
await tool.execute(command="git init", cwd=str(temp_dir))
|
|
result = await tool.execute(command="git status", cwd=str(temp_dir))
|
|
assert result.success
|
|
|
|
@pytest.mark.anyio
|
|
async def test_git_add_allowed(self, tool, temp_dir):
|
|
"""Test git add is allowed."""
|
|
await tool.execute(command="git init", cwd=str(temp_dir))
|
|
(temp_dir / "file.txt").write_text("content")
|
|
result = await tool.execute(command="git add file.txt", cwd=str(temp_dir))
|
|
assert result.success
|
|
|
|
@pytest.mark.anyio
|
|
async def test_pip_help(self, tool, temp_dir):
|
|
"""Test pip help is allowed."""
|
|
result = await tool.execute(
|
|
command="pip --help",
|
|
cwd=str(temp_dir)
|
|
)
|
|
assert result.success
|
|
|
|
@pytest.mark.anyio
|
|
async def test_rm_single_file(self, tool, temp_dir):
|
|
"""Test rm of a single file is allowed."""
|
|
file_path = temp_dir / "to_delete.txt"
|
|
file_path.write_text("content")
|
|
|
|
result = await tool.execute(
|
|
command=f"rm {file_path}",
|
|
cwd=str(temp_dir)
|
|
)
|
|
assert result.success
|
|
assert not file_path.exists()
|
|
|
|
# === Forbidden commands ===
|
|
|
|
@pytest.mark.anyio
|
|
async def test_forbidden_sudo(self, tool, temp_dir):
|
|
"""Test sudo is blocked."""
|
|
result = await tool.execute(command="sudo ls", cwd=str(temp_dir))
|
|
assert not result.success
|
|
assert "forbidden" in result.error.lower()
|
|
|
|
@pytest.mark.anyio
|
|
async def test_forbidden_curl(self, tool, temp_dir):
|
|
"""Test curl is blocked."""
|
|
result = await tool.execute(
|
|
command="curl http://example.com",
|
|
cwd=str(temp_dir)
|
|
)
|
|
assert not result.success
|
|
assert "not allowed" in result.error.lower() or "forbidden" in result.error.lower()
|
|
|
|
@pytest.mark.anyio
|
|
async def test_forbidden_wget(self, tool, temp_dir):
|
|
"""Test wget is blocked."""
|
|
result = await tool.execute(
|
|
command="wget http://example.com",
|
|
cwd=str(temp_dir)
|
|
)
|
|
assert not result.success
|
|
|
|
@pytest.mark.anyio
|
|
async def test_forbidden_ssh(self, tool, temp_dir):
|
|
"""Test ssh is blocked."""
|
|
result = await tool.execute(
|
|
command="ssh user@host",
|
|
cwd=str(temp_dir)
|
|
)
|
|
assert not result.success
|
|
|
|
@pytest.mark.anyio
|
|
async def test_forbidden_rm_rf(self, tool, temp_dir):
|
|
"""Test rm -rf is blocked."""
|
|
result = await tool.execute(
|
|
command="rm -rf /tmp/test",
|
|
cwd=str(temp_dir)
|
|
)
|
|
assert not result.success
|
|
assert "not allowed" in result.error.lower() or "forbidden" in result.error.lower()
|
|
|
|
@pytest.mark.anyio
|
|
async def test_forbidden_rm_rf_dot(self, tool, temp_dir):
|
|
"""Test rm -rf . is blocked."""
|
|
result = await tool.execute(
|
|
command="rm -rf .",
|
|
cwd=str(temp_dir)
|
|
)
|
|
assert not result.success
|
|
|
|
@pytest.mark.anyio
|
|
async def test_forbidden_chmod_777(self, tool, temp_dir):
|
|
"""Test chmod 777 is blocked."""
|
|
result = await tool.execute(
|
|
command="chmod 777 /tmp/file",
|
|
cwd=str(temp_dir)
|
|
)
|
|
assert not result.success
|
|
assert "not allowed" in result.error.lower() or "forbidden" in result.error.lower()
|
|
|
|
@pytest.mark.anyio
|
|
async def test_forbidden_git_push(self, tool, temp_dir):
|
|
"""Test git push is blocked."""
|
|
await tool.execute(command="git init", cwd=str(temp_dir))
|
|
result = await tool.execute(
|
|
command="git push origin main",
|
|
cwd=str(temp_dir)
|
|
)
|
|
assert not result.success
|
|
assert "not allowed" in result.error.lower()
|
|
|
|
@pytest.mark.anyio
|
|
async def test_forbidden_pip_uninstall(self, tool, temp_dir):
|
|
"""Test pip uninstall is blocked."""
|
|
result = await tool.execute(
|
|
command="pip uninstall requests",
|
|
cwd=str(temp_dir)
|
|
)
|
|
assert not result.success
|
|
assert "not allowed" in result.error.lower()
|
|
|
|
# === Path restrictions ===
|
|
|
|
@pytest.mark.anyio
|
|
async def test_working_dir_not_allowed(self, temp_dir):
|
|
"""Test working directory restriction."""
|
|
tool = BashTool(allowed_paths=["/some/other/path"])
|
|
result = await tool.execute(
|
|
command="ls",
|
|
cwd=str(temp_dir)
|
|
)
|
|
assert not result.success
|
|
assert "not allowed" in result.error.lower()
|
|
|
|
@pytest.mark.anyio
|
|
async def test_cp_outside_allowed_paths(self, temp_dir):
|
|
"""Test cp to path outside allowed_paths fails."""
|
|
tool = BashTool(allowed_paths=[str(temp_dir)])
|
|
(temp_dir / "source.txt").write_text("content")
|
|
|
|
result = await tool.execute(
|
|
command=f"cp {temp_dir}/source.txt /tmp/dest.txt",
|
|
cwd=str(temp_dir)
|
|
)
|
|
assert not result.success
|
|
assert "allowed" in result.error.lower()
|
|
|
|
@pytest.mark.anyio
|
|
async def test_rm_outside_allowed_paths(self, temp_dir):
|
|
"""Test rm of path outside allowed_paths fails."""
|
|
tool = BashTool(allowed_paths=[str(temp_dir)])
|
|
|
|
result = await tool.execute(
|
|
command="rm /tmp/some_file.txt",
|
|
cwd=str(temp_dir)
|
|
)
|
|
assert not result.success
|
|
assert "allowed" in result.error.lower()
|
|
|
|
# === Timeout ===
|
|
|
|
@pytest.mark.anyio
|
|
async def test_timeout(self, tool, temp_dir):
|
|
"""Test command timeout using find on root (slow)."""
|
|
# Use find on a large directory which will be slow
|
|
result = await tool.execute(
|
|
command="find / -name '*.nonexistent' 2>/dev/null",
|
|
cwd=str(temp_dir),
|
|
timeout=1
|
|
)
|
|
# The command should either timeout or fail
|
|
# (it may complete quickly with errors, which is also acceptable)
|
|
assert not result.success or result.truncated or "timeout" in str(result.error or "").lower()
|