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>
352 lines
12 KiB
Python
352 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).
|
|
"""
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
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."""
|
|
import inspect
|
|
|
|
from src.domains.agents.task.agent import TaskAgentImpl
|
|
|
|
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."""
|
|
import inspect
|
|
|
|
from src.domains.agents.task.agent import TaskAgentImpl
|
|
|
|
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."""
|
|
import inspect
|
|
|
|
from src.shared.logging import trace_span
|
|
|
|
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"]
|