762 lines
34 KiB
Python
762 lines
34 KiB
Python
"""Unit tests for the four domain-specific 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.timeline_agent import TimelineAgent
|
|
from app.agents.note_agent import NoteAgent
|
|
from app.agents.project_agent import ProjectAgent
|
|
from app.agents.task_agent import TaskAgent
|
|
from app.core.agent_registry import registry
|
|
from app.core.ws_context import clear_client_executor, set_client_executor
|
|
|
|
|
|
# ── WS executor mock ──────────────────────────────────────────────────
|
|
#
|
|
# Tools call execute_on_client() which reads a ContextVar set by the WS
|
|
# handler. In unit tests there is no WS session, so we install a fake
|
|
# executor that returns plausible data for each action type.
|
|
|
|
_FAKE_ROW: dict[str, Any] = {
|
|
"id": "fake-id",
|
|
"title": "Fake Title",
|
|
"name": "Fake Name",
|
|
"status": "todo",
|
|
"priority": "medium",
|
|
"content": "Fake content",
|
|
"date": 1700000000000,
|
|
"taskId": "fake-task-id",
|
|
"author": "Alice",
|
|
"projectId": None,
|
|
}
|
|
|
|
|
|
async def _fake_executor(payload: dict) -> dict:
|
|
action = payload.get("action", "")
|
|
if action == "select":
|
|
return {"rows": []}
|
|
if action == "insert":
|
|
data = payload.get("data", {})
|
|
return {"row": {**_FAKE_ROW, **data}}
|
|
if action == "update":
|
|
data = payload.get("data", {})
|
|
row = {**_FAKE_ROW, "id": data.get("id", "fake-id"), **data.get("updates", {})}
|
|
return {"row": row}
|
|
if action == "delete":
|
|
return {"deleted": True}
|
|
if action == "get":
|
|
data = payload.get("data", {})
|
|
return {"row": {**_FAKE_ROW, "id": data.get("id", "fake-id")}}
|
|
if action == "vector_upsert":
|
|
return {"ok": True}
|
|
return {}
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def ws_executor():
|
|
"""Install a fake WS executor for every test so tools can run without a real WS."""
|
|
set_client_executor(_fake_executor)
|
|
yield
|
|
clear_client_executor()
|
|
|
|
|
|
# ── 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", "timeline_agent", "project_agent", "note_agent"
|
|
}.issubset(names)
|
|
|
|
def test_registry_returns_correct_types(self) -> None:
|
|
assert isinstance(registry.get("task_agent"), TaskAgent)
|
|
assert isinstance(registry.get("timeline_agent"), TimelineAgent)
|
|
assert isinstance(registry.get("project_agent"), ProjectAgent)
|
|
assert isinstance(registry.get("note_agent"), NoteAgent)
|
|
|
|
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 and comments: list, create, update, delete, due-today, comments"
|
|
|
|
def test_get_tools_count(self) -> None:
|
|
assert len(TaskAgent().get_tools()) == 8
|
|
|
|
def test_tool_names(self) -> None:
|
|
names = {t.name for t in TaskAgent().get_tools()}
|
|
assert names == {
|
|
"list_tasks",
|
|
"create_task",
|
|
"update_task",
|
|
"delete_task",
|
|
"list_tasks_due_today",
|
|
"list_task_comments",
|
|
"add_task_comment",
|
|
"delete_task_comment",
|
|
}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_returns_string(self) -> None:
|
|
with patch("app.agents.task_agent.get_llm") 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.get_llm") 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.get_llm") as mock_cls:
|
|
mock_cls.return_value = _mock_llm_with_tool_call(
|
|
"create_task",
|
|
{"title": "Buy groceries", "priority": "low"},
|
|
"Task 'Buy groceries' created.",
|
|
)
|
|
result = await TaskAgent().handle("add a grocery task", {})
|
|
assert result == "Task 'Buy groceries' created."
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_accepts_empty_context(self) -> None:
|
|
with patch("app.agents.task_agent.get_llm") 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_rich_context(self) -> None:
|
|
context = {
|
|
"user_profile": {"id": "u1", "tier": "pro"},
|
|
"recent_tasks": [{"id": "t1", "title": "Old task"}],
|
|
}
|
|
with patch("app.agents.task_agent.get_llm") 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_list_tasks_defaults(self) -> None:
|
|
from app.agents.task_agent import list_tasks
|
|
with patch("app.agents.task_agent.execute_on_client", new_callable=AsyncMock) as m:
|
|
m.return_value = {"rows": []}
|
|
result = await list_tasks.ainvoke({})
|
|
m.assert_called_once_with(
|
|
action="select", table="tasks",
|
|
filters={"projectId": None, "status": None, "search": None, "orderBy": None},
|
|
)
|
|
assert result == "No tasks found matching the given filters."
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_tasks_with_status_filter(self) -> None:
|
|
from app.agents.task_agent import list_tasks
|
|
with patch("app.agents.task_agent.execute_on_client", new_callable=AsyncMock) as m:
|
|
m.return_value = {"rows": []}
|
|
await list_tasks.ainvoke({"status": "done"})
|
|
call_kwargs = m.call_args.kwargs
|
|
assert call_kwargs["filters"]["status"] == "done"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_task_defaults(self) -> None:
|
|
from app.agents.task_agent import create_task
|
|
fake_row = {"id": "t1", "title": "Test task", "status": "todo", "priority": "medium"}
|
|
with patch("app.agents.task_agent.execute_on_client", new_callable=AsyncMock) as m:
|
|
m.return_value = {"row": fake_row}
|
|
result = await create_task.ainvoke({"title": "Test task"})
|
|
call_kwargs = m.call_args.kwargs
|
|
assert call_kwargs["action"] == "insert"
|
|
assert call_kwargs["table"] == "tasks"
|
|
assert call_kwargs["data"]["title"] == "Test task"
|
|
assert call_kwargs["data"]["status"] == "todo"
|
|
assert call_kwargs["data"]["priority"] == "medium"
|
|
assert "Test task" in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_task_with_all_fields(self) -> None:
|
|
from app.agents.task_agent import create_task
|
|
fake_row = {"id": "t1", "title": "Deploy", "status": "in_progress", "priority": "high"}
|
|
with patch("app.agents.task_agent.execute_on_client", new_callable=AsyncMock) as m:
|
|
m.return_value = {"row": fake_row}
|
|
await create_task.ainvoke({
|
|
"title": "Deploy", "priority": "high", "status": "in_progress",
|
|
"project_id": "p1", "is_ai_suggested": 1,
|
|
})
|
|
call_kwargs = m.call_args.kwargs
|
|
assert call_kwargs["data"]["priority"] == "high"
|
|
assert call_kwargs["data"]["status"] == "in_progress"
|
|
assert call_kwargs["data"]["projectId"] == "p1"
|
|
assert call_kwargs["data"]["isAiSuggested"] == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_task_with_status(self) -> None:
|
|
from app.agents.task_agent import update_task
|
|
fake_row = {"id": "t1", "title": "Buy groceries", "status": "done"}
|
|
with patch("app.agents.task_agent.execute_on_client", new_callable=AsyncMock) as m:
|
|
m.return_value = {"row": fake_row}
|
|
result = await update_task.ainvoke({"task_id": "t1", "status": "done"})
|
|
call_kwargs = m.call_args.kwargs
|
|
assert call_kwargs["action"] == "update"
|
|
assert call_kwargs["data"]["id"] == "t1"
|
|
assert call_kwargs["data"]["updates"]["status"] == "done"
|
|
assert "t1" in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_task_empty_updates(self) -> None:
|
|
from app.agents.task_agent import update_task
|
|
fake_row = {"id": "t1", "title": "Task", "status": "todo"}
|
|
with patch("app.agents.task_agent.execute_on_client", new_callable=AsyncMock) as m:
|
|
m.return_value = {"row": fake_row}
|
|
await update_task.ainvoke({"task_id": "t1"})
|
|
call_kwargs = m.call_args.kwargs
|
|
assert call_kwargs["data"]["updates"] == {}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_task(self) -> None:
|
|
from app.agents.task_agent import delete_task
|
|
with patch("app.agents.task_agent.execute_on_client", new_callable=AsyncMock) as m:
|
|
m.return_value = {"deleted": True}
|
|
result = await delete_task.ainvoke({"task_id": "t1"})
|
|
call_kwargs = m.call_args.kwargs
|
|
assert call_kwargs["action"] == "delete"
|
|
assert call_kwargs["table"] == "tasks"
|
|
assert call_kwargs["data"]["id"] == "t1"
|
|
assert "t1" in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_tasks_due_today(self) -> None:
|
|
from app.agents.task_agent import list_tasks_due_today
|
|
with patch("app.agents.task_agent.execute_on_client", new_callable=AsyncMock) as m:
|
|
m.return_value = {"rows": []}
|
|
result = await list_tasks_due_today.ainvoke({})
|
|
call_kwargs = m.call_args.kwargs
|
|
assert call_kwargs["action"] == "select"
|
|
assert call_kwargs["table"] == "tasks"
|
|
assert "dueDateFrom" in call_kwargs["filters"]
|
|
assert result == "No tasks are due today."
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_task_comments(self) -> None:
|
|
from app.agents.task_agent import list_task_comments
|
|
with patch("app.agents.task_agent.execute_on_client", new_callable=AsyncMock) as m:
|
|
m.return_value = {"rows": []}
|
|
result = await list_task_comments.ainvoke({"task_id": "t1"})
|
|
call_kwargs = m.call_args.kwargs
|
|
assert call_kwargs["action"] == "select"
|
|
assert call_kwargs["table"] == "taskComments"
|
|
assert call_kwargs["filters"]["taskId"] == "t1"
|
|
assert "t1" in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_task_comment(self) -> None:
|
|
from app.agents.task_agent import add_task_comment
|
|
fake_row = {"id": "c1", "taskId": "t1", "author": "Alice", "content": "Looks good!"}
|
|
with patch("app.agents.task_agent.execute_on_client", new_callable=AsyncMock) as m:
|
|
m.return_value = {"row": fake_row}
|
|
result = await add_task_comment.ainvoke({
|
|
"task_id": "t1", "author": "Alice", "content": "Looks good!",
|
|
})
|
|
call_kwargs = m.call_args.kwargs
|
|
assert call_kwargs["action"] == "insert"
|
|
assert call_kwargs["table"] == "taskComments"
|
|
assert call_kwargs["data"]["taskId"] == "t1"
|
|
assert call_kwargs["data"]["author"] == "Alice"
|
|
assert call_kwargs["data"]["content"] == "Looks good!"
|
|
assert "Alice" in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_task_comment(self) -> None:
|
|
from app.agents.task_agent import delete_task_comment
|
|
with patch("app.agents.task_agent.execute_on_client", new_callable=AsyncMock) as m:
|
|
m.return_value = {"deleted": True}
|
|
result = await delete_task_comment.ainvoke({"comment_id": "c1"})
|
|
call_kwargs = m.call_args.kwargs
|
|
assert call_kwargs["action"] == "delete"
|
|
assert call_kwargs["table"] == "taskComments"
|
|
assert call_kwargs["data"]["id"] == "c1"
|
|
assert "c1" in result
|
|
|
|
|
|
# ── TimelineAgent ───────────────────────────────────────────────────
|
|
|
|
|
|
class TestTimelineAgent:
|
|
def test_name(self) -> None:
|
|
assert TimelineAgent().get_name() == "timeline_agent"
|
|
|
|
def test_description(self) -> None:
|
|
assert TimelineAgent().get_description() == "Manages project timelines (milestones): list, create, update, delete"
|
|
|
|
def test_get_tools_count(self) -> None:
|
|
assert len(TimelineAgent().get_tools()) == 4
|
|
|
|
def test_tool_names(self) -> None:
|
|
names = {t.name for t in TimelineAgent().get_tools()}
|
|
assert names == {"list_timelines", "create_timeline", "update_timeline", "delete_timeline"}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_no_tool_calls(self) -> None:
|
|
with patch("app.agents.timeline_agent.get_llm") as mock_cls:
|
|
mock_cls.return_value = _mock_llm("No timelines found.")
|
|
result = await TimelineAgent().handle("list timelines", {})
|
|
assert result == "No timelines found."
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_with_create_tool_call(self) -> None:
|
|
with patch("app.agents.timeline_agent.get_llm") as mock_cls:
|
|
mock_cls.return_value = _mock_llm_with_tool_call(
|
|
"create_timeline",
|
|
{"project_id": "p1", "title": "MVP Launch", "date": 1700000000000},
|
|
"Timeline 'MVP Launch' created.",
|
|
)
|
|
result = await TimelineAgent().handle("add MVP timeline", {})
|
|
assert result == "Timeline 'MVP Launch' created."
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_accepts_empty_context(self) -> None:
|
|
with patch("app.agents.timeline_agent.get_llm") as mock_cls:
|
|
mock_cls.return_value = _mock_llm("Done.")
|
|
result = await TimelineAgent().handle("show milestones", {})
|
|
assert isinstance(result, str)
|
|
|
|
|
|
class TestTimelineAgentTools:
|
|
@pytest.mark.asyncio
|
|
async def test_list_timelines_no_project(self) -> None:
|
|
from app.agents.timeline_agent import list_timelines
|
|
with patch("app.agents.timeline_agent.execute_on_client", new_callable=AsyncMock) as m:
|
|
m.return_value = {"rows": []}
|
|
result = await list_timelines.ainvoke({})
|
|
call_kwargs = m.call_args.kwargs
|
|
assert call_kwargs["action"] == "select"
|
|
assert call_kwargs["table"] == "timelines"
|
|
assert call_kwargs["filters"]["projectId"] is None
|
|
assert result == "No timelines found."
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_timelines_with_project(self) -> None:
|
|
from app.agents.timeline_agent import list_timelines
|
|
with patch("app.agents.timeline_agent.execute_on_client", new_callable=AsyncMock) as m:
|
|
m.return_value = {"rows": []}
|
|
await list_timelines.ainvoke({"project_id": "p1"})
|
|
assert m.call_args.kwargs["filters"]["projectId"] == "p1"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_timeline(self) -> None:
|
|
from app.agents.timeline_agent import create_timeline
|
|
fake_row = {"id": "cp1", "title": "Beta release", "date": 1700000000000}
|
|
with patch("app.agents.timeline_agent.execute_on_client", new_callable=AsyncMock) as m:
|
|
m.return_value = {"row": fake_row}
|
|
result = await create_timeline.ainvoke({
|
|
"project_id": "p1", "title": "Beta release", "date": 1700000000000,
|
|
})
|
|
call_kwargs = m.call_args.kwargs
|
|
assert call_kwargs["action"] == "insert"
|
|
assert call_kwargs["table"] == "timelines"
|
|
assert call_kwargs["data"]["projectId"] == "p1"
|
|
assert call_kwargs["data"]["title"] == "Beta release"
|
|
assert call_kwargs["data"]["date"] == 1700000000000
|
|
assert "Beta release" in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_timeline_ai_suggested(self) -> None:
|
|
from app.agents.timeline_agent import create_timeline
|
|
fake_row = {"id": "cp1", "title": "Review", "date": 1700000000000}
|
|
with patch("app.agents.timeline_agent.execute_on_client", new_callable=AsyncMock) as m:
|
|
m.return_value = {"row": fake_row}
|
|
await create_timeline.ainvoke({
|
|
"project_id": "p1", "title": "Review", "date": 1700000000000, "is_ai_suggested": 1,
|
|
})
|
|
call_kwargs = m.call_args.kwargs
|
|
assert call_kwargs["data"]["isAiSuggested"] == 1
|
|
assert call_kwargs["data"]["isApproved"] == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_timeline_approve(self) -> None:
|
|
from app.agents.timeline_agent import update_timeline
|
|
fake_row = {"id": "c1", "title": "MVP", "isApproved": 1}
|
|
with patch("app.agents.timeline_agent.execute_on_client", new_callable=AsyncMock) as m:
|
|
m.return_value = {"row": fake_row}
|
|
result = await update_timeline.ainvoke({"timeline_id": "c1", "is_approved": 1})
|
|
call_kwargs = m.call_args.kwargs
|
|
assert call_kwargs["action"] == "update"
|
|
assert call_kwargs["data"]["id"] == "c1"
|
|
assert call_kwargs["data"]["updates"]["isApproved"] == 1
|
|
assert "c1" in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_timeline_empty_updates(self) -> None:
|
|
from app.agents.timeline_agent import update_timeline
|
|
fake_row = {"id": "c1", "title": "MVP"}
|
|
with patch("app.agents.timeline_agent.execute_on_client", new_callable=AsyncMock) as m:
|
|
m.return_value = {"row": fake_row}
|
|
await update_timeline.ainvoke({"timeline_id": "c1"})
|
|
assert m.call_args.kwargs["data"]["updates"] == {}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_timeline(self) -> None:
|
|
from app.agents.timeline_agent import delete_timeline
|
|
with patch("app.agents.timeline_agent.execute_on_client", new_callable=AsyncMock) as m:
|
|
m.return_value = {"deleted": True}
|
|
result = await delete_timeline.ainvoke({"timeline_id": "c1"})
|
|
call_kwargs = m.call_args.kwargs
|
|
assert call_kwargs["action"] == "delete"
|
|
assert call_kwargs["table"] == "timelines"
|
|
assert call_kwargs["data"]["id"] == "c1"
|
|
assert "c1" in result
|
|
|
|
|
|
# ── ProjectAgent ──────────────────────────────────────────────────────
|
|
|
|
|
|
class TestProjectAgent:
|
|
def test_name(self) -> None:
|
|
assert ProjectAgent().get_name() == "project_agent"
|
|
|
|
def test_description(self) -> None:
|
|
assert ProjectAgent().get_description() == "Manages projects: list, get, create, update, archive, delete"
|
|
|
|
def test_get_tools_count(self) -> None:
|
|
assert len(ProjectAgent().get_tools()) == 6
|
|
|
|
def test_tool_names(self) -> None:
|
|
names = {t.name for t in ProjectAgent().get_tools()}
|
|
assert names == {
|
|
"list_projects",
|
|
"list_all_projects",
|
|
"get_project",
|
|
"create_project",
|
|
"update_project",
|
|
"delete_project",
|
|
}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_no_tool_calls(self) -> None:
|
|
with patch("app.agents.project_agent.get_llm") as mock_cls:
|
|
mock_cls.return_value = _mock_llm("Project Alpha is active.")
|
|
result = await ProjectAgent().handle("show my projects", {})
|
|
assert result == "Project Alpha is active."
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_with_create_project_tool_call(self) -> None:
|
|
with patch("app.agents.project_agent.get_llm") as mock_cls:
|
|
mock_cls.return_value = _mock_llm_with_tool_call(
|
|
"create_project",
|
|
{"name": "Pippo"},
|
|
"Project 'Pippo' created.",
|
|
)
|
|
result = await ProjectAgent().handle("create project Pippo", {})
|
|
assert result == "Project 'Pippo' created."
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_accepts_empty_context(self) -> None:
|
|
with patch("app.agents.project_agent.get_llm") as mock_cls:
|
|
mock_cls.return_value = _mock_llm("Done.")
|
|
result = await ProjectAgent().handle("archive old project", {})
|
|
assert isinstance(result, str)
|
|
|
|
|
|
class TestProjectAgentTools:
|
|
@pytest.mark.asyncio
|
|
async def test_list_projects_defaults(self) -> None:
|
|
from app.agents.project_agent import list_projects
|
|
with patch("app.agents.project_agent.execute_on_client", new_callable=AsyncMock) as m:
|
|
m.return_value = {"rows": []}
|
|
result = await list_projects.ainvoke({})
|
|
call_kwargs = m.call_args.kwargs
|
|
assert call_kwargs["action"] == "select"
|
|
assert call_kwargs["table"] == "projects"
|
|
assert call_kwargs["filters"]["includeArchived"] is False
|
|
assert result == "No projects found."
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_projects_include_archived(self) -> None:
|
|
from app.agents.project_agent import list_projects
|
|
with patch("app.agents.project_agent.execute_on_client", new_callable=AsyncMock) as m:
|
|
m.return_value = {"rows": []}
|
|
await list_projects.ainvoke({"include_archived": 1})
|
|
assert m.call_args.kwargs["filters"]["includeArchived"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_all_projects(self) -> None:
|
|
from app.agents.project_agent import list_all_projects
|
|
with patch("app.agents.project_agent.execute_on_client", new_callable=AsyncMock) as m:
|
|
m.return_value = {"rows": []}
|
|
result = await list_all_projects.ainvoke({})
|
|
call_kwargs = m.call_args.kwargs
|
|
assert call_kwargs["action"] == "select"
|
|
assert call_kwargs["table"] == "projects"
|
|
assert result == "No projects found."
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_project(self) -> None:
|
|
from app.agents.project_agent import get_project
|
|
fake_row = {"id": "p1", "name": "Alpha", "status": "active", "clientId": None}
|
|
with patch("app.agents.project_agent.execute_on_client", new_callable=AsyncMock) as m:
|
|
m.return_value = {"row": fake_row}
|
|
result = await get_project.ainvoke({"project_id": "p1"})
|
|
call_kwargs = m.call_args.kwargs
|
|
assert call_kwargs["action"] == "get"
|
|
assert call_kwargs["table"] == "projects"
|
|
assert call_kwargs["data"]["id"] == "p1"
|
|
assert "Alpha" in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_project_name_only(self) -> None:
|
|
from app.agents.project_agent import create_project
|
|
fake_row = {"id": "p1", "name": "Alpha"}
|
|
with patch("app.agents.project_agent.execute_on_client", new_callable=AsyncMock) as m:
|
|
m.return_value = {"row": fake_row}
|
|
result = await create_project.ainvoke({"name": "Alpha"})
|
|
call_kwargs = m.call_args.kwargs
|
|
assert call_kwargs["action"] == "insert"
|
|
assert call_kwargs["data"]["name"] == "Alpha"
|
|
assert call_kwargs["data"]["clientId"] is None
|
|
assert "Alpha" in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_project_with_client(self) -> None:
|
|
from app.agents.project_agent import create_project
|
|
fake_row = {"id": "p1", "name": "Beta"}
|
|
with patch("app.agents.project_agent.execute_on_client", new_callable=AsyncMock) as m:
|
|
m.return_value = {"row": fake_row}
|
|
await create_project.ainvoke({"name": "Beta", "client_id": "cl1"})
|
|
assert m.call_args.kwargs["data"]["clientId"] == "cl1"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_project_archive(self) -> None:
|
|
from app.agents.project_agent import update_project
|
|
fake_row = {"id": "p1", "name": "Alpha", "status": "archived"}
|
|
with patch("app.agents.project_agent.execute_on_client", new_callable=AsyncMock) as m:
|
|
m.return_value = {"row": fake_row}
|
|
result = await update_project.ainvoke({"project_id": "p1", "status": "archived"})
|
|
call_kwargs = m.call_args.kwargs
|
|
assert call_kwargs["action"] == "update"
|
|
assert call_kwargs["data"]["id"] == "p1"
|
|
assert call_kwargs["data"]["updates"]["status"] == "archived"
|
|
assert "p1" in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_project_empty_updates(self) -> None:
|
|
from app.agents.project_agent import update_project
|
|
fake_row = {"id": "p1", "name": "Alpha", "status": "active"}
|
|
with patch("app.agents.project_agent.execute_on_client", new_callable=AsyncMock) as m:
|
|
m.return_value = {"row": fake_row}
|
|
await update_project.ainvoke({"project_id": "p1"})
|
|
assert m.call_args.kwargs["data"]["updates"] == {}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_project(self) -> None:
|
|
from app.agents.project_agent import delete_project
|
|
with patch("app.agents.project_agent.execute_on_client", new_callable=AsyncMock) as m:
|
|
m.return_value = {"deleted": True}
|
|
result = await delete_project.ainvoke({"project_id": "p1"})
|
|
call_kwargs = m.call_args.kwargs
|
|
assert call_kwargs["action"] == "delete"
|
|
assert call_kwargs["data"]["id"] == "p1"
|
|
assert "p1" in result
|
|
|
|
|
|
# ── NoteAgent ─────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestNoteAgent:
|
|
def test_name(self) -> None:
|
|
assert NoteAgent().get_name() == "note_agent"
|
|
|
|
def test_description(self) -> None:
|
|
assert NoteAgent().get_description() == "Manages notes: list, get, create, update, delete"
|
|
|
|
def test_get_tools_count(self) -> None:
|
|
assert len(NoteAgent().get_tools()) == 5
|
|
|
|
def test_tool_names(self) -> None:
|
|
names = {t.name for t in NoteAgent().get_tools()}
|
|
assert names == {"list_notes", "get_note", "create_note", "update_note", "delete_note"}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_no_tool_calls(self) -> None:
|
|
with patch("app.agents.note_agent.get_llm") as mock_cls:
|
|
mock_cls.return_value = _mock_llm("Note created.")
|
|
result = await NoteAgent().handle("create a note", {})
|
|
assert result == "Note created."
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_with_create_note_tool_call(self) -> None:
|
|
with patch("app.agents.note_agent.get_llm") as mock_cls:
|
|
mock_cls.return_value = _mock_llm_with_tool_call(
|
|
"create_note",
|
|
{"title": "Daily log", "content": "# Today\nAll good."},
|
|
"Note 'Daily log' created.",
|
|
)
|
|
result = await NoteAgent().handle("log today's progress", {})
|
|
assert result == "Note 'Daily log' created."
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_accepts_empty_context(self) -> None:
|
|
with patch("app.agents.note_agent.get_llm") as mock_cls:
|
|
mock_cls.return_value = _mock_llm("Done.")
|
|
result = await NoteAgent().handle("show notes", {})
|
|
assert isinstance(result, str)
|
|
|
|
|
|
class TestNoteAgentTools:
|
|
@pytest.mark.asyncio
|
|
async def test_list_notes_no_project(self) -> None:
|
|
from app.agents.note_agent import list_notes
|
|
with patch("app.agents.note_agent.execute_on_client", new_callable=AsyncMock) as m:
|
|
m.return_value = {"rows": []}
|
|
result = await list_notes.ainvoke({})
|
|
call_kwargs = m.call_args.kwargs
|
|
assert call_kwargs["action"] == "select"
|
|
assert call_kwargs["table"] == "notes"
|
|
assert call_kwargs["filters"]["projectId"] is None
|
|
assert result == "No notes found."
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_notes_with_project(self) -> None:
|
|
from app.agents.note_agent import list_notes
|
|
with patch("app.agents.note_agent.execute_on_client", new_callable=AsyncMock) as m:
|
|
m.return_value = {"rows": []}
|
|
await list_notes.ainvoke({"project_id": "p1"})
|
|
assert m.call_args.kwargs["filters"]["projectId"] == "p1"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_note(self) -> None:
|
|
from app.agents.note_agent import get_note
|
|
fake_row = {"id": "n1", "title": "Daily log", "content": "# Today\nAll good."}
|
|
with patch("app.agents.note_agent.execute_on_client", new_callable=AsyncMock) as m:
|
|
m.return_value = {"row": fake_row}
|
|
result = await get_note.ainvoke({"note_id": "n1"})
|
|
call_kwargs = m.call_args.kwargs
|
|
assert call_kwargs["action"] == "get"
|
|
assert call_kwargs["table"] == "notes"
|
|
assert call_kwargs["data"]["id"] == "n1"
|
|
assert "Daily log" in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_note_minimal(self) -> None:
|
|
from app.agents.note_agent import create_note
|
|
fake_row = {"id": "n1", "title": "Daily log", "projectId": None}
|
|
with patch("app.agents.note_agent.execute_on_client", new_callable=AsyncMock) as m, \
|
|
patch("app.agents.note_agent.embed", new_callable=AsyncMock) as me:
|
|
m.return_value = {"row": fake_row}
|
|
me.return_value = [0.0] * 1536
|
|
result = await create_note.ainvoke({"title": "Daily log", "content": "# Today\nAll good."})
|
|
# First call: insert; second call: vector_upsert
|
|
first_call = m.call_args_list[0].kwargs
|
|
assert first_call["action"] == "insert"
|
|
assert first_call["table"] == "notes"
|
|
assert first_call["data"]["title"] == "Daily log"
|
|
assert first_call["data"]["content"] == "# Today\nAll good."
|
|
assert first_call["data"]["projectId"] is None
|
|
assert "Daily log" in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_note_with_project(self) -> None:
|
|
from app.agents.note_agent import create_note
|
|
fake_row = {"id": "n1", "title": "Sprint notes", "projectId": "p1"}
|
|
with patch("app.agents.note_agent.execute_on_client", new_callable=AsyncMock) as m, \
|
|
patch("app.agents.note_agent.embed", new_callable=AsyncMock) as me:
|
|
m.return_value = {"row": fake_row}
|
|
me.return_value = [0.0] * 1536
|
|
await create_note.ainvoke({"title": "Sprint notes", "content": "## Sprint 1", "project_id": "p1"})
|
|
first_call = m.call_args_list[0].kwargs
|
|
assert first_call["data"]["projectId"] == "p1"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_note_content_only(self) -> None:
|
|
from app.agents.note_agent import update_note
|
|
fake_row = {"id": "n1", "title": "Daily log", "projectId": None}
|
|
with patch("app.agents.note_agent.execute_on_client", new_callable=AsyncMock) as m, \
|
|
patch("app.agents.note_agent.embed", new_callable=AsyncMock) as me:
|
|
m.return_value = {"row": fake_row}
|
|
me.return_value = [0.0] * 1536
|
|
result = await update_note.ainvoke({"note_id": "n1", "content": "# Updated content"})
|
|
first_call = m.call_args_list[0].kwargs
|
|
assert first_call["action"] == "update"
|
|
assert first_call["data"]["id"] == "n1"
|
|
assert first_call["data"]["updates"]["content"] == "# Updated content"
|
|
assert "title" not in first_call["data"]["updates"]
|
|
assert "n1" in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_note_empty_updates(self) -> None:
|
|
from app.agents.note_agent import update_note
|
|
fake_row = {"id": "n1", "title": "Daily log", "projectId": None}
|
|
with patch("app.agents.note_agent.execute_on_client", new_callable=AsyncMock) as m:
|
|
m.return_value = {"row": fake_row}
|
|
await update_note.ainvoke({"note_id": "n1"})
|
|
assert m.call_args.kwargs["data"]["updates"] == {}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_note(self) -> None:
|
|
from app.agents.note_agent import delete_note
|
|
with patch("app.agents.note_agent.execute_on_client", new_callable=AsyncMock) as m:
|
|
m.return_value = {"deleted": True}
|
|
result = await delete_note.ainvoke({"note_id": "n1"})
|
|
call_kwargs = m.call_args.kwargs
|
|
assert call_kwargs["action"] == "delete"
|
|
assert call_kwargs["table"] == "notes"
|
|
assert call_kwargs["data"]["id"] == "n1"
|
|
assert "n1" in result
|