govoplan-calendar/tests/test_caldav.py

from __future__ import annotations

import unittest
from datetime import datetime, timezone
from unittest.mock import patch

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from govoplan_access.backend.db import models as access_models  # noqa: F401 - populate users/accounts tables
from govoplan_calendar.backend.caldav import CalDAVClient, CalDAVError, CalDAVNotFound, CalDAVObject, CalDAVPreconditionFailed, CalDAVReportResult, CalDAVWriteResult, parse_multistatus
from govoplan_calendar.backend.db.models import CalendarEvent, CalendarSyncCredential
from govoplan_calendar.backend.schemas import CalendarCalDavSourceCreateRequest, CalendarCollectionCreateRequest, CalendarEventCreateRequest, CalendarEventUpdateRequest
from govoplan_calendar.backend.service import (
    CALDAV_INTERNAL_CREDENTIAL_PREFIX,
    CalendarError,
    caldav_client_for_source,
    create_calendar,
    create_caldav_source,
    create_event,
    delete_calendar,
    delete_event,
    list_caldav_sources,
    list_freebusy,
    sync_caldav_source,
    sync_due_caldav_sources,
    update_event,
)
from govoplan_core.db.base import Base
from govoplan_core.tenancy.scope import create_scope_tables
from govoplan_tenancy.backend.db.models import Tenant


class FakeCalDAVClient:
    def __init__(
        self,
        *,
        list_result: CalDAVReportResult | None = None,
        sync_result: CalDAVReportResult | None = None,
        objects: dict[str, str] | None = None,
        fetch_not_found: set[str] | None = None,
        put_etags: list[str | None] | None = None,
        fail_precondition: bool = False,
    ) -> None:
        self.list_result = list_result or CalDAVReportResult()
        self.sync_result = sync_result
        self.objects = objects or {}
        self.fetch_not_found = fetch_not_found or set()
        self.put_etags = put_etags or ['"etag-written"']
        self.fail_precondition = fail_precondition
        self.fetched: list[str] = []
        self.puts: list[dict[str, object]] = []
        self.deletes: list[dict[str, object]] = []

    def propfind_collection(self) -> CalDAVReportResult:
        return CalDAVReportResult(sync_token="prop-token", ctag="ctag-1")

    def list_objects(self) -> CalDAVReportResult:
        return self.list_result

    def sync_collection(self, sync_token: str) -> CalDAVReportResult:
        if self.sync_result is None:
            raise AssertionError(f"Unexpected sync_collection call with {sync_token}")
        return self.sync_result

    def fetch_object(self, href: str) -> str:
        self.fetched.append(href)
        if href in self.fetch_not_found:
            raise CalDAVNotFound(f"GET {href} returned HTTP 404")
        return self.objects[href]

    def put_object(self, href: str, ics: str, *, etag: str | None = None, create: bool = False, overwrite: bool = False) -> CalDAVWriteResult:
        if self.fail_precondition:
            raise CalDAVPreconditionFailed("changed")
        self.puts.append({"href": href, "ics": ics, "etag": etag, "create": create, "overwrite": overwrite})
        result_etag = self.put_etags.pop(0) if self.put_etags else None
        return CalDAVWriteResult(href=href, etag=result_etag, status=201 if create else 204)

    def delete_object(self, href: str, *, etag: str | None = None, overwrite: bool = False) -> CalDAVWriteResult:
        if self.fail_precondition:
            raise CalDAVPreconditionFailed("changed")
        self.deletes.append({"href": href, "etag": etag, "overwrite": overwrite})
        return CalDAVWriteResult(href=href, status=204)


class FakeNotificationProvider:
    def __init__(self) -> None:
        self.requests: list[object] = []

    def enqueue_notification(self, _session, request, **_kwargs) -> None:
        self.requests.append(request)


