roberto
c8ef7b119b
- Updated `TestModuleSingletons` in `test_execution_plan.py` to reflect new agent templates and playbook names. - Changed assertions in playbook tests to match updated templates and agents. - Introduced `test_storage.py` to cover the storage layer, including encryption, BlobStore, and VectorStore functionalities. - Added tests for S3 interactions, ensuring upload, download, delete, and list operations work as expected. - Implemented mock tests for Pinecone and Qdrant vector stores to validate upsert, search, and delete operations.
621 lines
25 KiB
Python
621 lines
25 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.checkpoint_agent import CheckpointAgent
|
|
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
|
|
|
|
|
|
# ── 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", "checkpoint_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("checkpoint_agent"), CheckpointAgent)
|
|
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.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.",
|
|
)
|
|
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.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_rich_context(self) -> None:
|
|
context = {
|
|
"user_profile": {"id": "u1", "tier": "pro"},
|
|
"recent_tasks": [{"id": "t1", "title": "Old task"}],
|
|
}
|
|
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_list_tasks_defaults(self) -> None:
|
|
from app.agents.task_agent import list_tasks
|
|
result = await list_tasks.ainvoke({})
|
|
data = json.loads(result)
|
|
assert data["action"] == "list"
|
|
assert data["table"] == "tasks"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_tasks_with_status_filter(self) -> None:
|
|
from app.agents.task_agent import list_tasks
|
|
result = await list_tasks.ainvoke({"status": "done"})
|
|
data = json.loads(result)
|
|
assert data["filters"]["status"] == "done"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_task_defaults(self) -> None:
|
|
from app.agents.task_agent import create_task
|
|
result = await create_task.ainvoke({"title": "Test task"})
|
|
data = json.loads(result)
|
|
assert data["action"] == "create_record"
|
|
assert data["table"] == "tasks"
|
|
assert data["data"]["title"] == "Test task"
|
|
assert data["data"]["status"] == "todo"
|
|
assert data["data"]["priority"] == "medium"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_task_with_all_fields(self) -> None:
|
|
from app.agents.task_agent import create_task
|
|
result = await create_task.ainvoke({
|
|
"title": "Deploy",
|
|
"priority": "high",
|
|
"status": "in_progress",
|
|
"project_id": "p1",
|
|
"is_ai_suggested": 1,
|
|
})
|
|
data = json.loads(result)
|
|
assert data["data"]["priority"] == "high"
|
|
assert data["data"]["status"] == "in_progress"
|
|
assert data["data"]["projectId"] == "p1"
|
|
assert data["data"]["isAiSuggested"] == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_task_with_status(self) -> None:
|
|
from app.agents.task_agent import update_task
|
|
result = await update_task.ainvoke({"task_id": "t1", "status": "done"})
|
|
data = json.loads(result)
|
|
assert data["action"] == "update_record"
|
|
assert data["data"]["id"] == "t1"
|
|
assert data["data"]["updates"]["status"] == "done"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_task_empty_updates(self) -> None:
|
|
from app.agents.task_agent import update_task
|
|
result = await update_task.ainvoke({"task_id": "t1"})
|
|
data = json.loads(result)
|
|
assert data["data"]["updates"] == {}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_task(self) -> None:
|
|
from app.agents.task_agent import delete_task
|
|
result = await delete_task.ainvoke({"task_id": "t1"})
|
|
data = json.loads(result)
|
|
assert data["action"] == "delete_record"
|
|
assert data["table"] == "tasks"
|
|
assert data["data"]["id"] == "t1"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_tasks_due_today(self) -> None:
|
|
from app.agents.task_agent import list_tasks_due_today
|
|
result = await list_tasks_due_today.ainvoke({})
|
|
data = json.loads(result)
|
|
assert data["action"] == "list_due_today"
|
|
assert data["table"] == "tasks"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_task_comments(self) -> None:
|
|
from app.agents.task_agent import list_task_comments
|
|
result = await list_task_comments.ainvoke({"task_id": "t1"})
|
|
data = json.loads(result)
|
|
assert data["action"] == "list"
|
|
assert data["table"] == "taskComments"
|
|
assert data["filters"]["taskId"] == "t1"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_task_comment(self) -> None:
|
|
from app.agents.task_agent import add_task_comment
|
|
result = await add_task_comment.ainvoke({
|
|
"task_id": "t1",
|
|
"author": "Alice",
|
|
"content": "Looks good!",
|
|
})
|
|
data = json.loads(result)
|
|
assert data["action"] == "create_record"
|
|
assert data["table"] == "taskComments"
|
|
assert data["data"]["taskId"] == "t1"
|
|
assert data["data"]["author"] == "Alice"
|
|
assert data["data"]["content"] == "Looks good!"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_task_comment(self) -> None:
|
|
from app.agents.task_agent import delete_task_comment
|
|
result = await delete_task_comment.ainvoke({"comment_id": "c1"})
|
|
data = json.loads(result)
|
|
assert data["action"] == "delete_record"
|
|
assert data["table"] == "taskComments"
|
|
assert data["data"]["id"] == "c1"
|
|
|
|
|
|
# ── CheckpointAgent ───────────────────────────────────────────────────
|
|
|
|
|
|
class TestCheckpointAgent:
|
|
def test_name(self) -> None:
|
|
assert CheckpointAgent().get_name() == "checkpoint_agent"
|
|
|
|
def test_description(self) -> None:
|
|
assert CheckpointAgent().get_description() == "Manages project checkpoints (milestones): list, create, update, delete"
|
|
|
|
def test_get_tools_count(self) -> None:
|
|
assert len(CheckpointAgent().get_tools()) == 4
|
|
|
|
def test_tool_names(self) -> None:
|
|
names = {t.name for t in CheckpointAgent().get_tools()}
|
|
assert names == {"list_checkpoints", "create_checkpoint", "update_checkpoint", "delete_checkpoint"}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_no_tool_calls(self) -> None:
|
|
with patch("app.agents.checkpoint_agent.ChatOpenAI") as mock_cls:
|
|
mock_cls.return_value = _mock_llm("No checkpoints found.")
|
|
result = await CheckpointAgent().handle("list checkpoints", {})
|
|
assert result == "No checkpoints found."
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_with_create_tool_call(self) -> None:
|
|
with patch("app.agents.checkpoint_agent.ChatOpenAI") as mock_cls:
|
|
mock_cls.return_value = _mock_llm_with_tool_call(
|
|
"create_checkpoint",
|
|
{"project_id": "p1", "title": "MVP Launch", "date": 1700000000000},
|
|
"Checkpoint 'MVP Launch' created.",
|
|
)
|
|
result = await CheckpointAgent().handle("add MVP checkpoint", {})
|
|
assert result == "Checkpoint 'MVP Launch' created."
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_accepts_empty_context(self) -> None:
|
|
with patch("app.agents.checkpoint_agent.ChatOpenAI") as mock_cls:
|
|
mock_cls.return_value = _mock_llm("Done.")
|
|
result = await CheckpointAgent().handle("show milestones", {})
|
|
assert isinstance(result, str)
|
|
|
|
|
|
class TestCheckpointAgentTools:
|
|
@pytest.mark.asyncio
|
|
async def test_list_checkpoints_no_project(self) -> None:
|
|
from app.agents.checkpoint_agent import list_checkpoints
|
|
result = await list_checkpoints.ainvoke({})
|
|
data = json.loads(result)
|
|
assert data["action"] == "list"
|
|
assert data["table"] == "checkpoints"
|
|
assert data["filters"]["projectId"] is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_checkpoints_with_project(self) -> None:
|
|
from app.agents.checkpoint_agent import list_checkpoints
|
|
result = await list_checkpoints.ainvoke({"project_id": "p1"})
|
|
data = json.loads(result)
|
|
assert data["filters"]["projectId"] == "p1"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_checkpoint(self) -> None:
|
|
from app.agents.checkpoint_agent import create_checkpoint
|
|
result = await create_checkpoint.ainvoke({
|
|
"project_id": "p1",
|
|
"title": "Beta release",
|
|
"date": 1700000000000,
|
|
})
|
|
data = json.loads(result)
|
|
assert data["action"] == "create_record"
|
|
assert data["table"] == "checkpoints"
|
|
assert data["data"]["projectId"] == "p1"
|
|
assert data["data"]["title"] == "Beta release"
|
|
assert data["data"]["date"] == 1700000000000
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_checkpoint_ai_suggested(self) -> None:
|
|
from app.agents.checkpoint_agent import create_checkpoint
|
|
result = await create_checkpoint.ainvoke({
|
|
"project_id": "p1",
|
|
"title": "Review",
|
|
"date": 1700000000000,
|
|
"is_ai_suggested": 1,
|
|
})
|
|
data = json.loads(result)
|
|
assert data["data"]["isAiSuggested"] == 1
|
|
assert data["data"]["isApproved"] == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_checkpoint_approve(self) -> None:
|
|
from app.agents.checkpoint_agent import update_checkpoint
|
|
result = await update_checkpoint.ainvoke({
|
|
"checkpoint_id": "c1",
|
|
"is_approved": 1,
|
|
})
|
|
data = json.loads(result)
|
|
assert data["action"] == "update_record"
|
|
assert data["data"]["id"] == "c1"
|
|
assert data["data"]["updates"]["isApproved"] == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_checkpoint_empty_updates(self) -> None:
|
|
from app.agents.checkpoint_agent import update_checkpoint
|
|
result = await update_checkpoint.ainvoke({"checkpoint_id": "c1"})
|
|
data = json.loads(result)
|
|
assert data["data"]["updates"] == {}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_checkpoint(self) -> None:
|
|
from app.agents.checkpoint_agent import delete_checkpoint
|
|
result = await delete_checkpoint.ainvoke({"checkpoint_id": "c1"})
|
|
data = json.loads(result)
|
|
assert data["action"] == "delete_record"
|
|
assert data["table"] == "checkpoints"
|
|
assert data["data"]["id"] == "c1"
|
|
|
|
|
|
# ── 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.ChatOpenAI") 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.ChatOpenAI") 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.ChatOpenAI") 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
|
|
result = await list_projects.ainvoke({})
|
|
data = json.loads(result)
|
|
assert data["action"] == "list"
|
|
assert data["table"] == "projects"
|
|
assert data["filters"]["includeArchived"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_projects_include_archived(self) -> None:
|
|
from app.agents.project_agent import list_projects
|
|
result = await list_projects.ainvoke({"include_archived": 1})
|
|
data = json.loads(result)
|
|
assert data["filters"]["includeArchived"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_all_projects(self) -> None:
|
|
from app.agents.project_agent import list_all_projects
|
|
result = await list_all_projects.ainvoke({})
|
|
data = json.loads(result)
|
|
assert data["action"] == "list_all"
|
|
assert data["table"] == "projects"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_project(self) -> None:
|
|
from app.agents.project_agent import get_project
|
|
result = await get_project.ainvoke({"project_id": "p1"})
|
|
data = json.loads(result)
|
|
assert data["action"] == "get"
|
|
assert data["table"] == "projects"
|
|
assert data["data"]["id"] == "p1"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_project_name_only(self) -> None:
|
|
from app.agents.project_agent import create_project
|
|
result = await create_project.ainvoke({"name": "Alpha"})
|
|
data = json.loads(result)
|
|
assert data["action"] == "create_record"
|
|
assert data["data"]["name"] == "Alpha"
|
|
assert data["data"]["clientId"] is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_project_with_client(self) -> None:
|
|
from app.agents.project_agent import create_project
|
|
result = await create_project.ainvoke({"name": "Beta", "client_id": "cl1"})
|
|
data = json.loads(result)
|
|
assert data["data"]["clientId"] == "cl1"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_project_archive(self) -> None:
|
|
from app.agents.project_agent import update_project
|
|
result = await update_project.ainvoke({"project_id": "p1", "status": "archived"})
|
|
data = json.loads(result)
|
|
assert data["action"] == "update_record"
|
|
assert data["data"]["id"] == "p1"
|
|
assert data["data"]["updates"]["status"] == "archived"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_project_empty_updates(self) -> None:
|
|
from app.agents.project_agent import update_project
|
|
result = await update_project.ainvoke({"project_id": "p1"})
|
|
data = json.loads(result)
|
|
assert data["data"]["updates"] == {}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_project(self) -> None:
|
|
from app.agents.project_agent import delete_project
|
|
result = await delete_project.ainvoke({"project_id": "p1"})
|
|
data = json.loads(result)
|
|
assert data["action"] == "delete_record"
|
|
assert data["data"]["id"] == "p1"
|
|
|
|
|
|
# ── 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.ChatOpenAI") 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.ChatOpenAI") 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.ChatOpenAI") 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
|
|
result = await list_notes.ainvoke({})
|
|
data = json.loads(result)
|
|
assert data["action"] == "list"
|
|
assert data["table"] == "notes"
|
|
assert data["filters"]["projectId"] is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_notes_with_project(self) -> None:
|
|
from app.agents.note_agent import list_notes
|
|
result = await list_notes.ainvoke({"project_id": "p1"})
|
|
data = json.loads(result)
|
|
assert data["filters"]["projectId"] == "p1"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_note(self) -> None:
|
|
from app.agents.note_agent import get_note
|
|
result = await get_note.ainvoke({"note_id": "n1"})
|
|
data = json.loads(result)
|
|
assert data["action"] == "get"
|
|
assert data["table"] == "notes"
|
|
assert data["data"]["id"] == "n1"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_note_minimal(self) -> None:
|
|
from app.agents.note_agent import create_note
|
|
result = await create_note.ainvoke({
|
|
"title": "Daily log",
|
|
"content": "# Today\nAll good.",
|
|
})
|
|
data = json.loads(result)
|
|
assert data["action"] == "create_record"
|
|
assert data["table"] == "notes"
|
|
assert data["data"]["title"] == "Daily log"
|
|
assert data["data"]["content"] == "# Today\nAll good."
|
|
assert data["data"]["projectId"] is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_note_with_project(self) -> None:
|
|
from app.agents.note_agent import create_note
|
|
result = await create_note.ainvoke({
|
|
"title": "Sprint notes",
|
|
"content": "## Sprint 1",
|
|
"project_id": "p1",
|
|
})
|
|
data = json.loads(result)
|
|
assert data["data"]["projectId"] == "p1"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_note_content_only(self) -> None:
|
|
from app.agents.note_agent import update_note
|
|
result = await update_note.ainvoke({
|
|
"note_id": "n1",
|
|
"content": "# Updated content",
|
|
})
|
|
data = json.loads(result)
|
|
assert data["action"] == "update_record"
|
|
assert data["data"]["id"] == "n1"
|
|
assert data["data"]["updates"]["content"] == "# Updated content"
|
|
assert "title" not in data["data"]["updates"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_note_empty_updates(self) -> None:
|
|
from app.agents.note_agent import update_note
|
|
result = await update_note.ainvoke({"note_id": "n1"})
|
|
data = json.loads(result)
|
|
assert data["data"]["updates"] == {}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_note(self) -> None:
|
|
from app.agents.note_agent import delete_note
|
|
result = await delete_note.ainvoke({"note_id": "n1"})
|
|
data = json.loads(result)
|
|
assert data["action"] == "delete_record"
|
|
assert data["table"] == "notes"
|
|
assert data["data"]["id"] == "n1"
|