399 lines
13 KiB
Python
399 lines
13 KiB
Python
"""Tests for CalDAVSyncProvider — bidirectional Apple Calendar sync (U6).
|
|
|
|
The ``caldav`` library is not installed in the test environment, so we inject
|
|
a mock module into ``sys.modules`` before importing the provider. All CalDAV
|
|
interactions are mocked via the ``client_factory`` injection point.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
# caldav is not installed in the test env — inject a mock module so that
|
|
# `import caldav` in caldav_provider.py succeeds at import time.
|
|
if "caldav" not in sys.modules: # pragma: no cover
|
|
sys.modules["caldav"] = MagicMock()
|
|
|
|
from agentkit.calendar.db import (
|
|
get_event_by_external_id,
|
|
init_calendar_db,
|
|
insert_event,
|
|
list_events,
|
|
)
|
|
from agentkit.calendar.models import CalendarEvent, ExternalCalendarConfig
|
|
from agentkit.calendar.sync.caldav_provider import CalDAVSyncProvider
|
|
|
|
USER_ID = "user-1"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ICS + Mock helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def make_ics(
|
|
uid: str,
|
|
summary: str,
|
|
start: str = "20260701T100000Z",
|
|
end: str = "20260701T110000Z",
|
|
description: str = "",
|
|
location: str = "",
|
|
rrule: str | None = None,
|
|
last_modified: str = "20260601T000000Z",
|
|
) -> bytes:
|
|
"""Build a minimal valid ICS bytes payload for a single VEVENT."""
|
|
lines = [
|
|
b"BEGIN:VCALENDAR",
|
|
b"VERSION:2.0",
|
|
b"PRODID:-//Test//Test//EN",
|
|
b"BEGIN:VEVENT",
|
|
f"UID:{uid}".encode(),
|
|
f"SUMMARY:{summary}".encode(),
|
|
f"DTSTART:{start}".encode(),
|
|
f"DTEND:{end}".encode(),
|
|
]
|
|
if description:
|
|
lines.append(f"DESCRIPTION:{description}".encode())
|
|
if location:
|
|
lines.append(f"LOCATION:{location}".encode())
|
|
if rrule:
|
|
lines.append(f"RRULE:{rrule}".encode())
|
|
if last_modified:
|
|
lines.append(f"LAST-MODIFIED:{last_modified}".encode())
|
|
lines.extend([b"END:VEVENT", b"END:VCALENDAR"])
|
|
return b"\r\n".join(lines)
|
|
|
|
|
|
class MockCaldavEvent:
|
|
"""Mock caldav.Event — exposes .data returning ICS bytes."""
|
|
|
|
def __init__(self, ics_data: bytes) -> None:
|
|
self.data = ics_data
|
|
|
|
|
|
def make_mock_client(
|
|
remote_events: list[bytes],
|
|
saved_uid: str | None = None,
|
|
raise_on_connect: Exception | None = None,
|
|
) -> tuple[Any, Any]:
|
|
"""Create a mock CalDAV client and its mock calendar.
|
|
|
|
Returns ``(client, mock_calendar)``. ``mock_calendar`` is None when
|
|
``raise_on_connect`` is set (connection fails before calendar is reached).
|
|
"""
|
|
client = MagicMock()
|
|
if raise_on_connect is not None:
|
|
client.principal.side_effect = raise_on_connect
|
|
return client, None
|
|
|
|
principal = MagicMock()
|
|
calendar = MagicMock()
|
|
calendar.date_search.return_value = [MockCaldavEvent(d) for d in remote_events]
|
|
saved_ics = make_ics(saved_uid or "saved-uid", "Saved Event")
|
|
calendar.save_event.return_value = MockCaldavEvent(saved_ics)
|
|
principal.calendars.return_value = [calendar]
|
|
client.principal.return_value = principal
|
|
return client, calendar
|
|
|
|
|
|
def client_factory_from(client: Any) -> Any:
|
|
"""Wrap a mock client in a client_factory callable."""
|
|
|
|
def factory(config: ExternalCalendarConfig) -> Any:
|
|
return client
|
|
|
|
return factory
|
|
|
|
|
|
def make_config(
|
|
config_id: str = "config-1",
|
|
user_id: str = USER_ID,
|
|
provider: str = "caldav",
|
|
last_sync: str | None = None,
|
|
) -> ExternalCalendarConfig:
|
|
return ExternalCalendarConfig(
|
|
id=config_id,
|
|
user_id=user_id,
|
|
provider=provider,
|
|
credentials=json.dumps(
|
|
{"url": "https://caldav.example.com", "username": "user", "password": "pass"}
|
|
),
|
|
last_sync=last_sync,
|
|
)
|
|
|
|
|
|
def make_local_event(
|
|
event_id: str = "evt-1",
|
|
title: str = "Local Event",
|
|
external_id: str | None = None,
|
|
last_modified: str = "2026-01-01T00:00:00+00:00",
|
|
rrule: str | None = None,
|
|
) -> CalendarEvent:
|
|
return CalendarEvent(
|
|
id=event_id,
|
|
user_id=USER_ID,
|
|
title=title,
|
|
start_time="2026-07-01T10:00:00+00:00",
|
|
end_time="2026-07-01T11:00:00+00:00",
|
|
external_id=external_id,
|
|
external_provider="caldav" if external_id else None,
|
|
last_modified=last_modified,
|
|
created_at=last_modified,
|
|
rrule=rrule,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Pull tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def test_caldav_provider_pull_creates_local_events(calendar_db_path: Path) -> None:
|
|
"""Mock returns 2 events → 2 local events created with external_id set."""
|
|
ics1 = make_ics("uid-1", "Remote Event 1")
|
|
ics2 = make_ics("uid-2", "Remote Event 2")
|
|
client, _ = make_mock_client([ics1, ics2])
|
|
provider = CalDAVSyncProvider(
|
|
db_path=calendar_db_path, client_factory=client_factory_from(client)
|
|
)
|
|
config = make_config()
|
|
|
|
pulled = await provider.pull_changes(config)
|
|
|
|
assert len(pulled) == 2
|
|
events = await list_events(USER_ID, db_path=calendar_db_path)
|
|
assert len(events) == 2
|
|
titles = {e.title for e in events}
|
|
assert titles == {"Remote Event 1", "Remote Event 2"}
|
|
for event in events:
|
|
assert event.external_id is not None
|
|
assert event.external_provider == "caldav"
|
|
|
|
|
|
async def test_caldav_provider_pull_updates_existing_event(calendar_db_path: Path) -> None:
|
|
"""Local event exists (matched by external_id), remote is newer → local updated."""
|
|
local = make_local_event(
|
|
event_id="evt-1",
|
|
title="Old Title",
|
|
external_id="uid-1",
|
|
last_modified="2026-01-01T00:00:00+00:00",
|
|
)
|
|
await insert_event(local, calendar_db_path)
|
|
|
|
ics_remote = make_ics("uid-1", "Updated Title", last_modified="20260601T120000Z")
|
|
client, _ = make_mock_client([ics_remote])
|
|
provider = CalDAVSyncProvider(
|
|
db_path=calendar_db_path, client_factory=client_factory_from(client)
|
|
)
|
|
config = make_config()
|
|
|
|
pulled = await provider.pull_changes(config)
|
|
|
|
assert len(pulled) == 1
|
|
updated = await get_event_by_external_id("uid-1", "caldav", USER_ID, calendar_db_path)
|
|
assert updated is not None
|
|
assert updated.title == "Updated Title"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Push tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def test_caldav_provider_push_creates_remote_event(calendar_db_path: Path) -> None:
|
|
"""Local event with no external_id → push creates remote, external_id stored."""
|
|
local = make_local_event(event_id="evt-1", title="New Local Event", external_id=None)
|
|
await insert_event(local, calendar_db_path)
|
|
|
|
client, _ = make_mock_client([], saved_uid="remote-uid-1")
|
|
provider = CalDAVSyncProvider(
|
|
db_path=calendar_db_path, client_factory=client_factory_from(client)
|
|
)
|
|
config = make_config()
|
|
|
|
events = await list_events(USER_ID, db_path=calendar_db_path)
|
|
pushed = await provider.push_changes(config, events)
|
|
|
|
assert len(pushed) == 1
|
|
assert pushed[0].external_id == "remote-uid-1"
|
|
assert pushed[0].external_provider == "caldav"
|
|
db_event = await get_event_by_external_id("remote-uid-1", "caldav", USER_ID, calendar_db_path)
|
|
assert db_event is not None
|
|
|
|
|
|
async def test_caldav_provider_push_updates_remote_event(calendar_db_path: Path) -> None:
|
|
"""Local event modified, has external_id → push updates remote."""
|
|
local = make_local_event(
|
|
event_id="evt-1",
|
|
title="Updated Local Event",
|
|
external_id="existing-uid",
|
|
last_modified="2026-06-01T00:00:00+00:00",
|
|
)
|
|
await insert_event(local, calendar_db_path)
|
|
|
|
client, mock_calendar = make_mock_client([], saved_uid="existing-uid")
|
|
provider = CalDAVSyncProvider(
|
|
db_path=calendar_db_path, client_factory=client_factory_from(client)
|
|
)
|
|
config = make_config()
|
|
|
|
events = await list_events(USER_ID, db_path=calendar_db_path)
|
|
pushed = await provider.push_changes(config, events)
|
|
|
|
assert len(pushed) == 1
|
|
assert pushed[0].external_id == "existing-uid"
|
|
# Verify save_event was called with ICS containing the existing UID
|
|
saved_ics = mock_calendar.save_event.call_args[0][0]
|
|
assert b"UID:existing-uid" in saved_ics
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Conflict tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def test_caldav_conflict_last_write_wins(calendar_db_path: Path) -> None:
|
|
"""Both sides modified, local is newer → local wins, local not updated."""
|
|
local = make_local_event(
|
|
event_id="evt-1",
|
|
title="Local Updated",
|
|
external_id="uid-1",
|
|
last_modified="2026-06-15T00:00:00+00:00",
|
|
)
|
|
await insert_event(local, calendar_db_path)
|
|
|
|
ics_remote = make_ics("uid-1", "Remote Older", last_modified="20260601T000000Z")
|
|
client, _ = make_mock_client([ics_remote])
|
|
provider = CalDAVSyncProvider(
|
|
db_path=calendar_db_path, client_factory=client_factory_from(client)
|
|
)
|
|
config = make_config()
|
|
|
|
pulled = await provider.pull_changes(config)
|
|
|
|
# Local wins → no update, pulled is empty
|
|
assert len(pulled) == 0
|
|
db_event = await get_event_by_external_id("uid-1", "caldav", USER_ID, calendar_db_path)
|
|
assert db_event is not None
|
|
assert db_event.title == "Local Updated"
|
|
|
|
|
|
async def test_caldav_conflict_sends_ws_notification(calendar_db_path: Path) -> None:
|
|
"""Conflict detected → WS callback called with calendar_sync_conflict type (G4)."""
|
|
notifications: list[tuple[str, dict[str, Any]]] = []
|
|
|
|
async def notify(event_type: str, payload: dict[str, Any]) -> None:
|
|
notifications.append((event_type, payload))
|
|
|
|
local = make_local_event(
|
|
event_id="evt-1",
|
|
title="Local Updated",
|
|
external_id="uid-1",
|
|
last_modified="2026-06-15T00:00:00+00:00",
|
|
)
|
|
await insert_event(local, calendar_db_path)
|
|
|
|
ics_remote = make_ics("uid-1", "Remote Older", last_modified="20260601T000000Z")
|
|
client, _ = make_mock_client([ics_remote])
|
|
provider = CalDAVSyncProvider(
|
|
db_path=calendar_db_path,
|
|
client_factory=client_factory_from(client),
|
|
notify_callback=notify,
|
|
)
|
|
config = make_config()
|
|
|
|
await provider.pull_changes(config)
|
|
|
|
assert len(notifications) == 1
|
|
event_type, payload = notifications[0]
|
|
assert event_type == "calendar_sync_conflict"
|
|
assert payload["event_id"] == "evt-1"
|
|
assert payload["winner"] == "local"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# RRULE roundtrip test
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def test_caldav_rrule_roundtrip(calendar_db_path: Path) -> None:
|
|
"""Event with RRULE synced → RRULE preserved in both pull and push."""
|
|
rrule = "FREQ=WEEKLY;BYDAY=MO;COUNT=4"
|
|
ics_remote = make_ics("uid-rrule", "Weekly Meeting", rrule=rrule)
|
|
client, _ = make_mock_client([ics_remote])
|
|
provider = CalDAVSyncProvider(
|
|
db_path=calendar_db_path, client_factory=client_factory_from(client)
|
|
)
|
|
config = make_config()
|
|
|
|
# Pull: verify RRULE preserved
|
|
pulled = await provider.pull_changes(config)
|
|
assert len(pulled) == 1
|
|
assert pulled[0].rrule is not None
|
|
assert "FREQ=WEEKLY" in pulled[0].rrule
|
|
assert "BYDAY=MO" in pulled[0].rrule
|
|
assert "COUNT=4" in pulled[0].rrule
|
|
|
|
# Push: verify RRULE in ICS sent to remote
|
|
events = await list_events(USER_ID, db_path=calendar_db_path)
|
|
client2, mock_calendar2 = make_mock_client([], saved_uid="uid-rrule")
|
|
provider2 = CalDAVSyncProvider(
|
|
db_path=calendar_db_path, client_factory=client_factory_from(client2)
|
|
)
|
|
await provider2.push_changes(config, events)
|
|
|
|
saved_ics = mock_calendar2.save_event.call_args[0][0]
|
|
assert b"RRULE:FREQ=WEEKLY" in saved_ics
|
|
assert b"BYDAY=MO" in saved_ics
|
|
assert b"COUNT=4" in saved_ics
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# test_connection tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def test_caldav_test_connection_success(calendar_db_path: Path) -> None:
|
|
"""Mock returns calendars → test_connection() returns (True, "")."""
|
|
client, _ = make_mock_client([])
|
|
provider = CalDAVSyncProvider(
|
|
db_path=calendar_db_path, client_factory=client_factory_from(client)
|
|
)
|
|
config = make_config()
|
|
|
|
ok, msg = await provider.test_connection(config)
|
|
assert ok is True
|
|
assert msg == ""
|
|
|
|
|
|
async def test_caldav_test_connection_failure(calendar_db_path: Path) -> None:
|
|
"""Mock raises → test_connection() returns (False, error_msg)."""
|
|
client, _ = make_mock_client([], raise_on_connect=ConnectionError("Auth failed"))
|
|
provider = CalDAVSyncProvider(
|
|
db_path=calendar_db_path, client_factory=client_factory_from(client)
|
|
)
|
|
config = make_config()
|
|
|
|
ok, msg = await provider.test_connection(config)
|
|
assert ok is False
|
|
assert "Auth failed" in msg
|