class CalDAVParsingTests(unittest.TestCase):
    def test_parse_multistatus_extracts_objects_sync_token_and_deletions(self) -> None:
        result = parse_multistatus(
            b"""<?xml version="1.0" encoding="utf-8"?>
<D:multistatus xmlns:D="DAV:" xmlns:C="urn:ietf:params:xml:ns:caldav">
  <D:sync-token>token-2</D:sync-token>
  <D:response>
    <D:href>/cal/event-1.ics</D:href>
    <D:propstat>
      <D:prop>
        <D:getetag>"etag-1"</D:getetag>
        <C:calendar-data>BEGIN:VCALENDAR
BEGIN:VEVENT
UID:event-1@example.test
DTSTART:20260708T090000Z
SUMMARY:One
END:VEVENT
END:VCALENDAR</C:calendar-data>
      </D:prop>
      <D:status>HTTP/1.1 200 OK</D:status>
    </D:propstat>
  </D:response>
  <D:response>
    <D:href>/cal/event-2.ics</D:href>
    <D:propstat>
      <D:prop><D:getetag/></D:prop>
      <D:status>HTTP/1.1 404 Not Found</D:status>
    </D:propstat>
  </D:response>
</D:multistatus>
"""
        )

        self.assertEqual(result.sync_token, "token-2")
        self.assertEqual(result.objects[0].href, "/cal/event-1.ics")
        self.assertEqual(result.objects[0].etag, '"etag-1"')
        self.assertIn("BEGIN:VEVENT", result.objects[0].calendar_data or "")
        self.assertTrue(result.objects[1].deleted)

    def test_discovery_resolves_nextcloud_dav_root_to_calendar_collections(self) -> None:
        requests: list[tuple[str, str, str | None]] = []

        def transport(method: str, url: str, headers: dict[str, str], body: bytes | None, timeout: int):
            requests.append((method, url, headers.get("Authorization")))
            if url == "https://cloud.example.test/remote.php/dav/":
                return 207, {}, b"""<?xml version="1.0" encoding="utf-8"?>
<D:multistatus xmlns:D="DAV:">
  <D:response>
    <D:href>/remote.php/dav/</D:href>
    <D:propstat>
      <D:prop>
        <D:current-user-principal><D:href>/remote.php/dav/principals/users/ada/</D:href></D:current-user-principal>
      </D:prop>
      <D:status>HTTP/1.1 200 OK</D:status>
    </D:propstat>
  </D:response>
</D:multistatus>"""
            if url == "https://cloud.example.test/remote.php/dav/principals/users/ada/":
                return 207, {}, b"""<?xml version="1.0" encoding="utf-8"?>
<D:multistatus xmlns:D="DAV:" xmlns:C="urn:ietf:params:xml:ns:caldav">
  <D:response>
    <D:href>/remote.php/dav/principals/users/ada/</D:href>
    <D:propstat>
      <D:prop>
        <C:calendar-home-set><D:href>/remote.php/dav/calendars/ada/</D:href></C:calendar-home-set>
      </D:prop>
      <D:status>HTTP/1.1 200 OK</D:status>
    </D:propstat>
  </D:response>
</D:multistatus>"""
            if url == "https://cloud.example.test/remote.php/dav/calendars/ada/":
                return 207, {}, b"""<?xml version="1.0" encoding="utf-8"?>
<D:multistatus xmlns:D="DAV:" xmlns:C="urn:ietf:params:xml:ns:caldav" xmlns:CS="http://calendarserver.org/ns/" xmlns:ICAL="http://apple.com/ns/ical/">
  <D:response>
    <D:href>/remote.php/dav/calendars/ada/</D:href>
    <D:propstat>
      <D:prop><D:resourcetype><D:collection/></D:resourcetype></D:prop>
      <D:status>HTTP/1.1 200 OK</D:status>
    </D:propstat>
  </D:response>
  <D:response>
    <D:href>/remote.php/dav/calendars/ada/personal/</D:href>
    <D:propstat>
      <D:prop>
        <D:displayname>Personal</D:displayname>
        <D:resourcetype><D:collection/><C:calendar/></D:resourcetype>
        <C:supported-calendar-component-set><C:comp name="VEVENT"/></C:supported-calendar-component-set>
        <ICAL:calendar-color>#44aa99ff</ICAL:calendar-color>
        <CS:getctag>ctag-1</CS:getctag>
      </D:prop>
      <D:status>HTTP/1.1 200 OK</D:status>
    </D:propstat>
  </D:response>
</D:multistatus>"""
            raise AssertionError(f"Unexpected {method} {url}")

        client = CalDAVClient(collection_url="https://cloud.example.test/remote.php/dav", username="ada", password="secret", transport=transport)
        calendars = client.discover_calendars()

        self.assertEqual(len(calendars), 1)
        self.assertEqual(calendars[0].collection_url, "https://cloud.example.test/remote.php/dav/calendars/ada/personal/")
        self.assertEqual(calendars[0].display_name, "Personal")
        self.assertEqual(calendars[0].color, "#44aa99ff")
        self.assertEqual(requests[0][2], "Basic YWRhOnNlY3JldA==")

    def test_discovery_accepts_host_path_url_without_scheme(self) -> None:
        requested_urls: list[str] = []

        def transport(method: str, url: str, headers: dict[str, str], body: bytes | None, timeout: int):
            requested_urls.append(url)
            return 207, {}, b"""<?xml version="1.0" encoding="utf-8"?>
<D:multistatus xmlns:D="DAV:" xmlns:C="urn:ietf:params:xml:ns:caldav">
  <D:response>
    <D:href>/remote.php/dav/calendars/ada/personal/</D:href>
    <D:propstat>
      <D:prop>
        <D:displayname>Personal</D:displayname>
        <D:resourcetype><D:collection/><C:calendar/></D:resourcetype>
        <C:supported-calendar-component-set><C:comp name="VEVENT"/></C:supported-calendar-component-set>
      </D:prop>
      <D:status>HTTP/1.1 200 OK</D:status>
    </D:propstat>
  </D:response>
</D:multistatus>"""

        client = CalDAVClient(collection_url="cloud.example.test/remote.php/dav", transport=transport)
        calendars = client.discover_calendars()

        self.assertEqual(requested_urls[0], "https://cloud.example.test/remote.php/dav/")
        self.assertEqual(calendars[0].collection_url, "https://cloud.example.test/remote.php/dav/calendars/ada/personal/")

    def test_discovery_reports_relative_url_as_caldav_error(self) -> None:
        with self.assertRaisesRegex(CalDAVError, "absolute HTTP"):
            CalDAVClient(collection_url="/remote.php/dav")


