refactor: reorganize into monorepo with separate subprojects
Structure webber into three independent subprojects: - webber-api/: FastAPI backend server with all agent code - webber-cli/: Standalone CLI client (renamed from cli/ to webber_cli/) - webber-sandbox/: Test project for functional testing Key changes: - Each subproject has its own .venv (Python 3.12+) - Added sandbox.sh for managing test project templates - Created sandbox-templates/ with calculator-cli and empty starter - Updated CI/CD for prefixed tags (api/v*, cli/v*) - Added comprehensive AGENTS.md with operational instructions - Added gitignore filtering to glob and grep tools - Created pyproject.toml for each subproject Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,34 @@
|
||||
"""
|
||||
Pytest configuration and fixtures.
|
||||
"""
|
||||
import pytest
|
||||
from httpx import ASGITransport, AsyncClient
|
||||
|
||||
from src.main import app
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def anyio_backend():
|
||||
"""Use asyncio for async tests."""
|
||||
return "asyncio"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def client():
|
||||
"""Async HTTP client for testing."""
|
||||
async with AsyncClient(
|
||||
transport=ASGITransport(app=app),
|
||||
base_url="http://test"
|
||||
) as ac:
|
||||
yield ac
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def auth_client():
|
||||
"""Async HTTP client with API key for authenticated requests."""
|
||||
async with AsyncClient(
|
||||
transport=ASGITransport(app=app),
|
||||
base_url="http://test",
|
||||
headers={"X-API-Key": "test-api-key"}
|
||||
) as ac:
|
||||
yield ac
|
||||
@@ -0,0 +1,75 @@
|
||||
"""
|
||||
Tests for agent REST API endpoints.
|
||||
"""
|
||||
import pytest
|
||||
|
||||
|
||||
class TestAgentListEndpoint:
|
||||
"""Tests for GET /agents/ endpoint."""
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_list_agents(self, auth_client):
|
||||
"""Test listing available agents."""
|
||||
response = await auth_client.get("/agents/")
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert "agents" in data
|
||||
assert len(data["agents"]) >= 1
|
||||
|
||||
# Check explore agent is present
|
||||
agent_names = [a["name"] for a in data["agents"]]
|
||||
assert "explore" in agent_names
|
||||
|
||||
|
||||
class TestAgentInfoEndpoint:
|
||||
"""Tests for GET /agents/{agent_type} endpoint."""
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_get_explore_agent_info(self, auth_client):
|
||||
"""Test getting explore agent info."""
|
||||
response = await auth_client.get("/agents/explore")
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["name"] == "explore"
|
||||
assert "description" in data
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_get_unknown_agent(self, auth_client):
|
||||
"""Test getting info for unknown agent."""
|
||||
response = await auth_client.get("/agents/nonexistent")
|
||||
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
class TestAgentRunEndpoint:
|
||||
"""Tests for POST /agents/run endpoint."""
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_run_with_unknown_agent(self, auth_client):
|
||||
"""Test running unknown agent type."""
|
||||
response = await auth_client.post(
|
||||
"/agents/run",
|
||||
json={
|
||||
"prompt": "test",
|
||||
"agent_type": "nonexistent",
|
||||
"working_dir": "."
|
||||
}
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert "Unknown agent" in response.json()["detail"]
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_run_request_validation(self, auth_client):
|
||||
"""Test request validation."""
|
||||
# Missing required field
|
||||
response = await auth_client.post(
|
||||
"/agents/run",
|
||||
json={
|
||||
"working_dir": "."
|
||||
}
|
||||
)
|
||||
|
||||
assert response.status_code == 422 # Validation error
|
||||
@@ -0,0 +1,330 @@
|
||||
"""
|
||||
Tests for gitignore filtering functionality.
|
||||
"""
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from src.domains.tools.gitignore import GitignoreFilter, filter_gitignored
|
||||
from src.domains.tools.file.glob import GlobFilesTool
|
||||
from src.domains.tools.search.grep import GrepContentTool
|
||||
|
||||
|
||||
class TestGitignoreFilter:
|
||||
"""Tests for GitignoreFilter class."""
|
||||
|
||||
@pytest.fixture
|
||||
def temp_dir_with_gitignore(self):
|
||||
"""Create a temporary directory with a .gitignore file."""
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
path = Path(tmpdir)
|
||||
|
||||
# Create .gitignore
|
||||
(path / ".gitignore").write_text("""
|
||||
# Python artifacts
|
||||
*.pyc
|
||||
__pycache__/
|
||||
|
||||
# Virtual environments
|
||||
.venv/
|
||||
venv/
|
||||
|
||||
# IDE
|
||||
.idea/
|
||||
|
||||
# Custom patterns
|
||||
ignored_file.txt
|
||||
ignored_dir/
|
||||
""")
|
||||
|
||||
# Create various files and directories
|
||||
(path / "main.py").write_text("# Main file")
|
||||
(path / "test.pyc").write_bytes(b"compiled")
|
||||
(path / "ignored_file.txt").write_text("should be ignored")
|
||||
(path / "not_ignored.txt").write_text("should be visible")
|
||||
|
||||
# Create directories
|
||||
(path / "__pycache__").mkdir()
|
||||
(path / "__pycache__" / "module.cpython-312.pyc").write_bytes(b"cache")
|
||||
|
||||
(path / ".venv").mkdir()
|
||||
(path / ".venv" / "lib").mkdir(parents=True)
|
||||
(path / ".venv" / "lib" / "python.py").write_text("venv file")
|
||||
|
||||
(path / "ignored_dir").mkdir()
|
||||
(path / "ignored_dir" / "hidden.py").write_text("hidden")
|
||||
|
||||
(path / "src").mkdir()
|
||||
(path / "src" / "app.py").write_text("# App")
|
||||
(path / "src" / "utils.py").write_text("# Utils")
|
||||
|
||||
yield path
|
||||
|
||||
def test_filter_ignores_pyc_files(self, temp_dir_with_gitignore):
|
||||
"""Test that .pyc files are ignored."""
|
||||
filter_instance = GitignoreFilter(temp_dir_with_gitignore)
|
||||
|
||||
assert filter_instance.is_ignored(temp_dir_with_gitignore / "test.pyc")
|
||||
assert not filter_instance.is_ignored(temp_dir_with_gitignore / "main.py")
|
||||
|
||||
def test_filter_ignores_pycache_dir(self, temp_dir_with_gitignore):
|
||||
"""Test that __pycache__ directory is ignored."""
|
||||
filter_instance = GitignoreFilter(temp_dir_with_gitignore)
|
||||
|
||||
pycache = temp_dir_with_gitignore / "__pycache__"
|
||||
assert filter_instance.is_ignored(pycache)
|
||||
assert filter_instance.is_ignored(pycache / "module.cpython-312.pyc")
|
||||
|
||||
def test_filter_ignores_venv(self, temp_dir_with_gitignore):
|
||||
"""Test that .venv directory is ignored."""
|
||||
filter_instance = GitignoreFilter(temp_dir_with_gitignore)
|
||||
|
||||
venv = temp_dir_with_gitignore / ".venv"
|
||||
assert filter_instance.is_ignored(venv)
|
||||
assert filter_instance.is_ignored(venv / "lib" / "python.py")
|
||||
|
||||
def test_filter_ignores_custom_patterns(self, temp_dir_with_gitignore):
|
||||
"""Test that custom gitignore patterns work."""
|
||||
filter_instance = GitignoreFilter(temp_dir_with_gitignore)
|
||||
|
||||
assert filter_instance.is_ignored(temp_dir_with_gitignore / "ignored_file.txt")
|
||||
assert filter_instance.is_ignored(temp_dir_with_gitignore / "ignored_dir")
|
||||
assert filter_instance.is_ignored(temp_dir_with_gitignore / "ignored_dir" / "hidden.py")
|
||||
|
||||
def test_filter_allows_regular_files(self, temp_dir_with_gitignore):
|
||||
"""Test that regular files are not ignored."""
|
||||
filter_instance = GitignoreFilter(temp_dir_with_gitignore)
|
||||
|
||||
assert not filter_instance.is_ignored(temp_dir_with_gitignore / "main.py")
|
||||
assert not filter_instance.is_ignored(temp_dir_with_gitignore / "not_ignored.txt")
|
||||
assert not filter_instance.is_ignored(temp_dir_with_gitignore / "src" / "app.py")
|
||||
|
||||
def test_filter_paths_function(self, temp_dir_with_gitignore):
|
||||
"""Test the filter_paths helper function."""
|
||||
filter_instance = GitignoreFilter(temp_dir_with_gitignore)
|
||||
|
||||
paths = [
|
||||
temp_dir_with_gitignore / "main.py",
|
||||
temp_dir_with_gitignore / "test.pyc",
|
||||
temp_dir_with_gitignore / "src" / "app.py",
|
||||
temp_dir_with_gitignore / ".venv" / "lib" / "python.py",
|
||||
]
|
||||
|
||||
filtered = filter_instance.filter_paths(paths)
|
||||
|
||||
assert len(filtered) == 2
|
||||
assert temp_dir_with_gitignore / "main.py" in filtered
|
||||
assert temp_dir_with_gitignore / "src" / "app.py" in filtered
|
||||
assert temp_dir_with_gitignore / "test.pyc" not in filtered
|
||||
assert temp_dir_with_gitignore / ".venv" / "lib" / "python.py" not in filtered
|
||||
|
||||
def test_filter_gitignored_convenience(self, temp_dir_with_gitignore):
|
||||
"""Test the filter_gitignored convenience function."""
|
||||
paths = list(temp_dir_with_gitignore.rglob("*.py"))
|
||||
|
||||
filtered = filter_gitignored(paths, temp_dir_with_gitignore)
|
||||
|
||||
# Should only include main.py, src/app.py, src/utils.py
|
||||
# Should exclude .venv/lib/python.py, ignored_dir/hidden.py
|
||||
filenames = {p.name for p in filtered}
|
||||
assert "main.py" in filenames
|
||||
assert "app.py" in filenames
|
||||
assert "utils.py" in filenames
|
||||
|
||||
# Check that ignored files are not present
|
||||
ignored_paths = [str(p) for p in filtered]
|
||||
assert not any(".venv" in p for p in ignored_paths)
|
||||
assert not any("ignored_dir" in p for p in ignored_paths)
|
||||
|
||||
|
||||
class TestGlobWithGitignore:
|
||||
"""Tests for GlobFilesTool gitignore integration."""
|
||||
|
||||
@pytest.fixture
|
||||
def temp_dir(self):
|
||||
"""Create a directory with ignored and non-ignored files."""
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
path = Path(tmpdir)
|
||||
|
||||
# Create .gitignore
|
||||
(path / ".gitignore").write_text("""
|
||||
*.log
|
||||
build/
|
||||
""")
|
||||
|
||||
# Create files
|
||||
(path / "main.py").write_text("# Main")
|
||||
(path / "debug.log").write_text("log content")
|
||||
(path / "src").mkdir()
|
||||
(path / "src" / "app.py").write_text("# App")
|
||||
|
||||
# Create .venv (default ignored)
|
||||
(path / ".venv").mkdir()
|
||||
(path / ".venv" / "script.py").write_text("venv")
|
||||
|
||||
# Create build directory (gitignore pattern)
|
||||
(path / "build").mkdir()
|
||||
(path / "build" / "output.py").write_text("build")
|
||||
|
||||
yield path
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_glob_honors_gitignore_by_default(self, temp_dir):
|
||||
"""Test that glob filters gitignored files by default."""
|
||||
tool = GlobFilesTool()
|
||||
|
||||
result = await tool.execute(pattern="**/*.py", path=str(temp_dir))
|
||||
|
||||
assert result.success
|
||||
assert "main.py" in result.data
|
||||
assert "app.py" in result.data
|
||||
assert ".venv" not in result.data
|
||||
assert "build" not in result.data
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_glob_filters_log_files(self, temp_dir):
|
||||
"""Test that custom gitignore patterns (*.log) work."""
|
||||
tool = GlobFilesTool()
|
||||
|
||||
result = await tool.execute(pattern="**/*", path=str(temp_dir))
|
||||
|
||||
assert result.success
|
||||
assert "debug.log" not in result.data
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_glob_can_disable_gitignore(self, temp_dir):
|
||||
"""Test that gitignore filtering can be disabled."""
|
||||
tool = GlobFilesTool(honor_gitignore=False)
|
||||
|
||||
result = await tool.execute(pattern="**/*.py", path=str(temp_dir))
|
||||
|
||||
assert result.success
|
||||
# When disabled, should include ignored files
|
||||
assert ".venv" in result.data or "build" in result.data
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_glob_override_per_call(self, temp_dir):
|
||||
"""Test per-call gitignore override."""
|
||||
tool = GlobFilesTool(honor_gitignore=True)
|
||||
|
||||
# Default behavior - filters
|
||||
result1 = await tool.execute(pattern="**/*.py", path=str(temp_dir))
|
||||
assert ".venv" not in result1.data
|
||||
|
||||
# Override to disable
|
||||
result2 = await tool.execute(
|
||||
pattern="**/*.py",
|
||||
path=str(temp_dir),
|
||||
honor_gitignore=False
|
||||
)
|
||||
assert ".venv" in result2.data or "build" in result2.data
|
||||
|
||||
|
||||
class TestGrepWithGitignore:
|
||||
"""Tests for GrepContentTool gitignore integration."""
|
||||
|
||||
@pytest.fixture
|
||||
def temp_dir(self):
|
||||
"""Create a directory with searchable content in ignored and non-ignored files."""
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
path = Path(tmpdir)
|
||||
|
||||
# Create .gitignore
|
||||
(path / ".gitignore").write_text("""
|
||||
ignored/
|
||||
""")
|
||||
|
||||
# Create files with searchable content
|
||||
(path / "main.py").write_text("def search_target(): pass")
|
||||
(path / "src").mkdir()
|
||||
(path / "src" / "utils.py").write_text("def search_target(): # utils")
|
||||
|
||||
# Create .venv with matching content (default ignored)
|
||||
(path / ".venv").mkdir()
|
||||
(path / ".venv" / "site.py").write_text("def search_target(): # venv")
|
||||
|
||||
# Create ignored dir with matching content
|
||||
(path / "ignored").mkdir()
|
||||
(path / "ignored" / "hidden.py").write_text("def search_target(): # hidden")
|
||||
|
||||
yield path
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_grep_honors_gitignore_by_default(self, temp_dir):
|
||||
"""Test that grep filters gitignored files by default."""
|
||||
tool = GrepContentTool()
|
||||
|
||||
result = await tool.execute(pattern="search_target", path=str(temp_dir))
|
||||
|
||||
assert result.success
|
||||
assert "main.py" in result.data
|
||||
assert "utils.py" in result.data
|
||||
assert ".venv" not in result.data
|
||||
assert "ignored" not in result.data
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_grep_can_disable_gitignore(self, temp_dir):
|
||||
"""Test that gitignore filtering can be disabled."""
|
||||
tool = GrepContentTool(honor_gitignore=False)
|
||||
|
||||
result = await tool.execute(pattern="search_target", path=str(temp_dir))
|
||||
|
||||
assert result.success
|
||||
# When disabled, should include ignored files
|
||||
assert ".venv" in result.data or "ignored" in result.data
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_grep_override_per_call(self, temp_dir):
|
||||
"""Test per-call gitignore override."""
|
||||
tool = GrepContentTool(honor_gitignore=True)
|
||||
|
||||
# Default behavior - filters
|
||||
result1 = await tool.execute(pattern="search_target", path=str(temp_dir))
|
||||
assert ".venv" not in result1.data
|
||||
|
||||
# Override to disable
|
||||
result2 = await tool.execute(
|
||||
pattern="search_target",
|
||||
path=str(temp_dir),
|
||||
honor_gitignore=False
|
||||
)
|
||||
assert ".venv" in result2.data or "ignored" in result2.data
|
||||
|
||||
|
||||
class TestDefaultIgnores:
|
||||
"""Tests for default ignore patterns (no .gitignore file)."""
|
||||
|
||||
@pytest.fixture
|
||||
def temp_dir_no_gitignore(self):
|
||||
"""Create a directory without .gitignore but with common ignored dirs."""
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
path = Path(tmpdir)
|
||||
|
||||
# Create files
|
||||
(path / "main.py").write_text("# Main")
|
||||
|
||||
# Create commonly ignored directories
|
||||
(path / ".venv").mkdir()
|
||||
(path / ".venv" / "script.py").write_text("venv")
|
||||
|
||||
(path / "__pycache__").mkdir()
|
||||
(path / "__pycache__" / "cache.pyc").write_bytes(b"cache")
|
||||
|
||||
(path / "node_modules").mkdir()
|
||||
(path / "node_modules" / "package.js").write_text("js")
|
||||
|
||||
yield path
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_glob_ignores_defaults_without_gitignore(self, temp_dir_no_gitignore):
|
||||
"""Test that default ignores work even without .gitignore."""
|
||||
tool = GlobFilesTool()
|
||||
|
||||
result = await tool.execute(pattern="**/*", path=str(temp_dir_no_gitignore))
|
||||
|
||||
assert result.success
|
||||
assert "main.py" in result.data
|
||||
assert ".venv" not in result.data
|
||||
assert "__pycache__" not in result.data
|
||||
assert "node_modules" not in result.data
|
||||
@@ -0,0 +1,29 @@
|
||||
"""
|
||||
Health endpoint tests.
|
||||
"""
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_root(client):
|
||||
"""Test root endpoint returns service info."""
|
||||
response = await client.get("/")
|
||||
assert response.status_code == 200
|
||||
|
||||
data = response.json()
|
||||
assert "Webber" in data["service"]
|
||||
assert data["status"] == "healthy"
|
||||
assert "version" in data
|
||||
assert data["docs"] == "/docs"
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_health_check(client):
|
||||
"""Test health check endpoint."""
|
||||
response = await client.get("/health")
|
||||
assert response.status_code == 200
|
||||
|
||||
data = response.json()
|
||||
assert data["status"] == "healthy"
|
||||
assert "Webber" in data["service"]
|
||||
assert "version" in data
|
||||
@@ -0,0 +1,234 @@
|
||||
"""
|
||||
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()
|
||||
Reference in New Issue
Block a user