from __future__ import annotations import unittest from datetime import datetime, timedelta, timezone from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from govoplan_access.backend.db import models as access_models # noqa: F401 from govoplan_calendar.backend.db.models import CalendarMigrationBatch from govoplan_calendar.backend.migrations_saga import ( cancel_calendar_migration_batch, get_calendar_migration_batch, ) from govoplan_calendar.backend.outbox import dispatch_calendar_outbox from govoplan_calendar.backend.schemas import ( CalendarCalDavSourceCreateRequest, CalendarCollectionCreateRequest, CalendarCollectionDeleteRequest, CalendarEventCreateRequest, CalendarEventUpdateRequest, ) from govoplan_calendar.backend.service import ( CalendarError, create_caldav_source, create_calendar, create_event, delete_calendar, update_event, ) from govoplan_core.db.base import Base, utcnow from govoplan_core.tenancy.scope import create_scope_tables from govoplan_tenancy.backend.db.models import Tenant from tests.test_outbox import StatefulCalDAVClient class CalendarRemoteMoveTests(unittest.TestCase): def setUp(self) -> None: self.engine = create_engine("sqlite+pysqlite:///:memory:", future=True) create_scope_tables(self.engine) Base.metadata.create_all(self.engine) self.Session = sessionmaker(bind=self.engine, expire_on_commit=False) self.session = self.Session() self.session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant")) self.calendar = create_calendar( self.session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Remote source"), ) self.source = create_caldav_source( self.session, tenant_id="tenant-1", user_id=None, payload=CalendarCalDavSourceCreateRequest( calendar_id=self.calendar.id, collection_url="https://dav.example.test/source", ), ) self.session.commit() def tearDown(self) -> None: self.session.close() self.engine.dispose() def create_event(self, uid: str): return create_event( self.session, tenant_id="tenant-1", user_id=None, payload=CalendarEventCreateRequest( calendar_id=self.calendar.id, uid=uid, summary="Planning", start_at=datetime(2026, 7, 8, 9, 0, tzinfo=timezone.utc), end_at=datetime(2026, 7, 8, 10, 0, tzinfo=timezone.utc), ), ) def create_target(self): calendar = create_calendar( self.session, tenant_id="tenant-1", user_id=None, payload=CalendarCollectionCreateRequest(name="Remote target"), ) source = create_caldav_source( self.session, tenant_id="tenant-1", user_id=None, payload=CalendarCalDavSourceCreateRequest( calendar_id=calendar.id, collection_url="https://dav.example.test/target", ), ) self.session.commit() return calendar, source def deliver_source_events(self, *events): self.session.commit() client = StatefulCalDAVClient() result = dispatch_calendar_outbox( self.session, tenant_id="tenant-1", client_factory=lambda _session, _source: client, ) self.assertEqual(result["succeeded"], len(events)) for event in events: self.assertEqual(event.source_kind, "caldav") self.assertIsNotNone(event.source_href) self.assertIsNotNone(event.etag) return client def start_move( self, target_calendar, *, make_target_default: bool = False, ) -> CalendarMigrationBatch: batch = delete_calendar( self.session, tenant_id="tenant-1", calendar_id=self.calendar.id, payload=CalendarCollectionDeleteRequest( event_action="move", target_calendar_id=target_calendar.id, external_action="remote_move", destructive_confirmation="MOVE REMOTE EVENTS", evidence_note="Approved migration window 42", make_target_default=make_target_default, ), ) self.assertIsInstance(batch, CalendarMigrationBatch) self.session.commit() return batch def clients(self, source_client, target_client): return lambda _session, source: ( source_client if source.id == self.source.id else target_client ) def test_all_resources_are_copied_before_conditional_source_delete(self) -> None: first = self.create_event("move-first@example.test") second = self.create_event("move-second@example.test") source_client = self.deliver_source_events(first, second) target_calendar, target_source = self.create_target() target_client = StatefulCalDAVClient() self.source.conflict_policy = "overwrite" target_source.conflict_policy = "overwrite" self.calendar.is_default = True self.session.commit() batch = self.start_move(target_calendar, make_target_default=True) result = dispatch_calendar_outbox( self.session, tenant_id="tenant-1", client_factory=self.clients(source_client, target_client), ) self.assertEqual(result["failed"], 0) batch = get_calendar_migration_batch( self.session, tenant_id="tenant-1", batch_id=batch.id ) self.assertEqual(batch.status, "completed") self.assertEqual(len(target_client.resources), 2) self.assertEqual(source_client.resources, {}) self.assertEqual(len(source_client.deletes), 2) self.assertTrue(all(item["etag"] for item in source_client.deletes)) self.assertTrue(all(item["overwrite"] is False for item in source_client.deletes)) self.assertTrue(all(item["overwrite"] is False for item in target_client.puts)) self.assertIsNotNone(self.calendar.deleted_at) self.assertIsNotNone(self.source.deleted_at) self.assertEqual(first.calendar_id, target_calendar.id) self.assertNotIn("calendar_migration", first.metadata_ or {}) self.assertIsNone(target_source.deleted_at) self.assertTrue(target_calendar.is_default) def test_partial_copy_failure_never_deletes_source(self) -> None: first = self.create_event("partial-first@example.test") second = self.create_event("partial-second@example.test") source_client = self.deliver_source_events(first, second) target_calendar, _target_source = self.create_target() target_client = StatefulCalDAVClient() target_client.fail_before_write = True batch = self.start_move(target_calendar) dispatch_calendar_outbox( self.session, tenant_id="tenant-1", client_factory=self.clients(source_client, target_client), ) batch = get_calendar_migration_batch( self.session, tenant_id="tenant-1", batch_id=batch.id ) self.assertEqual(batch.phase, "copying_destination") self.assertEqual(source_client.deletes, []) self.assertEqual(len(source_client.resources), 2) self.assertIsNone(self.calendar.deleted_at) def test_uid_collision_is_rejected_before_mutation(self) -> None: source_event = self.create_event("collision@example.test") self.deliver_source_events(source_event) target_calendar, target_source = self.create_target() target_event = create_event( self.session, tenant_id="tenant-1", user_id=None, payload=CalendarEventCreateRequest( calendar_id=target_calendar.id, uid=source_event.uid, summary="Existing target", start_at=datetime(2026, 7, 9, 9, 0, tzinfo=timezone.utc), ), ) self.session.commit() target_client = StatefulCalDAVClient() dispatch_calendar_outbox( self.session, tenant_id="tenant-1", client_factory=lambda _session, _source: target_client, ) with self.assertRaisesRegex(CalendarError, "already contains UID"): self.start_move(target_calendar) self.session.rollback() self.assertEqual(source_event.calendar_id, self.calendar.id) self.assertEqual(target_event.calendar_id, target_calendar.id) self.assertIsNone(self.calendar.deleted_at) self.assertIsNotNone(target_source.id) def test_cancel_before_source_delete_retains_both_remote_copies(self) -> None: event = self.create_event("cancel@example.test") source_client = self.deliver_source_events(event) target_calendar, _target_source = self.create_target() target_client = StatefulCalDAVClient() self.calendar.is_default = True self.session.commit() batch = self.start_move(target_calendar, make_target_default=True) batch = cancel_calendar_migration_batch( self.session, tenant_id="tenant-1", batch_id=batch.id, evidence_note="Operator cancelled approved migration", user_id=None, api_key_id=None, ) self.session.commit() self.assertEqual(batch.status, "cancel_requested") dispatch_calendar_outbox( self.session, tenant_id="tenant-1", client_factory=self.clients(source_client, target_client), ) batch = get_calendar_migration_batch( self.session, tenant_id="tenant-1", batch_id=batch.id ) self.assertEqual(batch.status, "cancelled") self.assertEqual(source_client.deletes, []) self.assertIsNone(self.calendar.deleted_at) self.assertTrue(self.source.sync_enabled) self.assertEqual(len(target_client.resources), 1) self.assertTrue(self.calendar.is_default) self.assertFalse(target_calendar.is_default) def test_concurrent_event_edit_is_blocked(self) -> None: event = self.create_event("locked@example.test") self.deliver_source_events(event) target_calendar, _target_source = self.create_target() self.start_move(target_calendar) with self.assertRaisesRegex(CalendarError, "blocked while"): update_event( self.session, tenant_id="tenant-1", user_id=None, event_id=event.id, payload=CalendarEventUpdateRequest(summary="Unsafe edit"), ) self.session.rollback() def test_source_etag_change_becomes_reconciliation_conflict(self) -> None: event = self.create_event("etag-conflict@example.test") source_client = self.deliver_source_events(event) target_calendar, _target_source = self.create_target() target_client = StatefulCalDAVClient() batch = self.start_move(target_calendar) first = dispatch_calendar_outbox( self.session, tenant_id="tenant-1", limit=1, client_factory=self.clients(source_client, target_client), ) self.assertEqual(first["succeeded"], 1) href = next(iter(source_client.resources)) ics, _etag = source_client.resources[href] source_client.resources[href] = (ics, '"concurrent-edit"') dispatch_calendar_outbox( self.session, tenant_id="tenant-1", client_factory=self.clients(source_client, target_client), ) batch = get_calendar_migration_batch( self.session, tenant_id="tenant-1", batch_id=batch.id ) self.assertEqual(batch.status, "blocked") self.assertEqual(batch.phase, "source_delete_conflict") self.assertIn(href, source_client.resources) self.assertIsNone(self.calendar.deleted_at) def test_crash_after_destination_write_is_reconciled_without_duplicate_put(self) -> None: event = self.create_event("crash-recovery@example.test") source_client = self.deliver_source_events(event) target_calendar, _target_source = self.create_target() target_client = StatefulCalDAVClient() target_client.fail_after_write = True batch = self.start_move(target_calendar) first = dispatch_calendar_outbox( self.session, tenant_id="tenant-1", limit=1, client_factory=self.clients(source_client, target_client), ) self.assertEqual(first["succeeded"], 1) target_client.fail_after_write = False dispatch_calendar_outbox( self.session, tenant_id="tenant-1", now=utcnow() + timedelta(seconds=10), client_factory=self.clients(source_client, target_client), ) batch = get_calendar_migration_batch( self.session, tenant_id="tenant-1", batch_id=batch.id ) self.assertEqual(batch.status, "completed") self.assertEqual(len(target_client.puts), 1) def test_crash_after_source_delete_is_reconciled_without_duplicate_delete(self) -> None: event = self.create_event("delete-crash@example.test") source_client = self.deliver_source_events(event) target_calendar, _target_source = self.create_target() target_client = StatefulCalDAVClient() batch = self.start_move(target_calendar) copied = dispatch_calendar_outbox( self.session, tenant_id="tenant-1", limit=1, client_factory=self.clients(source_client, target_client), ) self.assertEqual(copied["succeeded"], 1) source_client.fail_after_write = True deleted = dispatch_calendar_outbox( self.session, tenant_id="tenant-1", limit=1, client_factory=self.clients(source_client, target_client), ) self.assertEqual(deleted["succeeded"], 1) source_client.fail_after_write = False dispatch_calendar_outbox( self.session, tenant_id="tenant-1", client_factory=self.clients(source_client, target_client), ) batch = get_calendar_migration_batch( self.session, tenant_id="tenant-1", batch_id=batch.id ) self.assertEqual(batch.status, "completed") self.assertEqual(len(source_client.deletes), 1) if __name__ == "__main__": unittest.main()