142 lines
5.6 KiB
Python
142 lines
5.6 KiB
Python
import pytest
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
from datetime import datetime, timezone
|
|
|
|
|
|
def _make_mock_session():
|
|
mock_session = AsyncMock()
|
|
mock_session.__aenter__ = AsyncMock(return_value=mock_session)
|
|
mock_session.__aexit__ = AsyncMock(return_value=False)
|
|
mock_session.add = MagicMock()
|
|
mock_session.commit = AsyncMock()
|
|
mock_session.refresh = AsyncMock()
|
|
return mock_session
|
|
|
|
|
|
def _make_mock_event(id=1, user_id=1, uid="uid-abc", title="Meeting",
|
|
caldav_uid="", color=""):
|
|
e = MagicMock()
|
|
e.id = id
|
|
e.user_id = user_id
|
|
e.uid = uid
|
|
e.title = title
|
|
e.caldav_uid = caldav_uid
|
|
e.color = color
|
|
e.start_dt = datetime(2026, 3, 25, 10, 0, tzinfo=timezone.utc)
|
|
e.end_dt = datetime(2026, 3, 25, 11, 0, tzinfo=timezone.utc)
|
|
e.all_day = False
|
|
e.description = ""
|
|
e.location = ""
|
|
e.recurrence = None
|
|
e.project_id = None
|
|
e.to_dict.return_value = {
|
|
"id": id, "uid": uid, "title": title,
|
|
"caldav_uid": caldav_uid, "color": color,
|
|
"start_dt": datetime(2026, 3, 25, 10, 0, tzinfo=timezone.utc).isoformat(),
|
|
"end_dt": datetime(2026, 3, 25, 11, 0, tzinfo=timezone.utc).isoformat(),
|
|
}
|
|
return e
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_event_stores_to_db():
|
|
mock_session = _make_mock_session()
|
|
with patch("fabledassistant.services.events.async_session") as mock_cls, \
|
|
patch("fabledassistant.services.events.asyncio.create_task") as mock_task:
|
|
mock_cls.return_value = mock_session
|
|
from fabledassistant.services.events import create_event
|
|
result = await create_event(
|
|
user_id=1,
|
|
title="Dentist",
|
|
start_dt=datetime(2026, 3, 25, 10, 0, tzinfo=timezone.utc),
|
|
)
|
|
assert mock_session.add.called
|
|
assert mock_session.commit.called
|
|
# CalDAV push background task should be scheduled
|
|
assert mock_task.called
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_find_events_by_query_returns_ilike_results():
|
|
mock_event = _make_mock_event(title="Team Meeting")
|
|
mock_session = _make_mock_session()
|
|
mock_result = MagicMock()
|
|
mock_result.scalars.return_value.all.return_value = [mock_event]
|
|
mock_session.execute = AsyncMock(return_value=mock_result)
|
|
|
|
with patch("fabledassistant.services.events.async_session") as mock_cls:
|
|
mock_cls.return_value = mock_session
|
|
from fabledassistant.services.events import find_events_by_query
|
|
results = await find_events_by_query(user_id=1, query="meeting")
|
|
assert len(results) == 1
|
|
assert results[0].title == "Team Meeting"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_events_returns_events_in_range():
|
|
mock_event = _make_mock_event()
|
|
mock_session = _make_mock_session()
|
|
mock_result = MagicMock()
|
|
mock_result.scalars.return_value.all.return_value = [mock_event]
|
|
mock_session.execute = AsyncMock(return_value=mock_result)
|
|
|
|
with patch("fabledassistant.services.events.async_session") as mock_cls:
|
|
mock_cls.return_value = mock_session
|
|
from fabledassistant.services.events import list_events
|
|
results = await list_events(
|
|
user_id=1,
|
|
date_from=datetime(2026, 3, 1, tzinfo=timezone.utc),
|
|
date_to=datetime(2026, 3, 31, tzinfo=timezone.utc),
|
|
)
|
|
assert len(results) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_event_fires_caldav_push_when_uid_set():
|
|
mock_event = _make_mock_event(caldav_uid="sync-uid")
|
|
mock_session = _make_mock_session()
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none.return_value = mock_event
|
|
mock_session.execute = AsyncMock(return_value=mock_result)
|
|
|
|
with patch("fabledassistant.services.events.async_session") as mock_cls, \
|
|
patch("fabledassistant.services.events.asyncio.create_task") as mock_task:
|
|
mock_cls.return_value = mock_session
|
|
from fabledassistant.services.events import delete_event
|
|
await delete_event(user_id=1, event_id=1)
|
|
# Push task fired because caldav_uid is set
|
|
assert mock_task.called
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_event_fires_caldav_push():
|
|
mock_event = _make_mock_event(caldav_uid="sync-uid")
|
|
mock_session = _make_mock_session()
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none.return_value = mock_event
|
|
mock_session.execute = AsyncMock(return_value=mock_result)
|
|
|
|
with patch("fabledassistant.services.events.async_session") as mock_cls, \
|
|
patch("fabledassistant.services.events.asyncio.create_task") as mock_task:
|
|
mock_cls.return_value = mock_session
|
|
from fabledassistant.services.events import update_event
|
|
await update_event(user_id=1, event_id=1, title="Updated Title")
|
|
assert mock_task.called
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tools_calendar_always_available():
|
|
"""Calendar tools must appear in get_tools_for_user even without CalDAV."""
|
|
with patch("fabledassistant.services.tools._registry.is_caldav_configured", new_callable=AsyncMock) as mock_configured, \
|
|
patch("fabledassistant.services.settings.get_setting", new_callable=AsyncMock) as mock_setting:
|
|
mock_configured.return_value = False
|
|
mock_setting.return_value = "false"
|
|
from fabledassistant.services.tools import get_tools_for_user
|
|
tools = await get_tools_for_user(user_id=1)
|
|
tool_names = {t["function"]["name"] for t in tools}
|
|
assert "create_event" in tool_names
|
|
assert "list_events" in tool_names
|
|
assert "search_events" in tool_names
|
|
assert "update_event" in tool_names
|
|
assert "delete_event" in tool_names
|