353 lines
11 KiB
Python
353 lines
11 KiB
Python
"""Tests for CalendarTool — Agent tool wrapper for ReAct integration (U3)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from agentkit.calendar.db import get_event_tags, init_calendar_db
|
|
from agentkit.calendar.service import CalendarService
|
|
from agentkit.tools.calendar_tool import CalendarTool
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def service(tmp_path: Path) -> CalendarService:
|
|
"""Provide a CalendarService backed by a temp DB."""
|
|
db_path = tmp_path / "test.db"
|
|
asyncio.run(init_calendar_db(db_path))
|
|
return CalendarService(db_path=db_path)
|
|
|
|
|
|
@pytest.fixture
|
|
def tool(service: CalendarService) -> CalendarTool:
|
|
return CalendarTool(calendar_service=service)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# create_event
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def test_create_event_action_returns_success(
|
|
tool: CalendarTool, service: CalendarService
|
|
) -> None:
|
|
"""create_event returns success and persists the event with source='agent'."""
|
|
result = await tool.execute(
|
|
action="create_event",
|
|
user_id="user-1",
|
|
title="Sprint Planning",
|
|
start_time="2026-07-01T10:00:00+00:00",
|
|
end_time="2026-07-01T11:00:00+00:00",
|
|
description="Bi-weekly sprint planning",
|
|
location="Room A",
|
|
)
|
|
assert result["success"] is True
|
|
event_dict = result["event"]
|
|
assert event_dict["title"] == "Sprint Planning"
|
|
assert event_dict["source"] == "agent"
|
|
assert event_dict["user_id"] == "user-1"
|
|
|
|
# Verify persisted in DB
|
|
fetched = await service.get_event(event_dict["id"])
|
|
assert fetched is not None
|
|
assert fetched.title == "Sprint Planning"
|
|
assert fetched.source == "agent"
|
|
|
|
|
|
async def test_create_event_with_recurrence_sets_rrule(
|
|
tool: CalendarTool, service: CalendarService
|
|
) -> None:
|
|
"""rrule param is stored correctly on the event."""
|
|
rrule = "FREQ=WEEKLY;BYDAY=MO;COUNT=10"
|
|
result = await tool.execute(
|
|
action="create_event",
|
|
user_id="user-1",
|
|
title="Weekly Standup",
|
|
start_time="2026-07-06T09:00:00+00:00",
|
|
end_time="2026-07-06T09:30:00+00:00",
|
|
rrule=rrule,
|
|
)
|
|
assert result["success"] is True
|
|
assert result["event"]["rrule"] == rrule
|
|
|
|
fetched = await service.get_event(result["event"]["id"])
|
|
assert fetched is not None
|
|
assert fetched.rrule == rrule
|
|
|
|
|
|
async def test_query_events_returns_list(tool: CalendarTool) -> None:
|
|
"""create 2 events, query, verify both returned."""
|
|
await tool.execute(
|
|
action="create_event",
|
|
user_id="user-1",
|
|
title="Event A",
|
|
start_time="2026-07-01T10:00:00+00:00",
|
|
end_time="2026-07-01T11:00:00+00:00",
|
|
)
|
|
await tool.execute(
|
|
action="create_event",
|
|
user_id="user-1",
|
|
title="Event B",
|
|
start_time="2026-07-02T14:00:00+00:00",
|
|
end_time="2026-07-02T15:00:00+00:00",
|
|
)
|
|
|
|
result = await tool.execute(
|
|
action="query_events",
|
|
user_id="user-1",
|
|
)
|
|
assert result["success"] is True
|
|
events = result["events"]
|
|
assert len(events) == 2
|
|
titles = {e["title"] for e in events}
|
|
assert titles == {"Event A", "Event B"}
|
|
|
|
|
|
async def test_update_event_action_modifies_fields(
|
|
tool: CalendarTool, service: CalendarService
|
|
) -> None:
|
|
"""create then update title; verify the field is modified."""
|
|
create_result = await tool.execute(
|
|
action="create_event",
|
|
user_id="user-1",
|
|
title="Original Title",
|
|
start_time="2026-07-01T10:00:00+00:00",
|
|
end_time="2026-07-01T11:00:00+00:00",
|
|
)
|
|
assert create_result["success"] is True
|
|
event_id = create_result["event"]["id"]
|
|
|
|
update_result = await tool.execute(
|
|
action="update_event",
|
|
user_id="user-1",
|
|
event_id=event_id,
|
|
title="Updated Title",
|
|
)
|
|
assert update_result["success"] is True
|
|
|
|
fetched = await service.get_event(event_id)
|
|
assert fetched is not None
|
|
assert fetched.title == "Updated Title"
|
|
|
|
|
|
async def test_delete_event_action_removes_record(
|
|
tool: CalendarTool, service: CalendarService
|
|
) -> None:
|
|
"""create then delete; verify the record is gone."""
|
|
create_result = await tool.execute(
|
|
action="create_event",
|
|
user_id="user-1",
|
|
title="To Be Deleted",
|
|
start_time="2026-07-01T10:00:00+00:00",
|
|
end_time="2026-07-01T11:00:00+00:00",
|
|
)
|
|
assert create_result["success"] is True
|
|
event_id = create_result["event"]["id"]
|
|
|
|
delete_result = await tool.execute(
|
|
action="delete_event",
|
|
user_id="user-1",
|
|
event_id=event_id,
|
|
)
|
|
assert delete_result["success"] is True
|
|
|
|
fetched = await service.get_event(event_id)
|
|
assert fetched is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# error paths
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def test_invalid_action_returns_error(tool: CalendarTool) -> None:
|
|
"""Unknown action returns success=False with error message."""
|
|
result = await tool.execute(
|
|
action="frobnicate",
|
|
user_id="user-1",
|
|
)
|
|
assert result["success"] is False
|
|
assert "Unknown action" in result["error"]
|
|
|
|
|
|
async def test_missing_required_field_returns_error(tool: CalendarTool) -> None:
|
|
"""create_event without title returns success=False."""
|
|
result = await tool.execute(
|
|
action="create_event",
|
|
user_id="user-1",
|
|
start_time="2026-07-01T10:00:00+00:00",
|
|
end_time="2026-07-01T11:00:00+00:00",
|
|
)
|
|
assert result["success"] is False
|
|
assert "Missing required field" in result["error"]
|
|
assert "title" in result["error"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# conversation_id
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def test_created_event_has_conversation_id(
|
|
tool: CalendarTool, service: CalendarService
|
|
) -> None:
|
|
"""conversation_id is set from context when provided."""
|
|
result = await tool.execute(
|
|
action="create_event",
|
|
user_id="user-1",
|
|
title="Chat-Initiated Event",
|
|
start_time="2026-07-01T10:00:00+00:00",
|
|
end_time="2026-07-01T11:00:00+00:00",
|
|
conversation_id="conv-abc-123",
|
|
)
|
|
assert result["success"] is True
|
|
assert result["event"]["conversation_id"] == "conv-abc-123"
|
|
|
|
fetched = await service.get_event(result["event"]["id"])
|
|
assert fetched is not None
|
|
assert fetched.conversation_id == "conv-abc-123"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# event_type_name resolution
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def test_create_event_with_event_type_name(
|
|
tool: CalendarTool, service: CalendarService
|
|
) -> None:
|
|
"""create event with event_type_name='Meeting' creates the type and links it."""
|
|
result = await tool.execute(
|
|
action="create_event",
|
|
user_id="user-1",
|
|
title="Strategy Sync",
|
|
start_time="2026-07-01T10:00:00+00:00",
|
|
end_time="2026-07-01T11:00:00+00:00",
|
|
event_type_name="Meeting",
|
|
)
|
|
assert result["success"] is True
|
|
event_type_id = result["event"]["event_type_id"]
|
|
assert event_type_id is not None
|
|
|
|
# Verify the event type was created
|
|
types = await service.list_event_types("user-1")
|
|
meeting_types = [t for t in types if t.name == "Meeting"]
|
|
assert len(meeting_types) == 1
|
|
assert meeting_types[0].id == event_type_id
|
|
|
|
# Verify the event references the type
|
|
fetched = await service.get_event(result["event"]["id"])
|
|
assert fetched is not None
|
|
assert fetched.event_type_id == event_type_id
|
|
|
|
|
|
async def test_create_event_reuses_existing_event_type(
|
|
tool: CalendarTool, service: CalendarService
|
|
) -> None:
|
|
"""If event_type_name matches an existing type, it is reused (not duplicated)."""
|
|
# Pre-create the type
|
|
existing = await service.create_event_type("user-1", "Meeting")
|
|
|
|
result = await tool.execute(
|
|
action="create_event",
|
|
user_id="user-1",
|
|
title="Second Meeting",
|
|
start_time="2026-07-02T10:00:00+00:00",
|
|
end_time="2026-07-02T11:00:00+00:00",
|
|
event_type_name="Meeting",
|
|
)
|
|
assert result["success"] is True
|
|
assert result["event"]["event_type_id"] == existing.id
|
|
|
|
# No duplicate type created
|
|
types = await service.list_event_types("user-1")
|
|
meeting_types = [t for t in types if t.name == "Meeting"]
|
|
assert len(meeting_types) == 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# tag_names resolution
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def test_create_event_with_tag_names(
|
|
tool: CalendarTool, service: CalendarService, tmp_path: Path
|
|
) -> None:
|
|
"""create event with tag_names=['urgent', 'work'] creates tags and links them."""
|
|
result = await tool.execute(
|
|
action="create_event",
|
|
user_id="user-1",
|
|
title="Urgent Work Task",
|
|
start_time="2026-07-01T10:00:00+00:00",
|
|
end_time="2026-07-01T11:00:00+00:00",
|
|
tag_names=["urgent", "work"],
|
|
)
|
|
assert result["success"] is True
|
|
event_id = result["event"]["id"]
|
|
|
|
# Verify tags were created
|
|
tags = await service.list_tags("user-1")
|
|
tag_names = {t.name for t in tags}
|
|
assert "urgent" in tag_names
|
|
assert "work" in tag_names
|
|
|
|
# Verify tags are linked to the event
|
|
linked_tags = await get_event_tags(event_id, service.db_path)
|
|
linked_names = {t.name for t in linked_tags}
|
|
assert linked_names == {"urgent", "work"}
|
|
|
|
|
|
async def test_create_event_reuses_existing_tags(
|
|
tool: CalendarTool, service: CalendarService
|
|
) -> None:
|
|
"""If a tag name matches an existing tag, it is reused (not duplicated)."""
|
|
# Pre-create a tag
|
|
existing = await service.create_tag("user-1", "urgent")
|
|
|
|
result = await tool.execute(
|
|
action="create_event",
|
|
user_id="user-1",
|
|
title="Tagged Event",
|
|
start_time="2026-07-01T10:00:00+00:00",
|
|
end_time="2026-07-01T11:00:00+00:00",
|
|
tag_names=["urgent", "new-tag"],
|
|
)
|
|
assert result["success"] is True
|
|
|
|
# 'urgent' should not be duplicated, 'new-tag' should be created
|
|
tags = await service.list_tags("user-1")
|
|
urgent_tags = [t for t in tags if t.name == "urgent"]
|
|
assert len(urgent_tags) == 1
|
|
assert urgent_tags[0].id == existing.id
|
|
new_tags = [t for t in tags if t.name == "new-tag"]
|
|
assert len(new_tags) == 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# tool registration / schema
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_tool_name_and_schema(tool: CalendarTool) -> None:
|
|
"""Tool has correct name and input_schema."""
|
|
assert tool.name == "calendar"
|
|
schema = tool.input_schema
|
|
assert schema["type"] == "object"
|
|
assert "action" in schema["properties"]
|
|
assert "user_id" in schema["properties"]
|
|
assert schema["properties"]["action"]["enum"] == [
|
|
"create_event",
|
|
"query_events",
|
|
"update_event",
|
|
"delete_event",
|
|
]
|
|
assert "action" in schema["required"]
|
|
assert "user_id" in schema["required"]
|