jpmschweitzerandClaude
eb3467d06a
97 findings to zero. Most were mechanical — 52 unsorted import blocks, 10 unsorted __all__, assorted pyupgrade and simplify hints. Two were not, and both were visible only because the lint made me look. `webber version` did not exist. src/cli/commands/version.py defines show_version(), main.py imported it, and the registration line was never written — the CLI exposed chat and explore only. The import carried `# noqa: F401`, which is what kept the omission quiet: someone marked the symptom as intentional instead of asking why it was unused. show_version is not redundant with the --version flag; it prints the resolved Ollama URL, model and debug state, which is the form worth having when something is misconfigured. Registered, and the suppression dropped because the import is now genuinely used. test_spawn_explore_agent asserted nothing. It built a mock RunContext, patched get_agent, and stopped at the comment "For now, verify the explore agent would be called correctly". It had been counted as a passing test. An AST sweep of all 238 test functions found it was the only one, which is worth knowing — the problem was contained, not systemic. It is now skipped with a reason, so it reports as unfinished rather than as passing. Reducing it rather than deleting its imports was the point: tidying the imports would have made a hollow test look clean. Two findings were false positives, and both are recorded rather than silently worked around: B023 flagged run_agent closing over full_prompt and ctx. Traced: agent_task is awaited at line 326 before `continue` reaches the next iteration, so neither name can be rebound while the closure is pending, and the exception path cancels and awaits too. Not a bug. Bound as defaults anyway, because that stays true if the await ever moves. I had called it a live bug before tracing it, which is the mistake Rule 5 exists for. RUF012 flagged `rules: list[ApprovalRule] = []` on ApprovalRuleSet. Its suggested fix — annotate ClassVar — would remove the field from the model. ApprovalRuleSet is a pydantic model and pydantic deep-copies defaults per instance; verified by constructing two and confirming their lists are distinct objects. Suppressed with that evidence in the comment. Ruff cannot see the pydantic base because BaseSchema is a local subclass of BaseModel. Also moved a stray `from src.shared.logging import ...` that had drifted below a function definition, and merged a nested if in the ollama provider. 215 passed, 23 skipped, unchanged except for the new skip. `webber version` exercised end to end. mypy is NOT addressed here and the gate still fails on it — 55 errors in 14 files, 35 of them no-any-return from pydantic_ai's untyped returns. That was hidden behind ruff, because the gate stops at the first failing stage. Co-Authored-By: Claude <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.glob import GlobFilesTool
|
|
from src.domains.tools.file.read import ReadFileTool
|
|
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()
|