class CalDAVSyncTests(unittest.TestCase):
    def setUp(self) -> None:
        self.engine = create_engine("sqlite:///:memory:")
        create_scope_tables(self.engine)
        Base.metadata.create_all(bind=self.engine)
        self.Session = sessionmaker(bind=self.engine)

    def tearDown(self) -> None:
        Base.metadata.drop_all(bind=self.engine)

    def test_source_creation_encrypts_credential_and_resolves_client_secret(self) -> None:
        session = self.Session()
        session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant"))
        calendar = create_calendar(session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Remote"))

        source = create_caldav_source(
            session,
            tenant_id="tenant-1",
            user_id=None,
            payload=CalendarCalDavSourceCreateRequest(
                calendar_id=calendar.id,
                collection_url="https://dav.example.test/cal",
                auth_type="basic",
                username="ada",
                password="secret",
            ),
        )
        session.commit()

        credential = session.query(CalendarSyncCredential).one()
        client = caldav_client_for_source(session, source)

        self.assertTrue(source.credential_ref.startswith(CALDAV_INTERNAL_CREDENTIAL_PREFIX))
        self.assertNotEqual(credential.secret_encrypted, "secret")
        self.assertEqual(client.username, "ada")
        self.assertEqual(client.password, "secret")

    def test_delete_calendar_retires_caldav_source_and_credential(self) -> None:
        session = self.Session()
        session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant"))
        calendar = create_calendar(session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Remote"))
        source = create_caldav_source(
            session,
            tenant_id="tenant-1",
            user_id=None,
            payload=CalendarCalDavSourceCreateRequest(
                calendar_id=calendar.id,
                collection_url="https://dav.example.test/cal",
                auth_type="basic",
                username="ada",
                password="secret",
            ),
        )
        session.commit()

        delete_calendar(session, tenant_id="tenant-1", calendar_id=calendar.id)
        session.commit()

        credential = session.query(CalendarSyncCredential).one()
        self.assertIsNotNone(source.deleted_at)
        self.assertIsNotNone(credential.deleted_at)
        self.assertEqual(list_caldav_sources(session, tenant_id="tenant-1"), [])

    def test_create_source_retires_orphaned_source_for_deleted_calendar(self) -> None:
        session = self.Session()
        session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant"))
        old_calendar = create_calendar(session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Old"))
        old_source = create_caldav_source(
            session,
            tenant_id="tenant-1",
            user_id=None,
            payload=CalendarCalDavSourceCreateRequest(calendar_id=old_calendar.id, collection_url="https://dav.example.test/cal"),
        )
        old_calendar.deleted_at = datetime(2026, 7, 7, 12, 0, tzinfo=timezone.utc)
        new_calendar = create_calendar(session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="New"))
        session.commit()

        new_source = create_caldav_source(
            session,
            tenant_id="tenant-1",
            user_id=None,
            payload=CalendarCalDavSourceCreateRequest(calendar_id=new_calendar.id, collection_url="https://dav.example.test/cal"),
        )
        session.commit()

        self.assertIsNotNone(old_source.deleted_at)
        self.assertEqual(new_source.calendar_id, new_calendar.id)
        self.assertEqual([source.id for source in list_caldav_sources(session, tenant_id="tenant-1")], [new_source.id])

    def test_create_source_reports_active_duplicate_as_calendar_error(self) -> None:
        session = self.Session()
        session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant"))
        first = create_calendar(session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="First"))
        second = create_calendar(session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Second"))
        create_caldav_source(
            session,
            tenant_id="tenant-1",
            user_id=None,
            payload=CalendarCalDavSourceCreateRequest(calendar_id=first.id, collection_url="https://dav.example.test/cal"),
        )
        session.commit()

        with self.assertRaisesRegex(CalendarError, "already linked"):
            create_caldav_source(
                session,
                tenant_id="tenant-1",
                user_id=None,
                payload=CalendarCalDavSourceCreateRequest(calendar_id=second.id, collection_url="https://dav.example.test/cal"),
            )

    def test_due_sync_runs_due_sources_and_reschedules(self) -> None:
        session = self.Session()
        session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant"))
        calendar = create_calendar(session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Remote"))
        source = create_caldav_source(
            session,
            tenant_id="tenant-1",
            user_id=None,
            payload=CalendarCalDavSourceCreateRequest(calendar_id=calendar.id, collection_url="https://dav.example.test/cal"),
        )
        source.next_sync_at = datetime(2026, 7, 7, 8, 0, tzinfo=timezone.utc)
        session.commit()

        fake = FakeCalDAVClient(list_result=CalDAVReportResult(sync_token="token-1"))
        results = sync_due_caldav_sources(
            session,
            tenant_id="tenant-1",
            now=datetime(2026, 7, 7, 8, 1, tzinfo=timezone.utc),
            client_factory=lambda _source: fake,
        )
        session.commit()

        self.assertEqual(results[0].status, "ok")
        self.assertEqual(source.sync_token, "token-1")
        next_sync_at = source.next_sync_at if source.next_sync_at.tzinfo else source.next_sync_at.replace(tzinfo=timezone.utc)
        self.assertGreater(next_sync_at, datetime(2026, 7, 7, 8, 1, tzinfo=timezone.utc))

    def test_due_sync_emits_recovery_notification_after_previous_error(self) -> None:
        session = self.Session()
        session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant"))
        calendar = create_calendar(session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Remote"))
        source = create_caldav_source(
            session,
            tenant_id="tenant-1",
            user_id=None,
            payload=CalendarCalDavSourceCreateRequest(calendar_id=calendar.id, collection_url="https://dav.example.test/cal"),
        )
        source.last_status = "error"
        source.next_sync_at = datetime(2026, 7, 7, 8, 0, tzinfo=timezone.utc)
        session.commit()

        provider = FakeNotificationProvider()
        fake = FakeCalDAVClient(list_result=CalDAVReportResult(sync_token="token-1"))
        with patch("govoplan_calendar.backend.service.notification_dispatch_provider", return_value=provider):
            results = sync_due_caldav_sources(
                session,
                tenant_id="tenant-1",
                now=datetime(2026, 7, 7, 8, 1, tzinfo=timezone.utc),
                client_factory=lambda _source: fake,
            )

        self.assertEqual(results[0].status, "ok")
        self.assertEqual(len(provider.requests), 1)
        self.assertEqual(provider.requests[0].event_kind, "calendar.sync.ok")

    def test_create_and_update_event_puts_caldav_resource_with_etag(self) -> None:
        session = self.Session()
        session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant"))
        calendar = create_calendar(session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Remote"))
        create_caldav_source(session, tenant_id="tenant-1", user_id=None, payload=CalendarCalDavSourceCreateRequest(calendar_id=calendar.id, collection_url="https://dav.example.test/cal"))
        session.commit()

        fake = FakeCalDAVClient(put_etags=['"etag-1"', '"etag-2"'])
        with patch("govoplan_calendar.backend.service.caldav_client_for_source", return_value=fake):
            event = create_event(
                session,
                tenant_id="tenant-1",
                user_id=None,
                payload=CalendarEventCreateRequest(
                    calendar_id=calendar.id,
                    uid="event-1@example.test",
                    summary="Planning",
                    start_at=datetime(2026, 7, 8, 9, 0, tzinfo=timezone.utc),
                    end_at=datetime(2026, 7, 8, 10, 0, tzinfo=timezone.utc),
                ),
            )
            update_event(session, tenant_id="tenant-1", user_id=None, event_id=event.id, payload=CalendarEventUpdateRequest(summary="Updated"))
        session.commit()

        self.assertEqual(len(fake.puts), 2)
        self.assertTrue(fake.puts[0]["create"])
        self.assertEqual(fake.puts[1]["etag"], '"etag-1"')
        self.assertFalse(fake.puts[1]["create"])
        self.assertEqual(event.source_kind, "caldav")
        self.assertEqual(event.etag, '"etag-2"')
        self.assertIn("SUMMARY:Updated", event.raw_ics or "")

    def test_update_event_reports_remote_etag_conflict(self) -> None:
        session = self.Session()
        session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant"))
        calendar = create_calendar(session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Remote"))
        create_caldav_source(session, tenant_id="tenant-1", user_id=None, payload=CalendarCalDavSourceCreateRequest(calendar_id=calendar.id, collection_url="https://dav.example.test/cal"))
        event = caldav_event(calendar.id, summary="Remote", href="https://dav.example.test/cal/event-1.ics", etag='"etag-1"')
        session.add(event)
        session.commit()

        fake = FakeCalDAVClient(fail_precondition=True)
        with patch("govoplan_calendar.backend.service.caldav_client_for_source", return_value=fake):
            with self.assertRaisesRegex(CalendarError, "changed remotely"):
                update_event(session, tenant_id="tenant-1", user_id=None, event_id=event.id, payload=CalendarEventUpdateRequest(summary="Updated"))

    def test_delete_one_component_puts_remaining_resource_instead_of_deleting_object(self) -> None:
        session = self.Session()
        session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant"))
        calendar = create_calendar(session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Remote"))
        create_caldav_source(session, tenant_id="tenant-1", user_id=None, payload=CalendarCalDavSourceCreateRequest(calendar_id=calendar.id, collection_url="https://dav.example.test/cal"))
        href = "https://dav.example.test/cal/series.ics"
        master = caldav_event(calendar.id, summary="Master", href=href, etag='"etag-1"')
        override = caldav_event(calendar.id, summary="Override", href=href, etag='"etag-1"', recurrence_id="20260715T090000Z")
        session.add_all([master, override])
        session.commit()

        fake = FakeCalDAVClient(put_etags=['"etag-2"'])
        with patch("govoplan_calendar.backend.service.caldav_client_for_source", return_value=fake):
            delete_event(session, tenant_id="tenant-1", event_id=override.id)
        session.commit()

        self.assertEqual(len(fake.puts), 1)
        self.assertEqual(fake.deletes, [])
        self.assertIn("SUMMARY:Master", fake.puts[0]["ics"])
        self.assertNotIn("SUMMARY:Override", fake.puts[0]["ics"])
        self.assertEqual(master.etag, '"etag-2"')
        self.assertIsNotNone(override.deleted_at)

    def test_freebusy_expands_recurring_events_and_skips_transparent_items(self) -> None:
        session = self.Session()
        session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant"))
        calendar = create_calendar(session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Local"))
        create_event(
            session,
            tenant_id="tenant-1",
            user_id=None,
            payload=CalendarEventCreateRequest(
                calendar_id=calendar.id,
                uid="busy@example.test",
                summary="Busy",
                start_at=datetime(2026, 7, 8, 9, 0, tzinfo=timezone.utc),
                end_at=datetime(2026, 7, 8, 10, 0, tzinfo=timezone.utc),
                rrule={"FREQ": "WEEKLY", "COUNT": "2"},
            ),
        )
        create_event(
            session,
            tenant_id="tenant-1",
            user_id=None,
            payload=CalendarEventCreateRequest(
                calendar_id=calendar.id,
                uid="free@example.test",
                summary="Free",
                transparency="TRANSPARENT",
                start_at=datetime(2026, 7, 8, 11, 0, tzinfo=timezone.utc),
                end_at=datetime(2026, 7, 8, 12, 0, tzinfo=timezone.utc),
            ),
        )
        session.commit()

        busy = list_freebusy(
            session,
            tenant_id="tenant-1",
            start_at=datetime(2026, 7, 1, tzinfo=timezone.utc),
            end_at=datetime(2026, 7, 31, tzinfo=timezone.utc),
        )

        self.assertEqual([item["uid"] for item in busy], ["busy@example.test", "busy@example.test"])
        self.assertEqual(busy[0]["start_at"], datetime(2026, 7, 8, 9, 0, tzinfo=timezone.utc))

    def test_full_sync_imports_multi_vevent_resource_and_fetches_missing_calendar_data(self) -> None:
        session = self.Session()
        tenant = Tenant(id="tenant-1", slug="tenant-1", name="Tenant")
        session.add(tenant)
        calendar = create_calendar(
            session,
            tenant_id="tenant-1",
            user_id=None,
            payload=CalendarCollectionCreateRequest(name="Remote", metadata={}),
        )
        source = create_caldav_source(
            session,
            tenant_id="tenant-1",
            user_id=None,
            payload=CalendarCalDavSourceCreateRequest(calendar_id=calendar.id, collection_url="https://dav.example.test/cal"),
        )
        session.commit()

        href = "/cal/series.ics"
        client = FakeCalDAVClient(
            list_result=CalDAVReportResult(
                objects=[CalDAVObject(href=href, etag='"etag-1"', calendar_data=None)],
                sync_token="token-1",
                ctag="ctag-2",
            ),
            objects={href: recurring_resource()},
        )
        source, stats = sync_caldav_source(session, tenant_id="tenant-1", user_id=None, source_id=source.id, client=client)
        session.commit()

        events = session.query(CalendarEvent).filter(CalendarEvent.calendar_id == calendar.id, CalendarEvent.deleted_at.is_(None)).order_by(CalendarEvent.recurrence_id.asc()).all()
        self.assertEqual(stats.created, 2)
        self.assertEqual(stats.fetched, 1)
        self.assertTrue(stats.full_sync)
        self.assertEqual(source.sync_token, "token-1")
        self.assertEqual(source.ctag, "ctag-2")
        self.assertEqual(client.fetched, [href])
        self.assertEqual({event.summary for event in events}, {"Series", "Moved instance"})
        self.assertEqual({event.source_href for event in events}, {"https://dav.example.test/cal/series.ics"})
        self.assertEqual(calendar.metadata_["source_kind"], "caldav")

    def test_sync_token_deletion_soft_deletes_remote_resource_events(self) -> None:
        session = self.Session()
        tenant = Tenant(id="tenant-1", slug="tenant-1", name="Tenant")
        session.add(tenant)
        calendar = create_calendar(session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Remote"))
        source = create_caldav_source(
            session,
            tenant_id="tenant-1",
            user_id=None,
            payload=CalendarCalDavSourceCreateRequest(calendar_id=calendar.id, collection_url="https://dav.example.test/cal"),
        )
        source.sync_token = "token-1"
        session.add(
            CalendarEvent(
                tenant_id="tenant-1",
                calendar_id=calendar.id,
                uid="event-1@example.test",
                recurrence_id=None,
                summary="Remote",
                status="CONFIRMED",
                transparency="OPAQUE",
                classification="PUBLIC",
                start_at=datetime(2026, 7, 8, 9, 0, tzinfo=timezone.utc),
                all_day=False,
                attendees=[],
                categories=[],
                rdate=[],
                exdate=[],
                reminders=[],
                attachments=[],
                related_to=[],
                source_kind="caldav",
                source_href="https://dav.example.test/cal/event-1.ics",
                icalendar={},
                metadata_={"caldav": {"source_id": source.id}},
            )
        )
        session.commit()

        client = FakeCalDAVClient(sync_result=CalDAVReportResult(objects=[CalDAVObject(href="/cal/event-1.ics", deleted=True)], sync_token="token-2"))
        source, stats = sync_caldav_source(session, tenant_id="tenant-1", user_id=None, source_id=source.id, client=client)
        session.commit()

        event = session.query(CalendarEvent).filter(CalendarEvent.uid == "event-1@example.test").one()
        self.assertTrue(stats.used_sync_token)
        self.assertEqual(stats.deleted, 1)
        self.assertEqual(source.sync_token, "token-2")
        self.assertIsNotNone(event.deleted_at)

    def test_missing_resource_during_fetch_is_treated_as_remote_delete(self) -> None:
        session = self.Session()
        tenant = Tenant(id="tenant-1", slug="tenant-1", name="Tenant")
        session.add(tenant)
        calendar = create_calendar(session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Remote"))
        source = create_caldav_source(
            session,
            tenant_id="tenant-1",
            user_id=None,
            payload=CalendarCalDavSourceCreateRequest(calendar_id=calendar.id, collection_url="https://dav.example.test/cal"),
        )
        source.sync_token = "token-1"
        event = CalendarEvent(
            tenant_id="tenant-1",
            calendar_id=calendar.id,
            uid="event-1@example.test",
            recurrence_id=None,
            summary="Remote",
            status="CONFIRMED",
            transparency="OPAQUE",
            classification="PUBLIC",
            start_at=datetime(2026, 7, 8, 9, 0, tzinfo=timezone.utc),
            all_day=False,
            attendees=[],
            categories=[],
            rdate=[],
            exdate=[],
            reminders=[],
            attachments=[],
            related_to=[],
            source_kind="caldav",
            source_href="https://dav.example.test/cal/event-1.ics",
            icalendar={},
            metadata_={"caldav": {"source_id": source.id}},
        )
        session.add(event)
        session.commit()

        client = FakeCalDAVClient(
            sync_result=CalDAVReportResult(objects=[CalDAVObject(href="/cal/event-1.ics", etag='"gone"', calendar_data=None)], sync_token="token-2"),
            fetch_not_found={"/cal/event-1.ics"},
        )
        source, stats = sync_caldav_source(session, tenant_id="tenant-1", user_id=None, source_id=source.id, client=client)
        session.commit()

        self.assertTrue(stats.used_sync_token)
        self.assertEqual(stats.deleted, 1)
        self.assertEqual(source.last_status, "ok")
        self.assertEqual(source.sync_token, "token-2")
        self.assertIsNotNone(event.deleted_at)


def recurring_resource() -> str:
    return """BEGIN:VCALENDAR
VERSION:2.0
BEGIN:VEVENT
UID:series@example.test
DTSTART;TZID=Europe/Berlin:20260708T090000
DTEND;TZID=Europe/Berlin:20260708T100000
RRULE:FREQ=WEEKLY;COUNT=2
SUMMARY:Series
END:VEVENT
BEGIN:VEVENT
UID:series@example.test
RECURRENCE-ID;TZID=Europe/Berlin:20260715T090000
DTSTART;TZID=Europe/Berlin:20260715T110000
DTEND;TZID=Europe/Berlin:20260715T120000
SUMMARY:Moved instance
END:VEVENT
END:VCALENDAR
"""


def caldav_event(calendar_id: str, *, summary: str, href: str, etag: str, recurrence_id: str | None = None) -> CalendarEvent:
    return CalendarEvent(
        tenant_id="tenant-1",
        calendar_id=calendar_id,
        uid="series@example.test" if recurrence_id else "event-1@example.test",
        recurrence_id=recurrence_id,
        summary=summary,
        status="CONFIRMED",
        transparency="OPAQUE",
        classification="PUBLIC",
        start_at=datetime(2026, 7, 8, 9, 0, tzinfo=timezone.utc),
        end_at=datetime(2026, 7, 8, 10, 0, tzinfo=timezone.utc),
        all_day=False,
        attendees=[],
        categories=[],
        rdate=[],
        exdate=[],
        reminders=[],
        attachments=[],
        related_to=[],
        source_kind="caldav",
        source_href=href,
        etag=etag,
        icalendar={},
        metadata_={},
    )


if __name__ == "__main__":
    unittest.main()