273 lines
8.5 KiB
Python
273 lines
8.5 KiB
Python
"""Tests for ICSProvider — iCalendar import/export (U8)."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
from icalendar import Calendar
|
||
|
||
from agentkit.calendar.db import init_calendar_db, list_events as db_list_events
|
||
from agentkit.calendar.service import CalendarService
|
||
from agentkit.calendar.sync.ics_provider import ICSProvider
|
||
from agentkit.server.auth.models import init_auth_db
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Fixtures
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.fixture
|
||
def calendar_db_path(tmp_path: Path) -> Path:
|
||
path = tmp_path / "test_calendar.db"
|
||
asyncio.run(init_calendar_db(path))
|
||
return path
|
||
|
||
|
||
@pytest.fixture
|
||
def auth_db_path(tmp_path: Path) -> Path:
|
||
path = tmp_path / "test_auth.db"
|
||
asyncio.run(init_auth_db(path))
|
||
return path
|
||
|
||
|
||
@pytest.fixture
|
||
def service(calendar_db_path: Path, auth_db_path: Path) -> CalendarService:
|
||
return CalendarService(db_path=calendar_db_path, auth_db_path=auth_db_path)
|
||
|
||
|
||
@pytest.fixture
|
||
def provider(service: CalendarService) -> ICSProvider:
|
||
return ICSProvider(service)
|
||
|
||
|
||
USER_ID = "user-1"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# ICS sample strings
|
||
# ---------------------------------------------------------------------------
|
||
|
||
SIMPLE_ICS = (
|
||
b"BEGIN:VCALENDAR\n"
|
||
b"VERSION:2.0\n"
|
||
b"PRODID:-//Test//Test//EN\n"
|
||
b"BEGIN:VEVENT\n"
|
||
b"UID:simple-uid@test\n"
|
||
b"SUMMARY:Test Event\n"
|
||
b"DTSTART:20260701T100000Z\n"
|
||
b"DTEND:20260701T110000Z\n"
|
||
b"DESCRIPTION:Test description\n"
|
||
b"LOCATION:Room A\n"
|
||
b"END:VEVENT\n"
|
||
b"END:VCALENDAR\n"
|
||
)
|
||
|
||
RECURRING_ICS = (
|
||
b"BEGIN:VCALENDAR\n"
|
||
b"VERSION:2.0\n"
|
||
b"PRODID:-//Test//Test//EN\n"
|
||
b"BEGIN:VEVENT\n"
|
||
b"UID:recur-uid@test\n"
|
||
b"SUMMARY:Weekly Meeting\n"
|
||
b"DTSTART:20260706T100000Z\n"
|
||
b"DTEND:20260706T110000Z\n"
|
||
b"RRULE:FREQ=WEEKLY;BYDAY=MO;COUNT=4\n"
|
||
b"END:VEVENT\n"
|
||
b"END:VCALENDAR\n"
|
||
)
|
||
|
||
ALL_DAY_ICS = (
|
||
b"BEGIN:VCALENDAR\n"
|
||
b"VERSION:2.0\n"
|
||
b"PRODID:-//Test//Test//EN\n"
|
||
b"BEGIN:VEVENT\n"
|
||
b"UID:allday-uid@test\n"
|
||
b"SUMMARY:All Day Event\n"
|
||
b"DTSTART;VALUE=DATE:20260701\n"
|
||
b"DTEND;VALUE=DATE:20260702\n"
|
||
b"END:VEVENT\n"
|
||
b"END:VCALENDAR\n"
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Import tests
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
async def test_import_simple_ics_creates_event(
|
||
provider: ICSProvider, calendar_db_path: Path
|
||
) -> None:
|
||
"""Single VEVENT ICS → event created with correct fields."""
|
||
result = await provider.import_ics(SIMPLE_ICS, USER_ID)
|
||
|
||
assert result["imported"] == 1
|
||
assert result["skipped"] == 0
|
||
assert result["errors"] == []
|
||
|
||
events = await db_list_events(USER_ID, db_path=calendar_db_path)
|
||
assert len(events) == 1
|
||
event = events[0]
|
||
assert event.title == "Test Event"
|
||
assert event.description == "Test description"
|
||
assert event.location == "Room A"
|
||
assert event.start_time == "2026-07-01T10:00:00+00:00"
|
||
assert event.end_time == "2026-07-01T11:00:00+00:00"
|
||
assert event.is_all_day is False
|
||
assert event.external_id == "simple-uid@test"
|
||
assert event.external_provider == "ics"
|
||
|
||
|
||
async def test_import_recurring_ics_preserves_rrule(
|
||
provider: ICSProvider, calendar_db_path: Path
|
||
) -> None:
|
||
"""VEVENT with RRULE → rrule field set on the created event."""
|
||
result = await provider.import_ics(RECURRING_ICS, USER_ID)
|
||
|
||
assert result["imported"] == 1
|
||
assert result["errors"] == []
|
||
|
||
events = await db_list_events(USER_ID, db_path=calendar_db_path)
|
||
assert len(events) == 1
|
||
event = events[0]
|
||
assert event.rrule is not None
|
||
assert "FREQ=WEEKLY" in event.rrule
|
||
assert "BYDAY=MO" in event.rrule
|
||
assert "COUNT=4" in event.rrule
|
||
|
||
|
||
async def test_import_all_day_event(provider: ICSProvider, calendar_db_path: Path) -> None:
|
||
"""DTSTART is date (not datetime) → is_all_day=True, ISO date stored."""
|
||
result = await provider.import_ics(ALL_DAY_ICS, USER_ID)
|
||
|
||
assert result["imported"] == 1
|
||
assert result["errors"] == []
|
||
|
||
events = await db_list_events(USER_ID, db_path=calendar_db_path)
|
||
assert len(events) == 1
|
||
event = events[0]
|
||
assert event.is_all_day is True
|
||
assert event.start_time == "2026-07-01"
|
||
assert event.end_time == "2026-07-02"
|
||
|
||
|
||
async def test_import_skips_duplicate_uid(provider: ICSProvider, calendar_db_path: Path) -> None:
|
||
"""Importing the same ICS twice → second import skips the duplicate UID."""
|
||
first = await provider.import_ics(SIMPLE_ICS, USER_ID)
|
||
assert first["imported"] == 1
|
||
assert first["skipped"] == 0
|
||
|
||
second = await provider.import_ics(SIMPLE_ICS, USER_ID)
|
||
assert second["imported"] == 0
|
||
assert second["skipped"] == 1
|
||
|
||
events = await db_list_events(USER_ID, db_path=calendar_db_path)
|
||
assert len(events) == 1 # Still only one event
|
||
|
||
|
||
async def test_import_malformed_ics_raises_error(provider: ICSProvider) -> None:
|
||
"""Invalid ICS bytes → ValueError raised (graceful, not a crash)."""
|
||
with pytest.raises(ValueError, match="Failed to parse ICS"):
|
||
await provider.import_ics(b"this is definitely not valid ics at all", USER_ID)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Export tests
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
async def test_export_produces_valid_ics(provider: ICSProvider, service: CalendarService) -> None:
|
||
"""Create event, export, parse result with icalendar → roundtrip."""
|
||
await service.create_event(
|
||
user_id=USER_ID,
|
||
title="Export Test",
|
||
start_time="2026-07-01T10:00:00+00:00",
|
||
end_time="2026-07-01T11:00:00+00:00",
|
||
description="Desc",
|
||
location="Room B",
|
||
)
|
||
|
||
events = await service.list_events(USER_ID)
|
||
ics_bytes = provider.export_ics(events)
|
||
|
||
# Parse the exported ICS back
|
||
cal = Calendar.from_ical(ics_bytes)
|
||
vevents = list(cal.walk("VEVENT"))
|
||
assert len(vevents) == 1
|
||
vevent = vevents[0]
|
||
assert str(vevent.get("SUMMARY")) == "Export Test"
|
||
assert str(vevent.get("DESCRIPTION")) == "Desc"
|
||
assert str(vevent.get("LOCATION")) == "Room B"
|
||
# DTSTART should be a datetime (not all-day)
|
||
dtstart = vevent.get("DTSTART").dt
|
||
assert dtstart.year == 2026
|
||
assert dtstart.month == 7
|
||
assert dtstart.day == 1
|
||
|
||
|
||
async def test_export_includes_recurrence(provider: ICSProvider, service: CalendarService) -> None:
|
||
"""Event with rrule → exported ICS contains RRULE line."""
|
||
await service.create_event(
|
||
user_id=USER_ID,
|
||
title="Recurring Export",
|
||
start_time="2026-07-06T10:00:00+00:00",
|
||
end_time="2026-07-06T11:00:00+00:00",
|
||
rrule="FREQ=WEEKLY;BYDAY=MO;COUNT=4",
|
||
)
|
||
|
||
events = await service.list_events(USER_ID)
|
||
ics_bytes = provider.export_ics(events)
|
||
|
||
cal = Calendar.from_ical(ics_bytes)
|
||
vevents = list(cal.walk("VEVENT"))
|
||
assert len(vevents) == 1
|
||
vevent = vevents[0]
|
||
rrule = vevent.get("RRULE")
|
||
assert rrule is not None
|
||
rrule_str = rrule.to_ical().decode("utf-8")
|
||
assert "FREQ=WEEKLY" in rrule_str
|
||
assert "BYDAY=MO" in rrule_str
|
||
assert "COUNT=4" in rrule_str
|
||
|
||
|
||
async def test_export_date_range_filter(provider: ICSProvider, service: CalendarService) -> None:
|
||
"""3 events, export range covering 2 → only 2 in ICS output."""
|
||
# Event 1: July 1
|
||
await service.create_event(
|
||
user_id=USER_ID,
|
||
title="Event 1",
|
||
start_time="2026-07-01T10:00:00+00:00",
|
||
end_time="2026-07-01T11:00:00+00:00",
|
||
)
|
||
# Event 2: July 5
|
||
await service.create_event(
|
||
user_id=USER_ID,
|
||
title="Event 2",
|
||
start_time="2026-07-05T10:00:00+00:00",
|
||
end_time="2026-07-05T11:00:00+00:00",
|
||
)
|
||
# Event 3: July 20
|
||
await service.create_event(
|
||
user_id=USER_ID,
|
||
title="Event 3",
|
||
start_time="2026-07-20T10:00:00+00:00",
|
||
end_time="2026-07-20T11:00:00+00:00",
|
||
)
|
||
|
||
# Export range: July 1 – July 10 (covers Event 1 and Event 2)
|
||
events = await service.list_events(
|
||
USER_ID,
|
||
start="2026-07-01T00:00:00+00:00",
|
||
end="2026-07-10T00:00:00+00:00",
|
||
)
|
||
ics_bytes = provider.export_ics(events)
|
||
|
||
cal = Calendar.from_ical(ics_bytes)
|
||
vevents = list(cal.walk("VEVENT"))
|
||
assert len(vevents) == 2
|
||
summaries = {str(v.get("SUMMARY")) for v in vevents}
|
||
assert summaries == {"Event 1", "Event 2"}
|