step 6 complete: four specialized agents, all registered and tested
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
389
tests/test_agents.py
Normal file
389
tests/test_agents.py
Normal file
@@ -0,0 +1,389 @@
|
||||
"""Unit tests for all four chat agents with mocked LLM."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from typing import Any
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
import app.agents # noqa: F401 — triggers @registry.register decorators
|
||||
from app.agents.analytics_agent import AnalyticsAgent
|
||||
from app.agents.calendar_agent import CalendarAgent
|
||||
from app.agents.email_agent import EmailAgent
|
||||
from app.agents.task_agent import TaskAgent
|
||||
from app.core.agent_registry import registry
|
||||
|
||||
|
||||
# ── Helpers ──────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _mock_llm(response_text: str) -> MagicMock:
|
||||
"""Return a mock LLM that responds with *response_text* (no tool calls)."""
|
||||
msg = MagicMock()
|
||||
msg.content = response_text
|
||||
msg.tool_calls = []
|
||||
llm = MagicMock()
|
||||
bound = MagicMock()
|
||||
bound.ainvoke = AsyncMock(return_value=msg)
|
||||
llm.bind_tools = MagicMock(return_value=bound)
|
||||
llm.ainvoke = AsyncMock(return_value=msg)
|
||||
return llm
|
||||
|
||||
|
||||
def _mock_llm_with_tool_call(
|
||||
tool_name: str, tool_args: dict[str, Any], final_text: str
|
||||
) -> MagicMock:
|
||||
"""Mock LLM that fires one tool call then returns *final_text*."""
|
||||
tool_msg = MagicMock()
|
||||
tool_msg.content = ""
|
||||
tool_msg.tool_calls = [{"id": "call_1", "name": tool_name, "args": tool_args}]
|
||||
|
||||
final_msg = MagicMock()
|
||||
final_msg.content = final_text
|
||||
final_msg.tool_calls = []
|
||||
|
||||
bound = MagicMock()
|
||||
bound.ainvoke = AsyncMock(side_effect=[tool_msg, final_msg])
|
||||
|
||||
llm = MagicMock()
|
||||
llm.bind_tools = MagicMock(return_value=bound)
|
||||
llm.ainvoke = AsyncMock(return_value=final_msg)
|
||||
return llm
|
||||
|
||||
|
||||
# ── Registration ──────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestAgentRegistration:
|
||||
def test_all_agents_registered(self) -> None:
|
||||
names = {a["name"] for a in registry.list_agents()}
|
||||
assert {"task_agent", "calendar_agent", "email_agent", "analytics_agent"}.issubset(
|
||||
names
|
||||
)
|
||||
|
||||
def test_registry_returns_correct_types(self) -> None:
|
||||
assert isinstance(registry.get("task_agent"), TaskAgent)
|
||||
assert isinstance(registry.get("calendar_agent"), CalendarAgent)
|
||||
assert isinstance(registry.get("email_agent"), EmailAgent)
|
||||
assert isinstance(registry.get("analytics_agent"), AnalyticsAgent)
|
||||
|
||||
def test_descriptions_present(self) -> None:
|
||||
for agent_info in registry.list_agents():
|
||||
assert agent_info["description"], f"Empty description: {agent_info['name']}"
|
||||
|
||||
|
||||
# ── TaskAgent ─────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestTaskAgent:
|
||||
def test_name(self) -> None:
|
||||
assert TaskAgent().get_name() == "task_agent"
|
||||
|
||||
def test_description(self) -> None:
|
||||
assert TaskAgent().get_description() == "Manages tasks: create, update, list, suggest"
|
||||
|
||||
def test_get_tools_count(self) -> None:
|
||||
assert len(TaskAgent().get_tools()) == 4
|
||||
|
||||
def test_tool_names(self) -> None:
|
||||
names = {t.name for t in TaskAgent().get_tools()}
|
||||
assert names == {"create_task", "update_task", "list_tasks", "suggest_tasks"}
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_returns_string(self) -> None:
|
||||
with patch("app.agents.task_agent.ChatOpenAI") as mock_cls:
|
||||
mock_cls.return_value = _mock_llm("Task created.")
|
||||
result = await TaskAgent().handle("create a task", {})
|
||||
assert isinstance(result, str)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_no_tool_calls(self) -> None:
|
||||
with patch("app.agents.task_agent.ChatOpenAI") as mock_cls:
|
||||
mock_cls.return_value = _mock_llm("Here are your tasks.")
|
||||
result = await TaskAgent().handle("list my tasks", {})
|
||||
assert result == "Here are your tasks."
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_with_create_task_tool_call(self) -> None:
|
||||
with patch("app.agents.task_agent.ChatOpenAI") as mock_cls:
|
||||
mock_cls.return_value = _mock_llm_with_tool_call(
|
||||
"create_task",
|
||||
{"title": "Buy groceries", "priority": "low"},
|
||||
"Task 'Buy groceries' created with low priority.",
|
||||
)
|
||||
result = await TaskAgent().handle("add a grocery task", {})
|
||||
assert result == "Task 'Buy groceries' created with low priority."
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_accepts_empty_context(self) -> None:
|
||||
with patch("app.agents.task_agent.ChatOpenAI") as mock_cls:
|
||||
mock_cls.return_value = _mock_llm("Done.")
|
||||
result = await TaskAgent().handle("help", {})
|
||||
assert isinstance(result, str)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_accepts_partial_context(self) -> None:
|
||||
with patch("app.agents.task_agent.ChatOpenAI") as mock_cls:
|
||||
mock_cls.return_value = _mock_llm("Done.")
|
||||
result = await TaskAgent().handle("list tasks", {"user_profile": {"id": "u1"}})
|
||||
assert isinstance(result, str)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_accepts_rich_context(self) -> None:
|
||||
context = {
|
||||
"user_profile": {"id": "u1", "tier": "pro"},
|
||||
"recent_tasks": [{"id": "t1", "title": "Old task"}],
|
||||
"relevant_documents": ["doc1"],
|
||||
"extra_plugin_data": {"batch_id": "b1"},
|
||||
}
|
||||
with patch("app.agents.task_agent.ChatOpenAI") as mock_cls:
|
||||
mock_cls.return_value = _mock_llm("Tasks listed.")
|
||||
result = await TaskAgent().handle("show tasks", context)
|
||||
assert isinstance(result, str)
|
||||
|
||||
|
||||
class TestTaskAgentTools:
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_task_returns_valid_json(self) -> None:
|
||||
from app.agents.task_agent import create_task
|
||||
result = await create_task.ainvoke({"title": "Test task", "priority": "high"})
|
||||
data = json.loads(result)
|
||||
assert data["action"] == "create_record"
|
||||
assert data["table"] == "tasks"
|
||||
assert data["data"]["title"] == "Test task"
|
||||
assert data["data"]["priority"] == "high"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_task_returns_valid_json(self) -> None:
|
||||
from app.agents.task_agent import update_task
|
||||
result = await update_task.ainvoke(
|
||||
{"task_id": "t1", "updates": '{"priority": "urgent"}'}
|
||||
)
|
||||
data = json.loads(result)
|
||||
assert data["action"] == "update_record"
|
||||
assert data["data"]["id"] == "t1"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_tasks_returns_valid_json(self) -> None:
|
||||
from app.agents.task_agent import list_tasks
|
||||
result = await list_tasks.ainvoke({"status": "open"})
|
||||
data = json.loads(result)
|
||||
assert data["action"] == "list"
|
||||
assert data["table"] == "tasks"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_suggest_tasks_returns_valid_json(self) -> None:
|
||||
from app.agents.task_agent import suggest_tasks
|
||||
result = await suggest_tasks.ainvoke({"context": "lots of meetings this week"})
|
||||
data = json.loads(result)
|
||||
assert data["action"] == "suggest"
|
||||
|
||||
|
||||
# ── CalendarAgent ─────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestCalendarAgent:
|
||||
def test_name(self) -> None:
|
||||
assert CalendarAgent().get_name() == "calendar_agent"
|
||||
|
||||
def test_description(self) -> None:
|
||||
assert CalendarAgent().get_description() == "Calendar management: events, conflicts, scheduling"
|
||||
|
||||
def test_get_tools_count(self) -> None:
|
||||
assert len(CalendarAgent().get_tools()) == 3
|
||||
|
||||
def test_tool_names(self) -> None:
|
||||
names = {t.name for t in CalendarAgent().get_tools()}
|
||||
assert names == {"list_events", "detect_conflicts", "suggest_reschedule"}
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_no_tool_calls(self) -> None:
|
||||
with patch("app.agents.calendar_agent.ChatOpenAI") as mock_cls:
|
||||
mock_cls.return_value = _mock_llm("No conflicts found.")
|
||||
result = await CalendarAgent().handle("check my schedule", {})
|
||||
assert result == "No conflicts found."
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_with_list_events_tool_call(self) -> None:
|
||||
with patch("app.agents.calendar_agent.ChatOpenAI") as mock_cls:
|
||||
mock_cls.return_value = _mock_llm_with_tool_call(
|
||||
"list_events",
|
||||
{"date_range": "2024-01-01/2024-01-07"},
|
||||
"You have 3 events next week.",
|
||||
)
|
||||
result = await CalendarAgent().handle("what events do I have?", {})
|
||||
assert result == "You have 3 events next week."
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_accepts_empty_context(self) -> None:
|
||||
with patch("app.agents.calendar_agent.ChatOpenAI") as mock_cls:
|
||||
mock_cls.return_value = _mock_llm("Done.")
|
||||
result = await CalendarAgent().handle("reschedule meeting", {})
|
||||
assert isinstance(result, str)
|
||||
|
||||
|
||||
class TestCalendarAgentTools:
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_events_returns_valid_json(self) -> None:
|
||||
from app.agents.calendar_agent import list_events
|
||||
result = await list_events.ainvoke({"date_range": "2024-01-01/2024-01-07"})
|
||||
data = json.loads(result)
|
||||
assert data["action"] == "list"
|
||||
assert data["table"] == "events"
|
||||
assert data["filters"]["date_range"] == "2024-01-01/2024-01-07"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_detect_conflicts_returns_valid_json(self) -> None:
|
||||
from app.agents.calendar_agent import detect_conflicts
|
||||
result = await detect_conflicts.ainvoke({"events": "[]"})
|
||||
data = json.loads(result)
|
||||
assert data["action"] == "analyse"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_suggest_reschedule_returns_valid_json(self) -> None:
|
||||
from app.agents.calendar_agent import suggest_reschedule
|
||||
result = await suggest_reschedule.ainvoke({"conflict": '{"event": "standup"}'})
|
||||
data = json.loads(result)
|
||||
assert data["action"] == "suggest_reschedule"
|
||||
|
||||
|
||||
# ── EmailAgent ────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestEmailAgent:
|
||||
def test_name(self) -> None:
|
||||
assert EmailAgent().get_name() == "email_agent"
|
||||
|
||||
def test_description(self) -> None:
|
||||
assert EmailAgent().get_description() == "Email analysis: classify, extract actions, draft responses"
|
||||
|
||||
def test_get_tools_count(self) -> None:
|
||||
assert len(EmailAgent().get_tools()) == 3
|
||||
|
||||
def test_tool_names(self) -> None:
|
||||
names = {t.name for t in EmailAgent().get_tools()}
|
||||
assert names == {"classify_email", "extract_action_items", "draft_response"}
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_no_tool_calls(self) -> None:
|
||||
with patch("app.agents.email_agent.ChatOpenAI") as mock_cls:
|
||||
mock_cls.return_value = _mock_llm("Email classified as action_required.")
|
||||
result = await EmailAgent().handle("classify this email", {})
|
||||
assert result == "Email classified as action_required."
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_with_classify_tool_call(self) -> None:
|
||||
with patch("app.agents.email_agent.ChatOpenAI") as mock_cls:
|
||||
mock_cls.return_value = _mock_llm_with_tool_call(
|
||||
"classify_email",
|
||||
{"metadata": '{"subject": "URGENT: action needed"}'},
|
||||
"This email requires immediate action.",
|
||||
)
|
||||
result = await EmailAgent().handle("what is this email about?", {})
|
||||
assert result == "This email requires immediate action."
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_accepts_empty_context(self) -> None:
|
||||
with patch("app.agents.email_agent.ChatOpenAI") as mock_cls:
|
||||
mock_cls.return_value = _mock_llm("Done.")
|
||||
result = await EmailAgent().handle("draft a reply", {})
|
||||
assert isinstance(result, str)
|
||||
|
||||
|
||||
class TestEmailAgentTools:
|
||||
@pytest.mark.asyncio
|
||||
async def test_classify_email_returns_valid_json(self) -> None:
|
||||
from app.agents.email_agent import classify_email
|
||||
result = await classify_email.ainvoke({"metadata": '{"subject": "Meeting"}' })
|
||||
data = json.loads(result)
|
||||
assert data["action"] == "classify"
|
||||
assert "result" in data
|
||||
assert "category" in data["result"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_extract_action_items_returns_valid_json(self) -> None:
|
||||
from app.agents.email_agent import extract_action_items
|
||||
result = await extract_action_items.ainvoke({"metadata": '{"subject": "Follow up"}'})
|
||||
data = json.loads(result)
|
||||
assert data["action"] == "extract"
|
||||
assert "action_items" in data["result"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_draft_response_returns_valid_json(self) -> None:
|
||||
from app.agents.email_agent import draft_response
|
||||
result = await draft_response.ainvoke({"thread_context": '{"thread_id": "t1"}'})
|
||||
data = json.loads(result)
|
||||
assert data["action"] == "draft"
|
||||
|
||||
|
||||
# ── AnalyticsAgent ────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestAnalyticsAgent:
|
||||
def test_name(self) -> None:
|
||||
assert AnalyticsAgent().get_name() == "analytics_agent"
|
||||
|
||||
def test_description(self) -> None:
|
||||
assert AnalyticsAgent().get_description() == "Workspace analytics: metrics, reports, trends"
|
||||
|
||||
def test_get_tools_count(self) -> None:
|
||||
assert len(AnalyticsAgent().get_tools()) == 3
|
||||
|
||||
def test_tool_names(self) -> None:
|
||||
names = {t.name for t in AnalyticsAgent().get_tools()}
|
||||
assert names == {"calculate_metrics", "generate_report", "trend_analysis"}
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_no_tool_calls(self) -> None:
|
||||
with patch("app.agents.analytics_agent.ChatOpenAI") as mock_cls:
|
||||
mock_cls.return_value = _mock_llm("Completion rate is 78%.")
|
||||
result = await AnalyticsAgent().handle("show my metrics", {})
|
||||
assert result == "Completion rate is 78%."
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_with_generate_report_tool_call(self) -> None:
|
||||
with patch("app.agents.analytics_agent.ChatOpenAI") as mock_cls:
|
||||
mock_cls.return_value = _mock_llm_with_tool_call(
|
||||
"generate_report",
|
||||
{"period": "last_7_days", "data": "[]"},
|
||||
"Weekly report: 12 tasks completed, 2 overdue.",
|
||||
)
|
||||
result = await AnalyticsAgent().handle("weekly report", {})
|
||||
assert result == "Weekly report: 12 tasks completed, 2 overdue."
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_accepts_empty_context(self) -> None:
|
||||
with patch("app.agents.analytics_agent.ChatOpenAI") as mock_cls:
|
||||
mock_cls.return_value = _mock_llm("Done.")
|
||||
result = await AnalyticsAgent().handle("analyse trends", {})
|
||||
assert isinstance(result, str)
|
||||
|
||||
|
||||
class TestAnalyticsAgentTools:
|
||||
@pytest.mark.asyncio
|
||||
async def test_calculate_metrics_returns_valid_json(self) -> None:
|
||||
from app.agents.analytics_agent import calculate_metrics
|
||||
result = await calculate_metrics.ainvoke({"task_data": "[]"})
|
||||
data = json.loads(result)
|
||||
assert data["action"] == "calculate"
|
||||
assert "result" in data
|
||||
assert "completion_rate" in data["result"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_generate_report_returns_valid_json(self) -> None:
|
||||
from app.agents.analytics_agent import generate_report
|
||||
result = await generate_report.ainvoke({"period": "last_7_days", "data": "[]"})
|
||||
data = json.loads(result)
|
||||
assert data["action"] == "report"
|
||||
assert data["period"] == "last_7_days"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_trend_analysis_returns_valid_json(self) -> None:
|
||||
from app.agents.analytics_agent import trend_analysis
|
||||
result = await trend_analysis.ainvoke({"data_points": "[]"})
|
||||
data = json.loads(result)
|
||||
assert data["action"] == "trend"
|
||||
assert "result" in data
|
||||
assert "anomalies" in data["result"]
|
||||
Reference in New Issue
Block a user