jpmschweitzerandClaude Opus 4.5
d385f47395
- Event-based streaming for task agent - Retry logic when LLM responds without calling tools - Hardened prompts to enforce tool use - Working directory context in all agent prompts - Project paused: local LLMs not capable enough for agentic use Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
348 lines
12 KiB
Python
348 lines
12 KiB
Python
"""
|
|
Tests for agent REST API endpoints.
|
|
|
|
Includes integration tests that verify real code paths work correctly
|
|
without over-mocking (only LLM calls are mocked).
|
|
"""
|
|
import pytest
|
|
from unittest.mock import AsyncMock, patch, MagicMock
|
|
|
|
from src.domains.agents.schemas import PermissionMode
|
|
|
|
|
|
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
|
|
|
|
@pytest.mark.anyio
|
|
async def test_list_agents_returns_descriptions(self, auth_client):
|
|
"""Test that agent list includes descriptions."""
|
|
response = await auth_client.get("/agents/")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
for agent in data["agents"]:
|
|
assert "name" in agent
|
|
assert "description" in agent
|
|
assert len(agent["description"]) > 0
|
|
|
|
|
|
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
|
|
|
|
@pytest.mark.anyio
|
|
async def test_get_agent_empty_name(self, auth_client):
|
|
"""Test getting agent with empty name."""
|
|
response = await auth_client.get("/agents/")
|
|
# This is the list endpoint, should return 200
|
|
assert response.status_code == 200
|
|
|
|
|
|
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
|
|
|
|
@pytest.mark.anyio
|
|
async def test_run_missing_prompt(self, auth_client):
|
|
"""Test running with missing prompt."""
|
|
response = await auth_client.post(
|
|
"/agents/run",
|
|
json={
|
|
"agent_type": "explore",
|
|
"working_dir": "."
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 422
|
|
|
|
@pytest.mark.anyio
|
|
async def test_run_empty_body(self, auth_client):
|
|
"""Test running with empty request body."""
|
|
response = await auth_client.post("/agents/run", json={})
|
|
|
|
assert response.status_code == 422
|
|
|
|
|
|
class TestAgentStreamEndpoint:
|
|
"""Tests for POST /agents/stream endpoint."""
|
|
|
|
@pytest.mark.anyio
|
|
async def test_stream_with_unknown_agent(self, auth_client):
|
|
"""Test streaming unknown agent type."""
|
|
response = await auth_client.post(
|
|
"/agents/stream",
|
|
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_stream_request_validation(self, auth_client):
|
|
"""Test stream request validation."""
|
|
response = await auth_client.post(
|
|
"/agents/stream",
|
|
json={
|
|
"agent_type": "explore"
|
|
# Missing prompt
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 422
|
|
|
|
@pytest.mark.anyio
|
|
async def test_stream_content_type(self, auth_client):
|
|
"""Test that stream endpoint returns correct content type."""
|
|
# Note: This test would require mocking the agent to avoid LLM calls
|
|
# For now, we just verify validation works
|
|
response = await auth_client.post(
|
|
"/agents/stream",
|
|
json={
|
|
"prompt": "test",
|
|
"agent_type": "nonexistent",
|
|
"working_dir": "."
|
|
}
|
|
)
|
|
# Unknown agent returns 400, not streaming
|
|
assert response.status_code == 400
|
|
|
|
|
|
class TestPermissionModeIntegration:
|
|
"""
|
|
Integration tests for permission mode handling.
|
|
|
|
These tests verify that mode strings are correctly converted to enums
|
|
and that the full request->router->agent flow works for each mode.
|
|
"""
|
|
|
|
@pytest.mark.anyio
|
|
async def test_run_with_default_mode(self, auth_client):
|
|
"""Test running agent with default mode passes through correctly."""
|
|
# Mock the agent.run method to avoid LLM calls
|
|
mock_result = MagicMock()
|
|
mock_result.output = "Test response"
|
|
|
|
with patch("src.domains.agents.task.agent.TaskAgentImpl._get_agent_for_mode") as mock_get_agent:
|
|
mock_agent = MagicMock()
|
|
mock_agent.run = AsyncMock(return_value=mock_result)
|
|
mock_get_agent.return_value = mock_agent
|
|
|
|
response = await auth_client.post(
|
|
"/agents/run",
|
|
json={
|
|
"prompt": "test prompt",
|
|
"agent_type": "task",
|
|
"working_dir": ".",
|
|
"mode": "default"
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["success"] is True
|
|
assert data["mode"] == "default"
|
|
|
|
# Verify mode was converted to enum and passed correctly
|
|
mock_get_agent.assert_called_once_with(PermissionMode.default)
|
|
|
|
@pytest.mark.anyio
|
|
async def test_run_with_plan_mode(self, auth_client):
|
|
"""Test running agent with plan mode passes through correctly."""
|
|
mock_result = MagicMock()
|
|
mock_result.output = "Plan response"
|
|
|
|
with patch("src.domains.agents.task.agent.TaskAgentImpl._get_agent_for_mode") as mock_get_agent:
|
|
mock_agent = MagicMock()
|
|
mock_agent.run = AsyncMock(return_value=mock_result)
|
|
mock_get_agent.return_value = mock_agent
|
|
|
|
response = await auth_client.post(
|
|
"/agents/run",
|
|
json={
|
|
"prompt": "test prompt",
|
|
"agent_type": "task",
|
|
"working_dir": ".",
|
|
"mode": "plan"
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["mode"] == "plan"
|
|
mock_get_agent.assert_called_once_with(PermissionMode.plan)
|
|
|
|
@pytest.mark.anyio
|
|
async def test_run_with_auto_accept_mode(self, auth_client):
|
|
"""Test running agent with auto_accept mode passes through correctly."""
|
|
mock_result = MagicMock()
|
|
mock_result.output = "Auto accept response"
|
|
|
|
with patch("src.domains.agents.task.agent.TaskAgentImpl._get_agent_for_mode") as mock_get_agent:
|
|
mock_agent = MagicMock()
|
|
mock_agent.run = AsyncMock(return_value=mock_result)
|
|
mock_get_agent.return_value = mock_agent
|
|
|
|
response = await auth_client.post(
|
|
"/agents/run",
|
|
json={
|
|
"prompt": "test prompt",
|
|
"agent_type": "task",
|
|
"working_dir": ".",
|
|
"mode": "auto_accept"
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["mode"] == "auto_accept"
|
|
mock_get_agent.assert_called_once_with(PermissionMode.auto_accept)
|
|
|
|
@pytest.mark.anyio
|
|
async def test_run_with_invalid_mode(self, auth_client):
|
|
"""Test running agent with invalid mode returns validation error."""
|
|
response = await auth_client.post(
|
|
"/agents/run",
|
|
json={
|
|
"prompt": "test prompt",
|
|
"agent_type": "task",
|
|
"working_dir": ".",
|
|
"mode": "invalid_mode"
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 422
|
|
|
|
@pytest.mark.anyio
|
|
async def test_stream_with_plan_mode(self, auth_client):
|
|
"""Test streaming agent with plan mode passes through correctly."""
|
|
from src.domains.agents.schemas import StreamEvent, StreamEventType
|
|
|
|
async def mock_event_stream(*args, **kwargs):
|
|
"""Mock event-based stream."""
|
|
yield StreamEvent(event=StreamEventType.thinking, message="Starting...")
|
|
yield StreamEvent(event=StreamEventType.response, text="chunk1")
|
|
yield StreamEvent(event=StreamEventType.response, text="chunk2")
|
|
yield StreamEvent(event=StreamEventType.done, mode="plan")
|
|
|
|
with patch("src.domains.agents.task.agent.TaskAgentImpl.run_stream") as mock_run_stream:
|
|
mock_run_stream.return_value = mock_event_stream()
|
|
|
|
response = await auth_client.post(
|
|
"/agents/stream",
|
|
json={
|
|
"prompt": "test prompt",
|
|
"agent_type": "task",
|
|
"working_dir": ".",
|
|
"mode": "plan"
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers["content-type"] == "text/event-stream; charset=utf-8"
|
|
mock_run_stream.assert_called_once()
|
|
|
|
|
|
class TestAgentMethodSignatures:
|
|
"""
|
|
Tests to verify agent method signatures match expected interfaces.
|
|
|
|
These catch issues like passing invalid kwargs to methods.
|
|
"""
|
|
|
|
def test_task_agent_run_accepts_mode(self):
|
|
"""Verify TaskAgentImpl.run() accepts mode parameter."""
|
|
from src.domains.agents.task.agent import TaskAgentImpl
|
|
import inspect
|
|
|
|
sig = inspect.signature(TaskAgentImpl.run)
|
|
params = list(sig.parameters.keys())
|
|
|
|
assert "mode" in params
|
|
# Verify mode has correct type annotation
|
|
mode_param = sig.parameters["mode"]
|
|
assert mode_param.default == PermissionMode.default
|
|
|
|
def test_task_agent_run_stream_accepts_mode(self):
|
|
"""Verify TaskAgentImpl.run_stream() accepts mode parameter."""
|
|
from src.domains.agents.task.agent import TaskAgentImpl
|
|
import inspect
|
|
|
|
sig = inspect.signature(TaskAgentImpl.run_stream)
|
|
params = list(sig.parameters.keys())
|
|
|
|
assert "mode" in params
|
|
mode_param = sig.parameters["mode"]
|
|
assert mode_param.default == PermissionMode.default
|
|
|
|
def test_trace_span_signature(self):
|
|
"""Verify trace_span only accepts expected parameters."""
|
|
from src.shared.logging import trace_span
|
|
import inspect
|
|
|
|
sig = inspect.signature(trace_span.__init__)
|
|
params = list(sig.parameters.keys())
|
|
|
|
# Should only have self, name, logger - not mode or other extras
|
|
assert params == ["self", "name", "logger"]
|