from __future__ import annotations import unittest from datetime import datetime, timezone from unittest.mock import patch from sqlalchemy import create_engine, inspect 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, CalDAVDiscoveryCalendar, CalDAVError, CalDAVNotFound, CalDAVObject, CalDAVPreconditionFailed, CalDAVReportResult, CalDAVWriteResult, parse_discovery_multistatus, parse_multistatus, ) from govoplan_calendar.backend.db.models import CalendarEvent, CalendarOutboxOperation, CalendarSyncCredential from govoplan_calendar.backend.manifest import manifest from govoplan_calendar.backend.outbox import dispatch_calendar_outbox from govoplan_calendar.backend.schemas import ( CalendarCalDavDiscoveryRequest, CalendarCalDavSourceCreateRequest, CalendarCalDavSourceUpdateRequest, CalendarCollectionCreateRequest, CalendarEventCreateRequest, CalendarEventUpdateRequest, ) from govoplan_calendar.backend.service import ( CALDAV_INTERNAL_CREDENTIAL_PREFIX, CalendarError, _CalDAVDiscoveryAuth, _caldav_discovery_client, _caldav_discovery_response, _plan_sync_source_creation, caldav_client_for_source, caldav_source_response, create_calendar, create_caldav_source, create_event, delete_caldav_source, delete_calendar, delete_event, discover_caldav_calendars, list_caldav_sources, list_freebusy, resolve_caldav_credential_ref, resolve_trusted_deployment_caldav_credential_ref, sync_caldav_source, sync_due_caldav_sources, update_caldav_source, update_event, ) from govoplan_calendar.backend.router import _caldav_discovery_http_error from govoplan_core.db.base import Base from govoplan_core.security.credential_envelopes import create_credential_envelope 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_discovery_parser_is_independent_from_transport(self) -> None: result = parse_discovery_multistatus( b""" /calendars/ada/work/ Work HTTP/1.1 200 OK """ ) self.assertEqual(1, len(result)) self.assertEqual("Work", result[0].display_name) self.assertTrue(result[0].is_calendar) self.assertEqual(("VEVENT",), result[0].supported_components) def test_discovery_auth_client_and_response_do_not_expose_secret(self) -> None: auth = _CalDAVDiscoveryAuth( auth_type="basic", username="ada", credential_ref=None, secret="calendar-secret", ) client = _caldav_discovery_client( "https://dav.example.test/cal", auth, ) response = _caldav_discovery_response( ( CalDAVDiscoveryCalendar( collection_url="https://dav.example.test/cal/work/", href="/cal/work/", display_name="Work", ), ) ) self.assertNotIn("calendar-secret", repr(auth)) self.assertEqual("calendar-secret", client.password) self.assertEqual("Work", response[0]["display_name"]) self.assertNotIn("password", response[0]) self.assertNotIn("credential_ref", response[0]) def test_discovery_error_translation_is_stable(self) -> None: error = _caldav_discovery_http_error( CalDAVError("CalDAV endpoint rejected PROPFIND") ) self.assertEqual(422, error.status_code) self.assertEqual( "CalDAV endpoint rejected PROPFIND", error.detail, ) def test_sync_source_creation_plan_is_secret_free(self) -> None: plan = _plan_sync_source_creation( CalendarCalDavSourceCreateRequest( calendar_id="calendar-1", collection_url="dav.example.test/cal", auth_type="basic", username="ada", password="calendar-secret", ) ) self.assertEqual("https://dav.example.test/cal/", plan.collection_url) self.assertTrue(plan.has_inline_secret) self.assertNotIn("calendar-secret", repr(plan)) def test_parse_multistatus_extracts_objects_sync_token_and_deletions(self) -> None: result = parse_multistatus( b""" token-2 /cal/event-1.ics "etag-1" BEGIN:VCALENDAR BEGIN:VEVENT UID:event-1@example.test DTSTART:20260708T090000Z SUMMARY:One END:VEVENT END:VCALENDAR HTTP/1.1 200 OK /cal/event-2.ics HTTP/1.1 404 Not Found """ ) 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""" /remote.php/dav/ /remote.php/dav/principals/users/ada/ HTTP/1.1 200 OK """ if url == "https://cloud.example.test/remote.php/dav/principals/users/ada/": return 207, {}, b""" /remote.php/dav/principals/users/ada/ /remote.php/dav/calendars/ada/ HTTP/1.1 200 OK """ if url == "https://cloud.example.test/remote.php/dav/calendars/ada/": return 207, {}, b""" /remote.php/dav/calendars/ada/ HTTP/1.1 200 OK /remote.php/dav/calendars/ada/personal/ Personal #44aa99ff ctag-1 HTTP/1.1 200 OK """ 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""" /remote.php/dav/calendars/ada/personal/ Personal HTTP/1.1 200 OK """ 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") def test_object_url_rejects_off_origin_and_cross_collection_hrefs(self) -> None: client = CalDAVClient(collection_url="https://dav.example.test/cal") with self.assertRaisesRegex(CalDAVError, "collection origin"): client.object_url("https://evil.example.test/steal.ics") with self.assertRaisesRegex(CalDAVError, "collection path"): client.object_url("https://dav.example.test/other/steal.ics") self.assertEqual( client.object_url("/cal/event.ics"), "https://dav.example.test/cal/event.ics", ) 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) self.sessions = [] def tearDown(self) -> None: for session in reversed(self.sessions): session.close() Base.metadata.drop_all(bind=self.engine) self.engine.dispose() def session(self): session = self.Session() self.sessions.append(session) return session 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_source_can_resolve_reusable_core_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"), ) credential = create_credential_envelope( session, tenant_id="tenant-1", scope_type="tenant", scope_id="tenant-1", name="Shared DAV login", credential_kind="username_password", public_data={"username": "ada"}, secret_data={"password": "secret"}, allowed_modules=["calendar"], inherit_to_lower_scopes=True, ) 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", credential_ref=f"credential-envelope:{credential.id}", ), ) session.commit() client = caldav_client_for_source(session, source) response = caldav_source_response(source) self.assertEqual(client.username, "ada") self.assertEqual(client.password, "secret") self.assertEqual(response["credential_envelope_id"], credential.id) self.assertEqual(session.query(CalendarSyncCredential).count(), 0) def test_api_source_and_discovery_reject_caller_selected_credential_references(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"), ) with self.assertRaisesRegex(CalendarError, "Caller-supplied credential references"): create_caldav_source( session, tenant_id="tenant-1", user_id=None, payload=CalendarCalDavSourceCreateRequest( calendar_id=calendar.id, collection_url="https://attacker.example.test/cal", auth_type="bearer", credential_ref="env:MASTER_KEY_B64", ), ) with self.assertRaisesRegex(CalendarError, "Caller-supplied credential references"): discover_caldav_calendars( session, tenant_id="tenant-1", payload=CalendarCalDavDiscoveryRequest( url="https://attacker.example.test/cal", auth_type="bearer", credential_ref="vault:another-tenant", ), ) 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", ), ) with self.assertRaisesRegex(CalendarError, "Caller-supplied credential references"): update_caldav_source( session, tenant_id="tenant-1", source_id=source.id, payload=CalendarCalDavSourceUpdateRequest( credential_ref="env:MASTER_KEY_B64", ), ) def test_provider_credentials_are_wrapped_in_tenant_and_source_owned_rows(self) -> None: class Provider: def __init__(self) -> None: self.values: dict[str, str] = {} self.deleted: list[str] = [] def store_secret(self, *, scope: str, name: str, value: str) -> str: reference = f"vault:{scope}:{name}" self.values[reference] = value return reference def read_secret(self, secret_ref: str) -> str | None: return self.values.get(secret_ref) def delete_secret(self, secret_ref: str) -> None: self.deleted.append(secret_ref) self.values.pop(secret_ref, None) provider = Provider() 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"), ) with patch("govoplan_calendar.backend.service.secret_provider", return_value=provider): 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() provider_ref = str((credential.metadata_ or {})["provider_ref"]) self.assertTrue(source.credential_ref.startswith(CALDAV_INTERNAL_CREDENTIAL_PREFIX)) self.assertNotEqual(source.credential_ref, provider_ref) self.assertEqual(caldav_client_for_source(session, source).password, "secret") self.assertIsNone(caldav_source_response(source)["credential_ref"]) self.assertTrue(caldav_source_response(source)["has_credential"]) self.assertIsNone( resolve_caldav_credential_ref( session, tenant_id="another-tenant", source_id=source.id, credential_ref=source.credential_ref, ) ) self.assertIsNone( resolve_caldav_credential_ref( session, tenant_id="tenant-1", source_id="another-source", credential_ref=source.credential_ref, ) ) other_calendar = create_calendar( session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Other remote"), ) other_source = create_caldav_source( session, tenant_id="tenant-1", user_id=None, payload=CalendarCalDavSourceCreateRequest( calendar_id=other_calendar.id, collection_url="https://dav.example.test/other-cal", ), ) other_source.auth_type = "basic" other_source.username = "mallory" other_source.credential_ref = source.credential_ref with self.assertRaisesRegex(CalendarError, "not a server-owned credential"): caldav_client_for_source(session, other_source) delete_calendar(session, tenant_id="tenant-1", calendar_id=other_calendar.id) self.assertEqual(provider.deleted, []) with patch("govoplan_calendar.backend.service.audit_event") as audit: delete_calendar(session, tenant_id="tenant-1", calendar_id=calendar.id) self.assertEqual(provider.deleted, [provider_ref]) self.assertNotIn(provider_ref, provider.values) self.assertIsNone(source.credential_ref) self.assertIsNone(credential.secret_encrypted) self.assertNotIn("provider_ref", credential.metadata_ or {}) audit.assert_called_once() audit_call = audit.call_args.kwargs self.assertEqual(audit_call["action"], "calendar.sync_credential_deleted") self.assertEqual(audit_call["object_id"], credential.id) self.assertEqual(audit_call["details"]["storage_backend"], "external_secret_provider") self.assertEqual(audit_call["details"]["deletion_reason"], "calendar_deleted") self.assertNotIn("credential_ref", audit_call["details"]) self.assertNotIn("provider_ref", audit_call["details"]) self.assertNotIn(provider_ref, repr(audit_call["details"])) def test_legacy_unowned_refs_are_neither_read_nor_deleted(self) -> None: class Provider: def __init__(self) -> None: self.read: list[str] = [] self.deleted: list[str] = [] def read_secret(self, secret_ref: str) -> str | None: self.read.append(secret_ref) return "other-tenant-secret" def delete_secret(self, secret_ref: str) -> None: self.deleted.append(secret_ref) provider = Provider() 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.auth_type = "bearer" source.credential_ref = "vault:another-tenant" session.commit() with patch("govoplan_calendar.backend.service.secret_provider", return_value=provider): with self.assertRaisesRegex(CalendarError, "not a server-owned credential"): caldav_client_for_source(session, source) delete_calendar(session, tenant_id="tenant-1", calendar_id=calendar.id) self.assertEqual(provider.read, []) self.assertEqual(provider.deleted, []) def test_trusted_deployment_env_resolution_is_explicit_and_separate(self) -> None: with patch.dict("os.environ", {"CALDAV_DEPLOYMENT_TOKEN": "trusted-token"}): self.assertEqual( resolve_trusted_deployment_caldav_credential_ref("env:CALDAV_DEPLOYMENT_TOKEN"), "trusted-token", ) with self.assertRaisesRegex(CalendarError, "must use the env: prefix"): resolve_trusted_deployment_caldav_credential_ref("vault:token") 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.assertIsNone(source.credential_ref) self.assertIsNotNone(credential.deleted_at) self.assertIsNone(credential.secret_encrypted) self.assertEqual(credential.metadata_, {"source_id": source.id}) self.assertEqual(list_caldav_sources(session, tenant_id="tenant-1"), []) def test_external_credential_deletion_fails_closed_when_provider_is_unavailable(self) -> None: class Provider: def __init__(self) -> None: self.values: dict[str, str] = {} def store_secret(self, *, scope: str, name: str, value: str) -> str: reference = f"vault:{scope}:{name}" self.values[reference] = value return reference def read_secret(self, secret_ref: str) -> str | None: return self.values.get(secret_ref) def delete_secret(self, secret_ref: str) -> None: self.values.pop(secret_ref, None) provider = Provider() 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"), ) with patch("govoplan_calendar.backend.service.secret_provider", return_value=provider): source = create_caldav_source( session, tenant_id="tenant-1", user_id=None, payload=CalendarCalDavSourceCreateRequest( calendar_id=calendar.id, collection_url="https://dav.example.test/provider-down", auth_type="bearer", bearer_token="do-not-log-this", ), ) session.commit() credential = session.query(CalendarSyncCredential).one() provider_ref = str((credential.metadata_ or {})["provider_ref"]) with patch("govoplan_calendar.backend.service.secret_provider", return_value=None), self.assertRaisesRegex( CalendarError, "secret provider is unavailable", ): delete_caldav_source(session, tenant_id="tenant-1", source_id=source.id) session.rollback() self.assertIsNone(source.deleted_at) self.assertEqual(source.credential_ref, f"{CALDAV_INTERNAL_CREDENTIAL_PREFIX}{credential.id}") self.assertIsNone(credential.deleted_at) self.assertEqual((credential.metadata_ or {})["provider_ref"], provider_ref) self.assertIn(provider_ref, provider.values) with patch("govoplan_calendar.backend.service.secret_provider", return_value=provider), patch.object( provider, "delete_secret", side_effect=RuntimeError(f"provider failed for {provider_ref}"), ): with self.assertRaises(CalendarError) as raised: delete_caldav_source(session, tenant_id="tenant-1", source_id=source.id) self.assertNotIn(provider_ref, str(raised.exception)) session.rollback() self.assertIsNone(source.deleted_at) self.assertIsNone(credential.deleted_at) def test_disabling_auth_immediately_scrubs_and_audits_database_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/disable-auth", auth_type="basic", username="ada", password="do-not-log-this", ), ) session.commit() credential = session.query(CalendarSyncCredential).one() self.assertIsNotNone(credential.secret_encrypted) with patch("govoplan_calendar.backend.service.audit_event") as audit: update_caldav_source( session, tenant_id="tenant-1", source_id=source.id, payload=CalendarCalDavSourceUpdateRequest(auth_type="none"), ) session.commit() self.assertIsNone(source.credential_ref) self.assertIsNone(credential.secret_encrypted) self.assertIsNotNone(credential.deleted_at) audit.assert_called_once() self.assertEqual( audit.call_args.kwargs["details"]["deletion_reason"], "authentication_disabled", ) self.assertNotIn("do-not-log-this", repr(audit.call_args)) def test_external_credential_replacement_fails_closed_when_provider_is_unavailable(self) -> None: class Provider: def __init__(self) -> None: self.values: dict[str, str] = {} def store_secret(self, *, scope: str, name: str, value: str) -> str: reference = f"vault:{scope}:{name}" self.values[reference] = value return reference def read_secret(self, secret_ref: str) -> str | None: return self.values.get(secret_ref) def delete_secret(self, secret_ref: str) -> None: self.values.pop(secret_ref, None) provider = Provider() 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"), ) with patch("govoplan_calendar.backend.service.secret_provider", return_value=provider): source = create_caldav_source( session, tenant_id="tenant-1", user_id=None, payload=CalendarCalDavSourceCreateRequest( calendar_id=calendar.id, collection_url="https://dav.example.test/replacement-provider-down", auth_type="bearer", bearer_token="old-secret", ), ) session.commit() credential = session.query(CalendarSyncCredential).one() provider_ref = str((credential.metadata_ or {})["provider_ref"]) with patch("govoplan_calendar.backend.service.secret_provider", return_value=None), self.assertRaisesRegex( CalendarError, "cannot be replaced.*provider is unavailable", ): update_caldav_source( session, tenant_id="tenant-1", source_id=source.id, payload=CalendarCalDavSourceUpdateRequest(bearer_token="new-secret"), ) session.rollback() self.assertEqual(source.credential_ref, f"{CALDAV_INTERNAL_CREDENTIAL_PREFIX}{credential.id}") self.assertEqual((credential.metadata_ or {})["provider_ref"], provider_ref) self.assertEqual(provider.values[provider_ref], "old-secret") def test_external_credential_delete_is_retryable_after_database_rollback(self) -> None: class Provider: def __init__(self) -> None: self.values: dict[str, str] = {} self.deletes: list[str] = [] def store_secret(self, *, scope: str, name: str, value: str) -> str: reference = f"vault:{scope}:{name}" self.values[reference] = value return reference def read_secret(self, secret_ref: str) -> str | None: return self.values.get(secret_ref) def delete_secret(self, secret_ref: str) -> None: self.deletes.append(secret_ref) if secret_ref not in self.values: raise KeyError("already deleted") self.values.pop(secret_ref, None) provider = Provider() 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"), ) with patch("govoplan_calendar.backend.service.secret_provider", return_value=provider): source = create_caldav_source( session, tenant_id="tenant-1", user_id=None, payload=CalendarCalDavSourceCreateRequest( calendar_id=calendar.id, collection_url="https://dav.example.test/retry-delete", auth_type="basic", username="ada", password="do-not-log-this", ), ) session.commit() credential = session.query(CalendarSyncCredential).one() provider_ref = str((credential.metadata_ or {})["provider_ref"]) with patch("govoplan_calendar.backend.service.audit_event"): delete_caldav_source(session, tenant_id="tenant-1", source_id=source.id) self.assertNotIn(provider_ref, provider.values) self.assertIsNotNone(credential.deleted_at) self.assertIsNone(source.credential_ref) session.rollback() self.assertIsNone(source.deleted_at) session.refresh(credential) self.assertIsNone(credential.deleted_at) self.assertEqual((credential.metadata_ or {})["provider_ref"], provider_ref) self.assertIsNone(resolve_caldav_credential_ref( session, tenant_id="tenant-1", source_id=source.id, credential_ref=source.credential_ref, )) with patch("govoplan_calendar.backend.service.audit_event") as audit: delete_caldav_source(session, tenant_id="tenant-1", source_id=source.id) session.commit() self.assertEqual(provider.deletes, [provider_ref, provider_ref]) self.assertIsNotNone(source.deleted_at) self.assertIsNone(source.credential_ref) self.assertIsNotNone(credential.deleted_at) self.assertEqual(audit.call_count, 1) def test_destructive_module_retirement_deletes_external_credentials_before_tables(self) -> None: class Provider: def __init__(self) -> None: self.values: dict[str, str] = {} self.deleted: list[str] = [] def store_secret(self, *, scope: str, name: str, value: str) -> str: reference = f"vault:{scope}:{name}" self.values[reference] = value return reference def read_secret(self, secret_ref: str) -> str | None: return self.values.get(secret_ref) def delete_secret(self, secret_ref: str) -> None: self.deleted.append(secret_ref) self.values.pop(secret_ref, None) provider = Provider() 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"), ) with patch("govoplan_calendar.backend.service.secret_provider", return_value=provider): create_caldav_source( session, tenant_id="tenant-1", user_id=None, payload=CalendarCalDavSourceCreateRequest( calendar_id=calendar.id, collection_url="https://dav.example.test/module-retirement", auth_type="bearer", bearer_token="do-not-log-this", ), ) session.commit() credential = session.query(CalendarSyncCredential).one() provider_ref = str((credential.metadata_ or {})["provider_ref"]) retirement_provider = manifest.migration_spec.retirement_provider assert retirement_provider is not None plan = retirement_provider(session, "calendar") assert plan.destroy_data_executor is not None with patch("govoplan_calendar.backend.service.audit_event") as audit: plan.destroy_data_executor(session, "calendar") # Retirement intentionally participates in the caller's database # transaction. Commit before inspecting through a new Engine # connection; otherwise SQLite rolls back the uncommitted DDL when # the temporary inspector connection is returned to the pool. session.commit() self.assertEqual(provider.deleted, [provider_ref]) self.assertEqual(provider.values, {}) self.assertEqual(audit.call_count, 1) self.assertEqual( audit.call_args.kwargs["details"]["deletion_reason"], "module_data_retired", ) self.assertFalse(inspect(self.engine).has_table("calendar_sync_credentials")) self.assertFalse(inspect(self.engine).has_table("calendar_sync_sources")) 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_coalesce_to_committed_outbox_state(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"']) 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")) self.assertEqual(fake.puts, []) session.commit() operations = session.query(CalendarOutboxOperation).all() self.assertCountEqual([operation.status for operation in operations], ["superseded", "pending"]) dispatch_calendar_outbox(session, tenant_id="tenant-1", client_factory=lambda _session, _source: fake) self.assertEqual(len(fake.puts), 1) self.assertTrue(fake.puts[0]["create"]) self.assertIn("SUMMARY:Updated", fake.puts[0]["ics"]) self.assertEqual(event.source_kind, "caldav") self.assertEqual(event.etag, '"etag-1"') 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, objects={event.source_href: recurring_resource()}, ) update_event(session, tenant_id="tenant-1", user_id=None, event_id=event.id, payload=CalendarEventUpdateRequest(summary="Updated")) session.commit() result = dispatch_calendar_outbox( session, tenant_id="tenant-1", client_factory=lambda _session, _source: fake, ) self.assertEqual(result["failed"], 1) operation = session.query(CalendarOutboxOperation).one() self.assertEqual(operation.status, "conflict") self.assertIn("changed remotely", operation.last_error or "") 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"']) delete_event(session, tenant_id="tenant-1", event_id=override.id) session.commit() dispatch_calendar_outbox(session, tenant_id="tenant-1", client_factory=lambda _session, _source: fake) 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()