jpmschweitzerandClaude Opus 4.5
f4e8552298
Implements the first Claude-like agent for codebase exploration: Core Features: - Explore agent with glob, grep, read, and bash tools - Native PydanticAI tool calling with Ollama/Mistral Nemo - Sanitized Ollama provider (fixes content:null issue) - REST API endpoints for agent execution Tool Infrastructure: - BaseTool abstract class with ToolResult dataclass - ReadFileTool, GlobFilesTool, GrepContentTool, BashReadOnlyTool - Path validation and sandboxing support CLI Client (separate package for future extraction): - webber-cli command with chat, explore, status commands - Communicates with Webber API backend - Rich console output with theming Configuration: - Dev server on port 8095 (production uses 8086) - Mistral Nemo optimizations (temp 0.3, tool_choice required) Tests: 24 tests covering tools and API endpoints Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
235 lines
7.2 KiB
Python
235 lines
7.2 KiB
Python
"""
|
|
Tests for tool implementations.
|
|
"""
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from src.domains.tools.file.read import ReadFileTool
|
|
from src.domains.tools.file.glob import GlobFilesTool
|
|
from src.domains.tools.search.grep import GrepContentTool
|
|
from src.domains.tools.shell.bash import BashReadOnlyTool
|
|
|
|
|
|
class TestReadFileTool:
|
|
"""Tests for ReadFileTool."""
|
|
|
|
@pytest.fixture
|
|
def tool(self):
|
|
return ReadFileTool()
|
|
|
|
@pytest.fixture
|
|
def temp_file(self):
|
|
"""Create a temporary file with content."""
|
|
with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) as f:
|
|
for i in range(100):
|
|
f.write(f"Line {i + 1}: This is test content\n")
|
|
f.flush()
|
|
yield Path(f.name)
|
|
Path(f.name).unlink(missing_ok=True)
|
|
|
|
@pytest.mark.anyio
|
|
async def test_read_file_success(self, tool, temp_file):
|
|
"""Test reading a file successfully."""
|
|
result = await tool.execute(file_path=str(temp_file))
|
|
|
|
assert result.success
|
|
assert "Line 1:" in result.data
|
|
assert result.metadata.get("total_lines") == 100
|
|
|
|
@pytest.mark.anyio
|
|
async def test_read_file_with_offset(self, tool, temp_file):
|
|
"""Test reading with offset."""
|
|
result = await tool.execute(file_path=str(temp_file), offset=10, limit=5)
|
|
|
|
assert result.success
|
|
assert "Line 11:" in result.data
|
|
assert result.metadata.get("lines_returned") == 5
|
|
|
|
@pytest.mark.anyio
|
|
async def test_read_file_not_found(self, tool):
|
|
"""Test reading non-existent file."""
|
|
result = await tool.execute(file_path="/nonexistent/file.txt")
|
|
|
|
assert not result.success
|
|
assert "not found" in result.error.lower()
|
|
|
|
@pytest.mark.anyio
|
|
async def test_read_file_path_restriction(self, temp_file):
|
|
"""Test path restriction enforcement."""
|
|
tool = ReadFileTool(allowed_paths=["/some/other/path"])
|
|
result = await tool.execute(file_path=str(temp_file))
|
|
|
|
assert not result.success
|
|
assert "not in allowed" in result.error.lower()
|
|
|
|
|
|
class TestGlobFilesTool:
|
|
"""Tests for GlobFilesTool."""
|
|
|
|
@pytest.fixture
|
|
def tool(self):
|
|
return GlobFilesTool()
|
|
|
|
@pytest.fixture
|
|
def temp_dir(self):
|
|
"""Create a temporary directory with files."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = Path(tmpdir)
|
|
# Create test files
|
|
(path / "file1.py").write_text("# Python file 1")
|
|
(path / "file2.py").write_text("# Python file 2")
|
|
(path / "readme.md").write_text("# Readme")
|
|
(path / "subdir").mkdir()
|
|
(path / "subdir" / "nested.py").write_text("# Nested")
|
|
yield path
|
|
|
|
@pytest.mark.anyio
|
|
async def test_glob_python_files(self, tool, temp_dir):
|
|
"""Test finding Python files."""
|
|
result = await tool.execute(pattern="**/*.py", path=str(temp_dir))
|
|
|
|
assert result.success
|
|
assert "file1.py" in result.data
|
|
assert "file2.py" in result.data
|
|
assert "nested.py" in result.data
|
|
|
|
@pytest.mark.anyio
|
|
async def test_glob_with_limit(self, tool, temp_dir):
|
|
"""Test result limiting."""
|
|
result = await tool.execute(pattern="**/*.py", path=str(temp_dir), limit=2)
|
|
|
|
assert result.success
|
|
assert result.metadata.get("returned") == 2
|
|
|
|
@pytest.mark.anyio
|
|
async def test_glob_no_matches(self, tool, temp_dir):
|
|
"""Test when no files match."""
|
|
result = await tool.execute(pattern="**/*.xyz", path=str(temp_dir))
|
|
|
|
assert result.success
|
|
assert "No files found" in result.data
|
|
|
|
|
|
class TestGrepContentTool:
|
|
"""Tests for GrepContentTool."""
|
|
|
|
@pytest.fixture
|
|
def tool(self):
|
|
return GrepContentTool()
|
|
|
|
@pytest.fixture
|
|
def temp_dir(self):
|
|
"""Create a temporary directory with searchable content."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = Path(tmpdir)
|
|
(path / "code.py").write_text("""
|
|
def hello_world():
|
|
print("Hello, World!")
|
|
|
|
def goodbye_world():
|
|
print("Goodbye!")
|
|
""")
|
|
(path / "config.py").write_text("""
|
|
DEBUG = True
|
|
API_KEY = "secret"
|
|
""")
|
|
yield path
|
|
|
|
@pytest.mark.anyio
|
|
async def test_grep_pattern(self, tool, temp_dir):
|
|
"""Test searching for a pattern."""
|
|
result = await tool.execute(pattern="def.*world", path=str(temp_dir))
|
|
|
|
assert result.success
|
|
assert "hello_world" in result.data
|
|
assert "goodbye_world" in result.data
|
|
|
|
@pytest.mark.anyio
|
|
async def test_grep_case_insensitive(self, tool, temp_dir):
|
|
"""Test case-insensitive search."""
|
|
result = await tool.execute(
|
|
pattern="DEBUG",
|
|
path=str(temp_dir),
|
|
case_sensitive=False
|
|
)
|
|
|
|
assert result.success
|
|
assert "DEBUG" in result.data
|
|
|
|
@pytest.mark.anyio
|
|
async def test_grep_with_file_glob(self, tool, temp_dir):
|
|
"""Test filtering by file glob."""
|
|
result = await tool.execute(
|
|
pattern="=",
|
|
path=str(temp_dir),
|
|
file_glob="config.py"
|
|
)
|
|
|
|
assert result.success
|
|
assert "config.py" in result.data
|
|
|
|
|
|
class TestBashReadOnlyTool:
|
|
"""Tests for BashReadOnlyTool."""
|
|
|
|
@pytest.fixture
|
|
def tool(self):
|
|
return BashReadOnlyTool()
|
|
|
|
@pytest.mark.anyio
|
|
async def test_ls_command(self, tool):
|
|
"""Test allowed ls command."""
|
|
result = await tool.execute(command="ls -la", cwd="/tmp")
|
|
|
|
assert result.success
|
|
|
|
@pytest.mark.anyio
|
|
async def test_pwd_command(self, tool):
|
|
"""Test allowed pwd command."""
|
|
result = await tool.execute(command="pwd", cwd="/tmp")
|
|
|
|
assert result.success
|
|
assert "/tmp" in result.data
|
|
|
|
@pytest.mark.anyio
|
|
async def test_forbidden_rm_command(self, tool):
|
|
"""Test that rm is blocked."""
|
|
result = await tool.execute(command="rm -rf /tmp/test")
|
|
|
|
assert not result.success
|
|
assert "forbidden" in result.error.lower() or "not allowed" in result.error.lower()
|
|
|
|
@pytest.mark.anyio
|
|
async def test_forbidden_redirect(self, tool):
|
|
"""Test that redirects are blocked."""
|
|
result = await tool.execute(command="echo test > /tmp/file")
|
|
|
|
assert not result.success
|
|
assert "forbidden" in result.error.lower()
|
|
|
|
@pytest.mark.anyio
|
|
async def test_forbidden_chaining(self, tool):
|
|
"""Test that command chaining is blocked."""
|
|
result = await tool.execute(command="ls && rm -rf /")
|
|
|
|
assert not result.success
|
|
|
|
@pytest.mark.anyio
|
|
async def test_git_status(self, tool, tmp_path):
|
|
"""Test git status on non-git directory."""
|
|
result = await tool.execute(command="git status", cwd=str(tmp_path))
|
|
|
|
# Should fail but not because command is forbidden
|
|
assert not result.success
|
|
assert "not a git repository" in result.error.lower() or "fatal" in result.data.lower()
|
|
|
|
@pytest.mark.anyio
|
|
async def test_forbidden_curl(self, tool):
|
|
"""Test that curl is blocked."""
|
|
result = await tool.execute(command="curl http://example.com")
|
|
|
|
assert not result.success
|
|
assert "forbidden" in result.error.lower() or "not allowed" in result.error.lower()
|