1820 lines
67 KiB
Python
1820 lines
67 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from datetime import datetime, timedelta, 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
|
|
from govoplan_calendar.backend.caldav import (
|
|
CalDAVError,
|
|
CalDAVNotFound,
|
|
CalDAVObject,
|
|
CalDAVPreconditionFailed,
|
|
CalDAVWriteResult,
|
|
)
|
|
from govoplan_calendar.backend.db.models import CalendarOutboxOperation
|
|
from govoplan_calendar.backend.outbox import (
|
|
SqlCalendarOutboxProvider,
|
|
claim_next_calendar_outbox_operation,
|
|
discard_calendar_outbox_operation,
|
|
dispatch_calendar_outbox,
|
|
execute_calendar_outbox_operation,
|
|
purge_terminal_calendar_outbox_operations,
|
|
reconcile_calendar_outbox_operation,
|
|
retry_calendar_outbox_operation,
|
|
)
|
|
from govoplan_calendar.backend.schemas import (
|
|
CalendarCalDavSourceCreateRequest,
|
|
CalendarCalDavSourceUpdateRequest,
|
|
CalendarCollectionCreateRequest,
|
|
CalendarCollectionDeleteRequest,
|
|
CalendarEventCreateRequest,
|
|
CalendarEventUpdateRequest,
|
|
)
|
|
from govoplan_calendar.backend.service import (
|
|
CalendarError,
|
|
apply_caldav_report,
|
|
create_caldav_source,
|
|
create_calendar,
|
|
create_event,
|
|
delete_caldav_source,
|
|
delete_calendar,
|
|
delete_event,
|
|
import_ics_event,
|
|
normalize_caldav_href,
|
|
update_caldav_source,
|
|
update_event,
|
|
)
|
|
from govoplan_calendar.backend.caldav import CalDAVReportResult
|
|
from govoplan_core.core.change_sequence import ChangeSequenceEntry
|
|
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
|
|
|
|
|
|
class StatefulCalDAVClient:
|
|
def __init__(self) -> None:
|
|
self.resources: dict[str, tuple[str, str]] = {}
|
|
self.puts: list[dict[str, object]] = []
|
|
self.deletes: list[dict[str, object]] = []
|
|
self.next_etag = 1
|
|
self.fail_before_write = False
|
|
self.fail_after_write = False
|
|
self.fetch_fails = False
|
|
self.omit_put_etag = False
|
|
|
|
def _etag(self) -> str:
|
|
value = f'"etag-{self.next_etag}"'
|
|
self.next_etag += 1
|
|
return value
|
|
|
|
def put_object(
|
|
self,
|
|
href: str,
|
|
ics: str,
|
|
*,
|
|
etag: str | None = None,
|
|
create: bool = False,
|
|
overwrite: bool = False,
|
|
) -> CalDAVWriteResult:
|
|
self.puts.append(
|
|
{"href": href, "ics": ics, "etag": etag, "create": create, "overwrite": overwrite}
|
|
)
|
|
if self.fail_before_write:
|
|
raise CalDAVError("network unavailable")
|
|
current = self.resources.get(href)
|
|
if create and current is not None and not overwrite:
|
|
raise CalDAVPreconditionFailed("already exists")
|
|
if not create and not overwrite and (not etag or current is None or current[1] != etag):
|
|
raise CalDAVPreconditionFailed("changed")
|
|
next_etag = self._etag()
|
|
self.resources[href] = (ics, next_etag)
|
|
if self.fail_after_write:
|
|
raise CalDAVError("connection lost after remote commit")
|
|
return CalDAVWriteResult(
|
|
href=href,
|
|
etag=None if self.omit_put_etag else next_etag,
|
|
status=201 if create else 204,
|
|
)
|
|
|
|
def delete_object(
|
|
self,
|
|
href: str,
|
|
*,
|
|
etag: str | None = None,
|
|
overwrite: bool = False,
|
|
) -> CalDAVWriteResult:
|
|
self.deletes.append({"href": href, "etag": etag, "overwrite": overwrite})
|
|
if self.fail_before_write:
|
|
raise CalDAVError("network unavailable")
|
|
current = self.resources.get(href)
|
|
if current is None:
|
|
return CalDAVWriteResult(href=href, status=404)
|
|
if not overwrite and (not etag or current[1] != etag):
|
|
raise CalDAVPreconditionFailed("changed")
|
|
del self.resources[href]
|
|
if self.fail_after_write:
|
|
raise CalDAVError("connection lost after remote commit")
|
|
return CalDAVWriteResult(href=href, status=204)
|
|
|
|
def fetch_object_state(self, href: str) -> CalDAVObject:
|
|
if self.fetch_fails:
|
|
raise CalDAVError("reconciliation unavailable")
|
|
current = self.resources.get(href)
|
|
if current is None:
|
|
raise CalDAVNotFound("not found")
|
|
return CalDAVObject(href=href, calendar_data=current[0], etag=current[1])
|
|
|
|
|
|
class CalendarOutboxTests(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"),
|
|
)
|
|
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/cal",
|
|
),
|
|
)
|
|
self.session.commit()
|
|
|
|
def tearDown(self) -> None:
|
|
self.session.close()
|
|
self.engine.dispose()
|
|
|
|
def create_local_event(
|
|
self,
|
|
*,
|
|
summary: str = "Planning",
|
|
uid: str = "event-1@example.test",
|
|
source_href: str | None = None,
|
|
):
|
|
return create_event(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id=None,
|
|
payload=CalendarEventCreateRequest(
|
|
calendar_id=self.calendar.id,
|
|
uid=uid,
|
|
summary=summary,
|
|
start_at=datetime(2026, 7, 8, 9, 0, tzinfo=timezone.utc),
|
|
end_at=datetime(2026, 7, 8, 10, 0, tzinfo=timezone.utc),
|
|
source_href=source_href,
|
|
),
|
|
)
|
|
|
|
def test_event_and_outbox_are_atomic_and_network_waits_for_commit(self) -> None:
|
|
event = self.create_local_event()
|
|
operation_id = (event.metadata_ or {})["caldav"]["outbox_operation_id"]
|
|
self.assertIsNotNone(self.session.get(CalendarOutboxOperation, operation_id))
|
|
|
|
self.session.rollback()
|
|
|
|
self.assertIsNone(self.session.get(CalendarOutboxOperation, operation_id))
|
|
self.assertEqual(self.session.query(CalendarOutboxOperation).count(), 0)
|
|
|
|
def test_after_commit_enqueues_registered_worker_without_doing_network_io(self) -> None:
|
|
with (
|
|
patch("govoplan_calendar.backend.outbox._enqueue_celery_dispatch") as enqueue,
|
|
patch("govoplan_calendar.backend.outbox._remote_matches_operation") as remote_probe,
|
|
):
|
|
self.create_local_event()
|
|
self.session.commit()
|
|
|
|
enqueue.assert_called_once_with("tenant-1")
|
|
remote_probe.assert_not_called()
|
|
|
|
def test_savepoint_commit_does_not_dispatch_before_outer_commit(self) -> None:
|
|
with patch("govoplan_calendar.backend.outbox._enqueue_celery_dispatch") as enqueue:
|
|
self.create_local_event()
|
|
with self.session.begin_nested():
|
|
self.session.flush()
|
|
enqueue.assert_not_called()
|
|
self.session.commit()
|
|
|
|
enqueue.assert_called_once_with("tenant-1")
|
|
|
|
def test_savepoint_rollback_preserves_outer_transaction_wakeup(self) -> None:
|
|
with patch("govoplan_calendar.backend.outbox._enqueue_celery_dispatch") as enqueue:
|
|
event = self.create_local_event(summary="Outer desired state")
|
|
try:
|
|
with self.session.begin_nested():
|
|
update_event(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id=None,
|
|
event_id=event.id,
|
|
payload=CalendarEventUpdateRequest(summary="Rolled back desired state"),
|
|
)
|
|
raise RuntimeError("roll back savepoint")
|
|
except RuntimeError:
|
|
pass
|
|
enqueue.assert_not_called()
|
|
self.session.commit()
|
|
|
|
enqueue.assert_called_once_with("tenant-1")
|
|
self.assertEqual(
|
|
self.session.query(CalendarOutboxOperation).filter_by(status="pending").count(),
|
|
1,
|
|
)
|
|
|
|
def test_terminal_retention_purges_only_resolved_old_history(self) -> None:
|
|
now = datetime(2026, 7, 20, 12, 0, tzinfo=timezone.utc)
|
|
operation_ids: dict[str, str] = {}
|
|
for index, status in enumerate(
|
|
("succeeded", "superseded", "cancelled", "conflict", "dead", "pending")
|
|
):
|
|
event = self.create_local_event(
|
|
summary=f"Retention {status}",
|
|
uid=f"retention-{index}@example.test",
|
|
)
|
|
operation_id = (event.metadata_ or {})["caldav"]["outbox_operation_id"]
|
|
operation = self.session.get(CalendarOutboxOperation, operation_id)
|
|
assert operation is not None
|
|
operation.status = status
|
|
operation.completed_at = now - timedelta(days=31)
|
|
operation_ids[status] = operation.id
|
|
recent_event = self.create_local_event(
|
|
summary="Recent success",
|
|
uid="retention-recent@example.test",
|
|
)
|
|
recent_operation_id = (recent_event.metadata_ or {})["caldav"]["outbox_operation_id"]
|
|
recent_operation = self.session.get(CalendarOutboxOperation, recent_operation_id)
|
|
assert recent_operation is not None
|
|
recent_operation.status = "succeeded"
|
|
recent_operation.completed_at = now - timedelta(days=29)
|
|
self.session.commit()
|
|
|
|
purged = purge_terminal_calendar_outbox_operations(
|
|
self.session,
|
|
retention_days=30,
|
|
tenant_id="tenant-1",
|
|
now=now,
|
|
)
|
|
|
|
self.assertEqual(purged, 3)
|
|
for status in ("succeeded", "superseded", "cancelled"):
|
|
self.assertEqual(
|
|
self.session.query(CalendarOutboxOperation)
|
|
.filter(CalendarOutboxOperation.id == operation_ids[status])
|
|
.count(),
|
|
0,
|
|
)
|
|
for status in ("conflict", "dead", "pending"):
|
|
self.assertEqual(
|
|
self.session.query(CalendarOutboxOperation)
|
|
.filter(CalendarOutboxOperation.id == operation_ids[status])
|
|
.count(),
|
|
1,
|
|
)
|
|
self.assertEqual(
|
|
self.session.query(CalendarOutboxOperation)
|
|
.filter(CalendarOutboxOperation.id == recent_operation.id)
|
|
.count(),
|
|
1,
|
|
)
|
|
|
|
def test_zero_retention_disables_cleanup_and_provider_validates_setting(self) -> None:
|
|
event = self.create_local_event(uid="retention-disabled@example.test")
|
|
operation_id = (event.metadata_ or {})["caldav"]["outbox_operation_id"]
|
|
operation = self.session.get(CalendarOutboxOperation, operation_id)
|
|
assert operation is not None
|
|
operation.status = "succeeded"
|
|
operation.completed_at = utcnow() - timedelta(days=365)
|
|
self.session.commit()
|
|
|
|
provider = SqlCalendarOutboxProvider(terminal_retention_days=0)
|
|
result = provider.dispatch_due(self.session, tenant_id="tenant-1")
|
|
|
|
self.assertEqual(result["purged"], 0)
|
|
self.assertIsNotNone(self.session.get(CalendarOutboxOperation, operation.id))
|
|
with self.assertRaisesRegex(ValueError, "zero or greater"):
|
|
SqlCalendarOutboxProvider(terminal_retention_days=-1)
|
|
with self.assertRaisesRegex(ValueError, "must be an integer"):
|
|
SqlCalendarOutboxProvider(terminal_retention_days=True)
|
|
|
|
def test_remote_success_before_local_completion_is_reconciled_idempotently(self) -> None:
|
|
self.create_local_event()
|
|
self.session.commit()
|
|
operation = claim_next_calendar_outbox_operation(self.session, tenant_id="tenant-1")
|
|
assert operation is not None
|
|
href = operation.resource_href
|
|
payload = operation.payload_ics or ""
|
|
operation_id = operation.id
|
|
self.session.commit()
|
|
|
|
client = StatefulCalDAVClient()
|
|
client.put_object(href, payload, create=True)
|
|
operation = self.session.get(CalendarOutboxOperation, operation_id)
|
|
assert operation is not None
|
|
operation.lease_expires_at = utcnow() - timedelta(seconds=1)
|
|
self.session.commit()
|
|
|
|
result = dispatch_calendar_outbox(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
|
|
operation = self.session.get(CalendarOutboxOperation, operation_id)
|
|
assert operation is not None
|
|
self.assertEqual(result["succeeded"], 1)
|
|
self.assertEqual(operation.status, "succeeded")
|
|
self.assertIsNotNone(operation.reconciled_at)
|
|
self.assertEqual(len(client.resources), 1)
|
|
|
|
def test_transient_failure_retries_with_backoff_then_becomes_dead(self) -> None:
|
|
self.create_local_event()
|
|
self.session.commit()
|
|
operation = self.session.query(CalendarOutboxOperation).one()
|
|
operation.max_attempts = 2
|
|
self.session.commit()
|
|
client = StatefulCalDAVClient()
|
|
client.fail_before_write = True
|
|
client.fetch_fails = True
|
|
|
|
dispatch_calendar_outbox(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
operation = self.session.get(CalendarOutboxOperation, operation.id)
|
|
assert operation is not None
|
|
self.assertEqual(operation.status, "retry")
|
|
self.assertGreater(operation.available_at.replace(tzinfo=timezone.utc), utcnow())
|
|
operation.available_at = utcnow() - timedelta(seconds=1)
|
|
self.session.commit()
|
|
|
|
dispatch_calendar_outbox(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
operation = self.session.get(CalendarOutboxOperation, operation.id)
|
|
assert operation is not None
|
|
self.assertEqual(operation.status, "dead")
|
|
self.assertEqual(operation.attempt_count, 2)
|
|
|
|
def test_newer_state_waits_behind_predecessor_backoff(self) -> None:
|
|
event = self.create_local_event(summary="First")
|
|
self.session.commit()
|
|
client = StatefulCalDAVClient()
|
|
client.fail_before_write = True
|
|
client.fetch_fails = True
|
|
dispatch_calendar_outbox(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
first = self.session.query(CalendarOutboxOperation).one()
|
|
self.assertEqual(first.status, "retry")
|
|
|
|
update_event(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id=None,
|
|
event_id=event.id,
|
|
payload=CalendarEventUpdateRequest(summary="Second"),
|
|
)
|
|
self.session.commit()
|
|
second = (
|
|
self.session.query(CalendarOutboxOperation)
|
|
.filter(CalendarOutboxOperation.id != first.id)
|
|
.one()
|
|
)
|
|
client.fail_before_write = False
|
|
client.fetch_fails = False
|
|
|
|
blocked = dispatch_calendar_outbox(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
self.assertEqual(blocked["processed"], 0)
|
|
self.assertEqual(second.status, "pending")
|
|
|
|
first.available_at = utcnow() - timedelta(seconds=1)
|
|
self.session.commit()
|
|
delivered = dispatch_calendar_outbox(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
self.assertEqual(delivered["succeeded"], 2)
|
|
self.assertEqual(client.puts[-1]["etag"], '"etag-1"')
|
|
self.assertIn("SUMMARY:Second", client.puts[-1]["ics"])
|
|
|
|
def test_ambiguous_predecessor_is_reconciled_before_newer_state(self) -> None:
|
|
event = self.create_local_event(summary="First")
|
|
self.session.commit()
|
|
first = claim_next_calendar_outbox_operation(self.session, tenant_id="tenant-1")
|
|
assert first is not None
|
|
href = first.resource_href
|
|
client = StatefulCalDAVClient()
|
|
client.put_object(href, first.payload_ics or "", create=True)
|
|
first.lease_expires_at = utcnow() - timedelta(seconds=1)
|
|
self.session.commit()
|
|
|
|
update_event(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id=None,
|
|
event_id=event.id,
|
|
payload=CalendarEventUpdateRequest(summary="Second"),
|
|
)
|
|
self.session.commit()
|
|
second = (
|
|
self.session.query(CalendarOutboxOperation)
|
|
.filter(CalendarOutboxOperation.id != first.id)
|
|
.one()
|
|
)
|
|
|
|
result = dispatch_calendar_outbox(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
|
|
self.assertEqual(result["succeeded"], 2)
|
|
self.assertIsNotNone(first.reconciled_at)
|
|
self.assertEqual(second.expected_etag, '"etag-1"')
|
|
self.assertEqual(client.puts[-1]["etag"], '"etag-1"')
|
|
self.assertIn("SUMMARY:Second", client.puts[-1]["ics"])
|
|
|
|
def test_only_one_resource_per_source_can_hold_a_delivery_lease(self) -> None:
|
|
self.create_local_event(uid="event-1@example.test")
|
|
self.create_local_event(uid="event-2@example.test")
|
|
self.session.commit()
|
|
|
|
first = claim_next_calendar_outbox_operation(self.session, tenant_id="tenant-1")
|
|
assert first is not None
|
|
self.session.commit()
|
|
second = claim_next_calendar_outbox_operation(self.session, tenant_id="tenant-1")
|
|
|
|
self.assertIsNone(second)
|
|
self.assertEqual(
|
|
self.session.query(CalendarOutboxOperation).filter_by(status="in_progress").count(),
|
|
1,
|
|
)
|
|
|
|
def test_manual_retry_rejects_stale_resource_generation(self) -> None:
|
|
event = self.create_local_event(summary="First")
|
|
self.session.commit()
|
|
client = StatefulCalDAVClient()
|
|
client.fail_before_write = True
|
|
client.fetch_fails = True
|
|
dispatch_calendar_outbox(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
first = self.session.query(CalendarOutboxOperation).one()
|
|
self.assertEqual(first.status, "retry")
|
|
update_event(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id=None,
|
|
event_id=event.id,
|
|
payload=CalendarEventUpdateRequest(summary="Second"),
|
|
)
|
|
self.session.commit()
|
|
|
|
with self.assertRaisesRegex(ValueError, "newer desired state"):
|
|
retry_calendar_outbox_operation(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
operation_id=first.id,
|
|
)
|
|
with self.assertRaisesRegex(ValueError, "newer desired state"):
|
|
discard_calendar_outbox_operation(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
operation_id=first.id,
|
|
)
|
|
|
|
def test_discard_latest_cancels_attempted_predecessor_chain(self) -> None:
|
|
event = self.create_local_event(summary="First")
|
|
self.session.commit()
|
|
client = StatefulCalDAVClient()
|
|
client.fail_before_write = True
|
|
client.fetch_fails = True
|
|
dispatch_calendar_outbox(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
first = self.session.query(CalendarOutboxOperation).one()
|
|
update_event(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id=None,
|
|
event_id=event.id,
|
|
payload=CalendarEventUpdateRequest(summary="Second"),
|
|
)
|
|
self.session.commit()
|
|
latest = (
|
|
self.session.query(CalendarOutboxOperation)
|
|
.filter(CalendarOutboxOperation.id != first.id)
|
|
.one()
|
|
)
|
|
|
|
discard_calendar_outbox_operation(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
operation_id=latest.id,
|
|
)
|
|
self.session.commit()
|
|
self.assertEqual(first.status, "cancelled")
|
|
self.assertEqual(latest.status, "cancelled")
|
|
self.assertEqual((event.metadata_ or {})["caldav"]["external_state"], "discarded")
|
|
client.fail_before_write = False
|
|
client.fetch_fails = False
|
|
result = dispatch_calendar_outbox(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
self.assertEqual(result["processed"], 0)
|
|
|
|
def test_client_or_credential_setup_failure_is_persisted_for_retry(self) -> None:
|
|
self.create_local_event()
|
|
self.session.commit()
|
|
|
|
def unavailable_client(_session, _source):
|
|
raise CalendarError("credential temporarily unavailable")
|
|
|
|
result = dispatch_calendar_outbox(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
client_factory=unavailable_client,
|
|
)
|
|
operation = self.session.query(CalendarOutboxOperation).one()
|
|
|
|
self.assertEqual(result["retrying"], 1)
|
|
self.assertEqual(operation.status, "retry")
|
|
self.assertIn("credential temporarily unavailable", operation.last_error or "")
|
|
|
|
def test_newer_update_uses_etag_from_older_in_progress_put(self) -> None:
|
|
event = self.create_local_event(summary="First")
|
|
self.session.commit()
|
|
first = claim_next_calendar_outbox_operation(self.session, tenant_id="tenant-1")
|
|
assert first is not None
|
|
first_id = first.id
|
|
first_lease = first.lease_token or ""
|
|
self.session.commit()
|
|
|
|
update_event(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id=None,
|
|
event_id=event.id,
|
|
payload=CalendarEventUpdateRequest(summary="Second"),
|
|
)
|
|
self.session.commit()
|
|
second = (
|
|
self.session.query(CalendarOutboxOperation)
|
|
.filter(CalendarOutboxOperation.id != first_id)
|
|
.one()
|
|
)
|
|
self.assertIsNone(second.expected_etag)
|
|
client = StatefulCalDAVClient()
|
|
|
|
execute_calendar_outbox_operation(
|
|
self.session,
|
|
operation_id=first_id,
|
|
lease_token=first_lease,
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
self.session.commit()
|
|
self.session.refresh(second)
|
|
self.assertEqual(second.expected_etag, '"etag-1"')
|
|
self.assertEqual(event.etag, '"etag-1"')
|
|
self.assertEqual((event.metadata_ or {})["caldav"]["outbox_operation_id"], second.id)
|
|
|
|
dispatch_calendar_outbox(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
self.assertEqual(client.puts[-1]["etag"], '"etag-1"')
|
|
self.assertIn("SUMMARY:Second", client.puts[-1]["ics"])
|
|
|
|
def test_put_without_response_etag_fetches_it_before_succeeding(self) -> None:
|
|
event = self.create_local_event(summary="First")
|
|
self.session.commit()
|
|
client = StatefulCalDAVClient()
|
|
client.omit_put_etag = True
|
|
|
|
result = dispatch_calendar_outbox(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
first = self.session.query(CalendarOutboxOperation).one()
|
|
self.assertEqual(result["succeeded"], 1)
|
|
self.assertEqual(first.remote_etag, '"etag-1"')
|
|
self.assertIsNotNone(first.reconciled_at)
|
|
self.assertEqual(event.etag, '"etag-1"')
|
|
|
|
update_event(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id=None,
|
|
event_id=event.id,
|
|
payload=CalendarEventUpdateRequest(summary="Second"),
|
|
)
|
|
self.session.commit()
|
|
dispatch_calendar_outbox(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
self.assertEqual(client.puts[-1]["etag"], '"etag-1"')
|
|
|
|
def test_delete_without_known_etag_refuses_unknown_remote_version(self) -> None:
|
|
event = self.create_local_event()
|
|
self.session.commit()
|
|
href = event.source_href or ""
|
|
delete_event(self.session, tenant_id="tenant-1", event_id=event.id)
|
|
self.session.commit()
|
|
delete_operation = (
|
|
self.session.query(CalendarOutboxOperation)
|
|
.filter_by(operation_kind="delete")
|
|
.one()
|
|
)
|
|
self.assertIsNone(delete_operation.expected_etag)
|
|
client = StatefulCalDAVClient()
|
|
client.resources[href] = (different_event_ics(), '"unknown"')
|
|
|
|
result = dispatch_calendar_outbox(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
|
|
self.assertEqual(result["failed"], 1)
|
|
self.assertEqual(delete_operation.status, "conflict")
|
|
self.assertEqual(client.deletes, [])
|
|
self.assertIn("no known ETag", delete_operation.last_error or "")
|
|
|
|
def test_newer_delete_follows_in_progress_update_without_resurrecting_event(self) -> None:
|
|
event = self.create_local_event()
|
|
self.session.commit()
|
|
client = StatefulCalDAVClient()
|
|
dispatch_calendar_outbox(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
update_event(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id=None,
|
|
event_id=event.id,
|
|
payload=CalendarEventUpdateRequest(summary="In flight"),
|
|
)
|
|
self.session.commit()
|
|
update_operation = claim_next_calendar_outbox_operation(self.session, tenant_id="tenant-1")
|
|
assert update_operation is not None
|
|
update_id = update_operation.id
|
|
update_lease = update_operation.lease_token or ""
|
|
self.session.commit()
|
|
delete_event(self.session, tenant_id="tenant-1", event_id=event.id)
|
|
self.session.commit()
|
|
|
|
execute_calendar_outbox_operation(
|
|
self.session,
|
|
operation_id=update_id,
|
|
lease_token=update_lease,
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
self.session.commit()
|
|
delete_operation = (
|
|
self.session.query(CalendarOutboxOperation)
|
|
.filter(CalendarOutboxOperation.operation_kind == "delete")
|
|
.one()
|
|
)
|
|
self.assertEqual(delete_operation.expected_etag, '"etag-2"')
|
|
dispatch_calendar_outbox(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
|
|
self.assertEqual(client.resources, {})
|
|
self.assertEqual(delete_operation.status, "succeeded")
|
|
self.assertIsNotNone(event.deleted_at)
|
|
|
|
def test_source_direction_change_rejects_unresolved_but_policy_and_credentials_do_not(self) -> None:
|
|
self.create_local_event()
|
|
self.session.commit()
|
|
operation = self.session.query(CalendarOutboxOperation).one()
|
|
update_caldav_source(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
source_id=self.source.id,
|
|
payload=CalendarCalDavSourceUpdateRequest(
|
|
conflict_policy="overwrite",
|
|
username="rotated-user",
|
|
),
|
|
)
|
|
self.session.commit()
|
|
self.assertEqual(operation.status, "pending")
|
|
client = StatefulCalDAVClient()
|
|
dispatch_calendar_outbox(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
self.assertFalse(client.puts[0]["overwrite"])
|
|
|
|
update_event(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id=None,
|
|
event_id=operation.event_id or "",
|
|
payload=CalendarEventUpdateRequest(summary="Pending again"),
|
|
)
|
|
self.session.commit()
|
|
pending = (
|
|
self.session.query(CalendarOutboxOperation)
|
|
.filter(CalendarOutboxOperation.status == "pending")
|
|
.one()
|
|
)
|
|
with self.assertRaisesRegex(CalendarError, "unresolved local desired"):
|
|
update_caldav_source(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
source_id=self.source.id,
|
|
payload=CalendarCalDavSourceUpdateRequest(sync_direction="inbound"),
|
|
)
|
|
self.session.rollback()
|
|
self.assertEqual(pending.status, "pending")
|
|
self.assertEqual(self.source.sync_direction, "two_way")
|
|
|
|
def test_disabled_source_rejects_local_edits_instead_of_losing_them(self) -> None:
|
|
event = self.create_local_event()
|
|
self.session.commit()
|
|
client = StatefulCalDAVClient()
|
|
dispatch_calendar_outbox(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
update_caldav_source(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
source_id=self.source.id,
|
|
payload=CalendarCalDavSourceUpdateRequest(sync_enabled=False),
|
|
)
|
|
self.session.commit()
|
|
|
|
with self.assertRaisesRegex(CalendarError, "disabled"):
|
|
update_event(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id=None,
|
|
event_id=event.id,
|
|
payload=CalendarEventUpdateRequest(summary="Would be lost"),
|
|
)
|
|
self.session.rollback()
|
|
self.assertEqual(event.summary, "Planning")
|
|
self.assertEqual(
|
|
self.session.query(CalendarOutboxOperation).filter_by(status="pending").count(),
|
|
0,
|
|
)
|
|
|
|
def test_source_endpoint_change_is_rejected_while_events_reference_it(self) -> None:
|
|
self.create_local_event()
|
|
self.session.commit()
|
|
client = StatefulCalDAVClient()
|
|
dispatch_calendar_outbox(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
|
|
with self.assertRaisesRegex(CalendarError, "events still reference"):
|
|
update_caldav_source(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
source_id=self.source.id,
|
|
payload=CalendarCalDavSourceUpdateRequest(
|
|
collection_url="https://dav.example.test/new-calendar",
|
|
),
|
|
)
|
|
self.session.rollback()
|
|
self.assertEqual(self.source.collection_url, "https://dav.example.test/cal/")
|
|
|
|
def test_calendar_rejects_a_second_active_sync_source(self) -> None:
|
|
with self.assertRaisesRegex(CalendarError, "only one active"):
|
|
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/another",
|
|
),
|
|
)
|
|
|
|
def test_soft_deleted_events_do_not_permanently_block_source_endpoint_change(self) -> None:
|
|
event = self.create_local_event()
|
|
self.session.commit()
|
|
client = StatefulCalDAVClient()
|
|
dispatch_calendar_outbox(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
delete_event(self.session, tenant_id="tenant-1", event_id=event.id)
|
|
self.session.commit()
|
|
dispatch_calendar_outbox(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
|
|
update_caldav_source(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
source_id=self.source.id,
|
|
payload=CalendarCalDavSourceUpdateRequest(
|
|
collection_url="https://dav.example.test/new-calendar",
|
|
),
|
|
)
|
|
|
|
self.assertEqual(
|
|
self.source.collection_url,
|
|
"https://dav.example.test/new-calendar/",
|
|
)
|
|
|
|
def test_bulk_move_requires_the_matching_explicit_external_action(self) -> None:
|
|
event = self.create_local_event()
|
|
local_calendar = create_calendar(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id=None,
|
|
payload=CalendarCollectionCreateRequest(name="Local"),
|
|
)
|
|
self.session.commit()
|
|
|
|
with self.assertRaisesRegex(CalendarError, "detach_keep_remote"):
|
|
delete_calendar(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
calendar_id=self.calendar.id,
|
|
payload=CalendarCollectionDeleteRequest(
|
|
event_action="move",
|
|
target_calendar_id=local_calendar.id,
|
|
),
|
|
)
|
|
self.session.rollback()
|
|
|
|
self.assertEqual(event.calendar_id, self.calendar.id)
|
|
self.assertIsNone(self.calendar.deleted_at)
|
|
self.assertIsNone(self.source.deleted_at)
|
|
|
|
another_local_calendar = create_calendar(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id=None,
|
|
payload=CalendarCollectionCreateRequest(name="Another local"),
|
|
)
|
|
create_event(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id=None,
|
|
payload=CalendarEventCreateRequest(
|
|
calendar_id=another_local_calendar.id,
|
|
uid="local-event@example.test",
|
|
summary="Local event",
|
|
start_at=datetime(2026, 7, 9, 9, 0, tzinfo=timezone.utc),
|
|
),
|
|
)
|
|
self.session.commit()
|
|
with self.assertRaisesRegex(CalendarError, "copy_to_remote"):
|
|
delete_calendar(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
calendar_id=another_local_calendar.id,
|
|
payload=CalendarCollectionDeleteRequest(
|
|
event_action="move",
|
|
target_calendar_id=self.calendar.id,
|
|
),
|
|
)
|
|
self.session.rollback()
|
|
self.assertIsNone(another_local_calendar.deleted_at)
|
|
|
|
with self.assertRaisesRegex(CalendarError, "only valid"):
|
|
delete_calendar(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
calendar_id=another_local_calendar.id,
|
|
payload=CalendarCollectionDeleteRequest(
|
|
event_action="move",
|
|
target_calendar_id=local_calendar.id,
|
|
external_action="detach_keep_remote",
|
|
),
|
|
)
|
|
self.session.rollback()
|
|
|
|
with self.assertRaisesRegex(CalendarError, "remote_move.*not supported"):
|
|
delete_calendar(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
calendar_id=self.calendar.id,
|
|
payload=CalendarCollectionDeleteRequest(
|
|
event_action="move",
|
|
target_calendar_id=local_calendar.id,
|
|
external_action="remote_move",
|
|
),
|
|
)
|
|
self.session.rollback()
|
|
|
|
def test_detach_move_cancels_pending_state_and_never_deletes_remote_events(self) -> None:
|
|
delivered_event = self.create_local_event(uid="delivered@example.test")
|
|
self.session.commit()
|
|
client = StatefulCalDAVClient()
|
|
dispatch_calendar_outbox(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
delivered_href = delivered_event.source_href or ""
|
|
self.assertIn(delivered_href, client.resources)
|
|
|
|
pending_event = self.create_local_event(uid="pending@example.test")
|
|
pending_operation_id = (pending_event.metadata_ or {})["caldav"]["outbox_operation_id"]
|
|
local_calendar = create_calendar(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id=None,
|
|
payload=CalendarCollectionCreateRequest(name="Detached"),
|
|
)
|
|
self.session.commit()
|
|
|
|
delete_calendar(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
calendar_id=self.calendar.id,
|
|
payload=CalendarCollectionDeleteRequest(
|
|
event_action="move",
|
|
target_calendar_id=local_calendar.id,
|
|
external_action="detach_keep_remote",
|
|
),
|
|
)
|
|
|
|
pending_operation = self.session.get(CalendarOutboxOperation, pending_operation_id)
|
|
assert pending_operation is not None
|
|
self.assertEqual(pending_operation.status, "cancelled")
|
|
self.assertTrue(all(item.operation_kind != "delete" for item in self.session.query(CalendarOutboxOperation).all()))
|
|
for event in (delivered_event, pending_event):
|
|
self.assertEqual(event.calendar_id, local_calendar.id)
|
|
self.assertEqual(event.source_kind, "local")
|
|
self.assertIsNone(event.source_href)
|
|
self.assertIsNone(event.etag)
|
|
self.assertNotIn("caldav", event.metadata_ or {})
|
|
self.assertIsNone(event.deleted_at)
|
|
self.assertIsNotNone(self.calendar.deleted_at)
|
|
self.assertIsNotNone(self.source.deleted_at)
|
|
self.session.commit()
|
|
|
|
result = dispatch_calendar_outbox(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
self.assertEqual(result["processed"], 0)
|
|
self.assertEqual(client.deletes, [])
|
|
self.assertIn(delivered_href, client.resources)
|
|
|
|
def test_detach_move_is_rejected_while_source_has_a_live_delivery_lease(self) -> None:
|
|
event = self.create_local_event(uid="leased@example.test")
|
|
local_calendar = create_calendar(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id=None,
|
|
payload=CalendarCollectionCreateRequest(name="Local"),
|
|
)
|
|
self.session.commit()
|
|
operation = claim_next_calendar_outbox_operation(self.session, tenant_id="tenant-1")
|
|
assert operation is not None
|
|
self.session.commit()
|
|
|
|
with self.assertRaisesRegex(CalendarError, "active lease"):
|
|
delete_calendar(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
calendar_id=self.calendar.id,
|
|
payload=CalendarCollectionDeleteRequest(
|
|
event_action="move",
|
|
target_calendar_id=local_calendar.id,
|
|
external_action="detach_keep_remote",
|
|
),
|
|
)
|
|
self.session.rollback()
|
|
|
|
self.assertEqual(event.calendar_id, self.calendar.id)
|
|
self.assertEqual(operation.status, "in_progress")
|
|
self.assertIsNone(self.calendar.deleted_at)
|
|
self.assertIsNone(self.source.deleted_at)
|
|
|
|
def test_copy_move_queues_destination_put_and_keeps_failure_retryable(self) -> None:
|
|
local_calendar = create_calendar(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id=None,
|
|
payload=CalendarCollectionCreateRequest(name="Local"),
|
|
)
|
|
event = create_event(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id=None,
|
|
payload=CalendarEventCreateRequest(
|
|
calendar_id=local_calendar.id,
|
|
uid="copy@example.test",
|
|
summary="Copy through durable outbox",
|
|
start_at=datetime(2026, 7, 9, 9, 0, tzinfo=timezone.utc),
|
|
),
|
|
)
|
|
self.session.commit()
|
|
|
|
delete_calendar(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
calendar_id=local_calendar.id,
|
|
payload=CalendarCollectionDeleteRequest(
|
|
event_action="move",
|
|
target_calendar_id=self.calendar.id,
|
|
external_action="copy_to_remote",
|
|
),
|
|
)
|
|
|
|
operation_id = (event.metadata_ or {})["caldav"]["outbox_operation_id"]
|
|
operation = self.session.get(CalendarOutboxOperation, operation_id)
|
|
assert operation is not None
|
|
self.assertEqual(operation.operation_kind, "put")
|
|
self.assertEqual(operation.status, "pending")
|
|
self.assertEqual(event.calendar_id, self.calendar.id)
|
|
self.assertEqual(event.source_kind, "caldav")
|
|
self.assertIsNone(event.deleted_at)
|
|
self.assertIsNotNone(local_calendar.deleted_at)
|
|
move_change = (
|
|
self.session.query(ChangeSequenceEntry)
|
|
.filter_by(resource_id=event.id)
|
|
.order_by(ChangeSequenceEntry.id.desc())
|
|
.first()
|
|
)
|
|
assert move_change is not None
|
|
self.assertEqual(move_change.payload["previous_calendar_id"], local_calendar.id)
|
|
self.assertEqual(move_change.payload["calendar_id"], self.calendar.id)
|
|
self.assertEqual(move_change.payload["external_state"], "queued")
|
|
self.session.commit()
|
|
|
|
client = StatefulCalDAVClient()
|
|
client.fail_before_write = True
|
|
result = dispatch_calendar_outbox(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
self.assertEqual(result["retrying"], 1)
|
|
self.assertEqual(operation.status, "retry")
|
|
self.assertEqual((event.metadata_ or {})["caldav"]["external_state"], "retry")
|
|
self.assertEqual(client.deletes, [])
|
|
|
|
retry_calendar_outbox_operation(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
operation_id=operation.id,
|
|
)
|
|
self.session.commit()
|
|
client.fail_before_write = False
|
|
result = dispatch_calendar_outbox(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
self.assertEqual(result["succeeded"], 1)
|
|
self.assertEqual(operation.status, "succeeded")
|
|
self.assertIn(event.source_href or "", client.resources)
|
|
self.assertEqual(client.deletes, [])
|
|
|
|
def test_copy_move_rejects_an_inbound_only_target(self) -> None:
|
|
local_calendar = create_calendar(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id=None,
|
|
payload=CalendarCollectionCreateRequest(name="Local"),
|
|
)
|
|
self.source.sync_direction = "inbound"
|
|
self.session.commit()
|
|
|
|
with self.assertRaisesRegex(CalendarError, "active two-way CalDAV target"):
|
|
delete_calendar(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
calendar_id=local_calendar.id,
|
|
payload=CalendarCollectionDeleteRequest(
|
|
event_action="move",
|
|
target_calendar_id=self.calendar.id,
|
|
external_action="copy_to_remote",
|
|
),
|
|
)
|
|
self.session.rollback()
|
|
self.assertIsNone(local_calendar.deleted_at)
|
|
|
|
def test_local_to_local_move_remains_compatible_without_external_action(self) -> None:
|
|
source_calendar = create_calendar(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id=None,
|
|
payload=CalendarCollectionCreateRequest(name="Local source"),
|
|
)
|
|
target_calendar = create_calendar(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id=None,
|
|
payload=CalendarCollectionCreateRequest(name="Local target"),
|
|
)
|
|
event = create_event(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id=None,
|
|
payload=CalendarEventCreateRequest(
|
|
calendar_id=source_calendar.id,
|
|
uid="local-move@example.test",
|
|
summary="Local move",
|
|
start_at=datetime(2026, 7, 9, 9, 0, tzinfo=timezone.utc),
|
|
),
|
|
)
|
|
self.session.commit()
|
|
|
|
delete_calendar(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
calendar_id=source_calendar.id,
|
|
payload=CalendarCollectionDeleteRequest(
|
|
event_action="move",
|
|
target_calendar_id=target_calendar.id,
|
|
),
|
|
)
|
|
|
|
self.assertEqual(event.calendar_id, target_calendar.id)
|
|
self.assertEqual(event.source_kind, "local")
|
|
self.assertIsNone(event.deleted_at)
|
|
self.assertIsNotNone(source_calendar.deleted_at)
|
|
self.assertEqual(self.session.query(CalendarOutboxOperation).count(), 0)
|
|
|
|
def test_move_between_synchronized_calendars_remains_rejected(self) -> None:
|
|
other_calendar = create_calendar(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id=None,
|
|
payload=CalendarCollectionCreateRequest(name="Other remote"),
|
|
)
|
|
create_caldav_source(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id=None,
|
|
payload=CalendarCalDavSourceCreateRequest(
|
|
calendar_id=other_calendar.id,
|
|
collection_url="https://dav.example.test/other",
|
|
),
|
|
)
|
|
self.session.commit()
|
|
|
|
with self.assertRaisesRegex(CalendarError, "between synchronized calendars"):
|
|
delete_calendar(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
calendar_id=self.calendar.id,
|
|
payload=CalendarCollectionDeleteRequest(
|
|
event_action="move",
|
|
target_calendar_id=other_calendar.id,
|
|
external_action="detach_keep_remote",
|
|
),
|
|
)
|
|
self.session.rollback()
|
|
self.assertIsNone(self.calendar.deleted_at)
|
|
self.assertIsNone(other_calendar.deleted_at)
|
|
|
|
def test_deleting_synced_collection_is_local_unlink_and_preserves_remote(self) -> None:
|
|
event = self.create_local_event(uid="preserve-on-unlink@example.test")
|
|
self.session.commit()
|
|
client = StatefulCalDAVClient()
|
|
dispatch_calendar_outbox(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
href = event.source_href or ""
|
|
self.assertIn(href, client.resources)
|
|
|
|
delete_calendar(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
calendar_id=self.calendar.id,
|
|
)
|
|
self.session.commit()
|
|
result = dispatch_calendar_outbox(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
|
|
self.assertEqual(result["processed"], 0)
|
|
self.assertEqual(client.deletes, [])
|
|
self.assertIn(href, client.resources)
|
|
self.assertIsNotNone(event.deleted_at)
|
|
self.assertIsNotNone(self.source.deleted_at)
|
|
|
|
def test_material_source_change_is_rejected_while_delivery_lease_is_live(self) -> None:
|
|
self.create_local_event()
|
|
self.session.commit()
|
|
operation = claim_next_calendar_outbox_operation(self.session, tenant_id="tenant-1")
|
|
assert operation is not None
|
|
self.session.commit()
|
|
|
|
with self.assertRaisesRegex(CalendarError, "active lease"):
|
|
update_caldav_source(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
source_id=self.source.id,
|
|
payload=CalendarCalDavSourceUpdateRequest(sync_direction="inbound"),
|
|
)
|
|
self.session.rollback()
|
|
|
|
self.assertEqual(self.source.sync_direction, "two_way")
|
|
self.assertEqual(operation.status, "in_progress")
|
|
|
|
def test_source_and_calendar_retirement_are_rejected_while_lease_is_live(self) -> None:
|
|
self.create_local_event()
|
|
self.session.commit()
|
|
operation = claim_next_calendar_outbox_operation(self.session, tenant_id="tenant-1")
|
|
assert operation is not None
|
|
self.session.commit()
|
|
|
|
with self.assertRaisesRegex(CalendarError, "cannot be retired.*active lease"):
|
|
delete_caldav_source(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
source_id=self.source.id,
|
|
)
|
|
self.session.rollback()
|
|
with self.assertRaisesRegex(CalendarError, "cannot be retired.*active lease"):
|
|
delete_calendar(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
calendar_id=self.calendar.id,
|
|
)
|
|
self.session.rollback()
|
|
|
|
self.assertIsNone(self.source.deleted_at)
|
|
self.assertIsNone(self.calendar.deleted_at)
|
|
self.assertEqual(operation.status, "in_progress")
|
|
|
|
def test_tenant_dispatch_does_not_reclaim_another_tenants_expired_lease(self) -> None:
|
|
self.create_local_event()
|
|
self.session.add(Tenant(id="tenant-2", slug="tenant-2", name="Tenant 2"))
|
|
calendar_2 = create_calendar(
|
|
self.session,
|
|
tenant_id="tenant-2",
|
|
user_id=None,
|
|
payload=CalendarCollectionCreateRequest(name="Remote 2"),
|
|
)
|
|
create_caldav_source(
|
|
self.session,
|
|
tenant_id="tenant-2",
|
|
user_id=None,
|
|
payload=CalendarCalDavSourceCreateRequest(
|
|
calendar_id=calendar_2.id,
|
|
collection_url="https://dav.example.test/tenant-2",
|
|
),
|
|
)
|
|
create_event(
|
|
self.session,
|
|
tenant_id="tenant-2",
|
|
user_id=None,
|
|
payload=CalendarEventCreateRequest(
|
|
calendar_id=calendar_2.id,
|
|
uid="event-2@example.test",
|
|
summary="Tenant 2",
|
|
start_at=datetime(2026, 7, 8, 11, 0, tzinfo=timezone.utc),
|
|
),
|
|
)
|
|
self.session.commit()
|
|
tenant_2_operation = claim_next_calendar_outbox_operation(
|
|
self.session,
|
|
tenant_id="tenant-2",
|
|
)
|
|
assert tenant_2_operation is not None
|
|
tenant_2_operation.lease_expires_at = utcnow() - timedelta(seconds=1)
|
|
self.session.commit()
|
|
|
|
tenant_1_operation = claim_next_calendar_outbox_operation(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
)
|
|
|
|
self.assertIsNotNone(tenant_1_operation)
|
|
self.assertEqual(tenant_2_operation.status, "in_progress")
|
|
|
|
def test_corrupt_cross_tenant_source_reference_is_quarantined_without_network(self) -> None:
|
|
self.session.add(Tenant(id="tenant-2", slug="tenant-2", name="Tenant 2"))
|
|
calendar_2 = create_calendar(
|
|
self.session,
|
|
tenant_id="tenant-2",
|
|
user_id=None,
|
|
payload=CalendarCollectionCreateRequest(name="Remote 2"),
|
|
)
|
|
source_2 = create_caldav_source(
|
|
self.session,
|
|
tenant_id="tenant-2",
|
|
user_id=None,
|
|
payload=CalendarCalDavSourceCreateRequest(
|
|
calendar_id=calendar_2.id,
|
|
collection_url="https://dav.example.test/tenant-2",
|
|
),
|
|
)
|
|
create_event(
|
|
self.session,
|
|
tenant_id="tenant-2",
|
|
user_id=None,
|
|
payload=CalendarEventCreateRequest(
|
|
calendar_id=calendar_2.id,
|
|
uid="event-2@example.test",
|
|
summary="Tenant 2",
|
|
start_at=datetime(2026, 7, 8, 11, 0, tzinfo=timezone.utc),
|
|
),
|
|
)
|
|
self.session.commit()
|
|
operation = claim_next_calendar_outbox_operation(self.session, tenant_id="tenant-2")
|
|
assert operation is not None
|
|
lease_token = operation.lease_token or ""
|
|
operation.source_id = self.source.id
|
|
self.session.commit()
|
|
client = StatefulCalDAVClient()
|
|
|
|
execute_calendar_outbox_operation(
|
|
self.session,
|
|
operation_id=operation.id,
|
|
lease_token=lease_token,
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
self.session.commit()
|
|
|
|
self.assertEqual(operation.status, "cancelled")
|
|
self.assertIn("tenant does not match", operation.last_error or "")
|
|
self.assertEqual(client.puts, [])
|
|
self.assertNotEqual(source_2.id, self.source.id)
|
|
|
|
def test_manual_reconcile_respects_live_lease_and_marks_mismatch_conflict(self) -> None:
|
|
event = self.create_local_event()
|
|
self.session.commit()
|
|
operation = claim_next_calendar_outbox_operation(self.session, tenant_id="tenant-1")
|
|
assert operation is not None
|
|
self.session.commit()
|
|
client = StatefulCalDAVClient()
|
|
|
|
with self.assertRaisesRegex(ValueError, "actively leased"):
|
|
reconcile_calendar_outbox_operation(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
operation_id=operation.id,
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
|
|
operation.lease_expires_at = utcnow() - timedelta(seconds=1)
|
|
operation.status = "retry"
|
|
operation.lease_token = None
|
|
self.session.commit()
|
|
client.resources[operation.resource_href] = (different_event_ics(), '"remote"')
|
|
reconcile_calendar_outbox_operation(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
operation_id=operation.id,
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
self.session.commit()
|
|
|
|
self.assertEqual(operation.status, "conflict")
|
|
self.assertEqual(self.source.last_status, "outbound_error")
|
|
self.assertEqual((event.metadata_ or {})["caldav"]["external_state"], "failed")
|
|
with self.assertRaisesRegex(ValueError, "succeeded"):
|
|
operation.status = "succeeded"
|
|
self.session.flush()
|
|
retry_calendar_outbox_operation(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
operation_id=operation.id,
|
|
)
|
|
|
|
def test_manual_reconcile_live_lease_is_timezone_safe_in_fresh_session(self) -> None:
|
|
self.create_local_event()
|
|
self.session.commit()
|
|
operation = claim_next_calendar_outbox_operation(self.session, tenant_id="tenant-1")
|
|
assert operation is not None
|
|
operation_id = operation.id
|
|
self.session.commit()
|
|
self.session.close()
|
|
self.session = self.Session()
|
|
|
|
with self.assertRaisesRegex(ValueError, "actively leased"):
|
|
reconcile_calendar_outbox_operation(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
operation_id=operation_id,
|
|
client_factory=lambda _session, _source: StatefulCalDAVClient(),
|
|
)
|
|
|
|
def test_delete_retry_is_projected_on_deleted_trigger_event(self) -> None:
|
|
event = self.create_local_event()
|
|
self.session.commit()
|
|
client = StatefulCalDAVClient()
|
|
dispatch_calendar_outbox(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
delete_event(self.session, tenant_id="tenant-1", event_id=event.id)
|
|
self.session.commit()
|
|
client.fail_before_write = True
|
|
client.fetch_fails = True
|
|
|
|
result = dispatch_calendar_outbox(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
|
|
self.assertEqual(result["retrying"], 1)
|
|
self.assertIsNotNone(event.deleted_at)
|
|
self.assertEqual((event.metadata_ or {})["caldav"]["external_state"], "retry")
|
|
|
|
def test_old_source_delete_does_not_rebind_event_moved_to_local_calendar(self) -> None:
|
|
event = self.create_local_event()
|
|
local_calendar = create_calendar(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id=None,
|
|
payload=CalendarCollectionCreateRequest(name="Local"),
|
|
)
|
|
self.session.commit()
|
|
client = StatefulCalDAVClient()
|
|
dispatch_calendar_outbox(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
|
|
update_event(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id=None,
|
|
event_id=event.id,
|
|
payload=CalendarEventUpdateRequest(calendar_id=local_calendar.id),
|
|
)
|
|
self.session.commit()
|
|
dispatch_calendar_outbox(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
|
|
self.assertEqual(event.calendar_id, local_calendar.id)
|
|
self.assertEqual(event.source_kind, "local")
|
|
self.assertIsNone(event.source_href)
|
|
self.assertIsNone(event.etag)
|
|
|
|
def test_resource_href_must_remain_in_configured_collection(self) -> None:
|
|
for href in (
|
|
"https://evil.example.test/steal.ics",
|
|
"https://dav.example.test/other/steal.ics",
|
|
):
|
|
with self.subTest(href=href), self.assertRaisesRegex(
|
|
CalendarError,
|
|
"origin|collection path",
|
|
):
|
|
normalize_caldav_href(self.source.collection_url, href)
|
|
|
|
self.assertEqual(
|
|
normalize_caldav_href(self.source.collection_url, "/cal/legit.ics"),
|
|
"https://dav.example.test/cal/legit.ics",
|
|
)
|
|
with self.assertRaisesRegex(CalendarError, "sync-owned"):
|
|
self.create_local_event(source_href="/cal/legit.ics")
|
|
self.session.rollback()
|
|
self.assertEqual(self.session.query(CalendarOutboxOperation).count(), 0)
|
|
|
|
def test_public_update_cannot_retarget_sync_owned_resource_fields(self) -> None:
|
|
event = self.create_local_event()
|
|
self.session.commit()
|
|
|
|
with self.assertRaisesRegex(CalendarError, "sync-owned"):
|
|
update_event(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id=None,
|
|
event_id=event.id,
|
|
payload=CalendarEventUpdateRequest(
|
|
source_href="https://dav.example.test/cal/other.ics",
|
|
),
|
|
)
|
|
self.session.rollback()
|
|
self.assertEqual(
|
|
self.session.query(CalendarOutboxOperation).filter_by(status="pending").count(),
|
|
1,
|
|
)
|
|
|
|
def test_public_ics_import_create_and_upsert_are_local_desired_mutations(self) -> None:
|
|
created = import_ics_event(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id=None,
|
|
calendar_id=self.calendar.id,
|
|
ics=imported_event_ics("Imported first"),
|
|
upsert=True,
|
|
source_kind="ical",
|
|
source_href="https://untrusted.example.test/chosen.ics",
|
|
etag='"untrusted"',
|
|
)
|
|
self.session.commit()
|
|
first_operation = self.session.query(CalendarOutboxOperation).one()
|
|
|
|
self.assertEqual(created.source_kind, "caldav")
|
|
self.assertTrue((created.source_href or "").startswith(self.source.collection_url))
|
|
self.assertNotEqual(
|
|
created.source_href,
|
|
"https://untrusted.example.test/chosen.ics",
|
|
)
|
|
self.assertIsNone(created.etag)
|
|
self.assertEqual(first_operation.status, "pending")
|
|
self.assertIn("SUMMARY:Imported first", first_operation.payload_ics or "")
|
|
|
|
updated = import_ics_event(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id=None,
|
|
calendar_id=self.calendar.id,
|
|
ics=imported_event_ics("Imported updated"),
|
|
upsert=True,
|
|
source_kind="ical",
|
|
source_href="https://untrusted.example.test/other.ics",
|
|
etag='"other-untrusted"',
|
|
)
|
|
self.session.commit()
|
|
latest_operation = (
|
|
self.session.query(CalendarOutboxOperation)
|
|
.order_by(CalendarOutboxOperation.created_at.desc())
|
|
.first()
|
|
)
|
|
|
|
assert latest_operation is not None
|
|
self.assertEqual(updated.id, created.id)
|
|
self.assertEqual(updated.summary, "Imported updated")
|
|
self.assertEqual(updated.source_href, created.source_href)
|
|
self.assertEqual(first_operation.status, "superseded")
|
|
self.assertEqual(latest_operation.status, "pending")
|
|
self.assertIn("SUMMARY:Imported updated", latest_operation.payload_ics or "")
|
|
|
|
def test_outbox_projection_emits_followup_calendar_delta(self) -> None:
|
|
event = self.create_local_event()
|
|
self.session.commit()
|
|
initial_change = (
|
|
self.session.query(ChangeSequenceEntry)
|
|
.filter_by(resource_id=event.id)
|
|
.order_by(ChangeSequenceEntry.id.desc())
|
|
.first()
|
|
)
|
|
assert initial_change is not None
|
|
self.assertEqual(initial_change.payload["external_state"], "queued")
|
|
client = StatefulCalDAVClient()
|
|
|
|
dispatch_calendar_outbox(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
projected_change = (
|
|
self.session.query(ChangeSequenceEntry)
|
|
.filter(
|
|
ChangeSequenceEntry.resource_id == event.id,
|
|
ChangeSequenceEntry.id > initial_change.id,
|
|
)
|
|
.order_by(ChangeSequenceEntry.id.desc())
|
|
.first()
|
|
)
|
|
|
|
assert projected_change is not None
|
|
self.assertEqual(projected_change.operation, "updated")
|
|
self.assertEqual(projected_change.actor_type, "system")
|
|
self.assertEqual(projected_change.payload["previous_external_state"], "queued")
|
|
self.assertEqual(projected_change.payload["external_state"], "synced")
|
|
|
|
def test_inbound_sync_does_not_overwrite_active_local_desired_state(self) -> None:
|
|
event = self.create_local_event(summary="Local desired")
|
|
self.session.commit()
|
|
report = CalDAVReportResult(
|
|
objects=[
|
|
CalDAVObject(
|
|
href=event.source_href or "",
|
|
etag='"remote"',
|
|
calendar_data=different_event_ics(),
|
|
)
|
|
]
|
|
)
|
|
stats = type("Stats", (), {"created": 0, "updated": 0, "deleted": 0, "unchanged": 0, "fetched": 0, "full_sync": False})()
|
|
apply_caldav_report(
|
|
self.session,
|
|
source=self.source,
|
|
client=StatefulCalDAVClient(),
|
|
report=report,
|
|
stats=stats,
|
|
user_id=None,
|
|
)
|
|
|
|
self.assertEqual(event.summary, "Local desired")
|
|
self.assertEqual(stats.unchanged, 1)
|
|
|
|
def test_conflict_blocks_inbound_until_admin_discards_local_desired_state(self) -> None:
|
|
event = self.create_local_event(summary="Local desired")
|
|
self.session.commit()
|
|
client = StatefulCalDAVClient()
|
|
client.resources[event.source_href or ""] = (different_event_ics(), '"remote"')
|
|
dispatch_calendar_outbox(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
operation = self.session.query(CalendarOutboxOperation).one()
|
|
self.assertEqual(operation.status, "conflict")
|
|
report = CalDAVReportResult(
|
|
objects=[
|
|
CalDAVObject(
|
|
href=event.source_href or "",
|
|
etag='"remote"',
|
|
calendar_data=different_event_ics(),
|
|
)
|
|
]
|
|
)
|
|
stats = type(
|
|
"Stats",
|
|
(),
|
|
{
|
|
"created": 0,
|
|
"updated": 0,
|
|
"deleted": 0,
|
|
"unchanged": 0,
|
|
"fetched": 0,
|
|
"full_sync": False,
|
|
},
|
|
)()
|
|
apply_caldav_report(
|
|
self.session,
|
|
source=self.source,
|
|
client=client,
|
|
report=report,
|
|
stats=stats,
|
|
user_id=None,
|
|
)
|
|
self.assertEqual(event.summary, "Local desired")
|
|
self.assertEqual(stats.unchanged, 1)
|
|
|
|
self.source.sync_token = "incremental-token"
|
|
discard_calendar_outbox_operation(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
operation_id=operation.id,
|
|
)
|
|
self.session.commit()
|
|
self.assertEqual(operation.status, "cancelled")
|
|
self.assertEqual((event.metadata_ or {})["caldav"]["external_state"], "discarded")
|
|
self.assertIsNone(self.source.sync_token)
|
|
self.assertIsNotNone(self.source.next_sync_at)
|
|
|
|
stats.unchanged = 0
|
|
apply_caldav_report(
|
|
self.session,
|
|
source=self.source,
|
|
client=client,
|
|
report=report,
|
|
stats=stats,
|
|
user_id=None,
|
|
)
|
|
self.assertEqual(event.summary, "Remote different")
|
|
|
|
def test_newer_success_supersedes_old_conflict_and_unblocks_inbound(self) -> None:
|
|
event = self.create_local_event(summary="First local")
|
|
self.session.commit()
|
|
client = StatefulCalDAVClient()
|
|
client.resources[event.source_href or ""] = (different_event_ics(), '"remote"')
|
|
dispatch_calendar_outbox(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
first = self.session.query(CalendarOutboxOperation).one()
|
|
self.assertEqual(first.status, "conflict")
|
|
|
|
update_event(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id=None,
|
|
event_id=event.id,
|
|
payload=CalendarEventUpdateRequest(summary="Second local"),
|
|
)
|
|
self.session.commit()
|
|
self.assertEqual(first.status, "superseded")
|
|
client.resources.clear()
|
|
dispatch_calendar_outbox(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
client_factory=lambda _session, _source: client,
|
|
)
|
|
latest = (
|
|
self.session.query(CalendarOutboxOperation)
|
|
.order_by(CalendarOutboxOperation.created_at.desc())
|
|
.first()
|
|
)
|
|
assert latest is not None
|
|
self.assertEqual(latest.status, "succeeded")
|
|
|
|
report = CalDAVReportResult(
|
|
objects=[
|
|
CalDAVObject(
|
|
href=event.source_href or "",
|
|
etag='"remote-new"',
|
|
calendar_data=different_event_ics(),
|
|
)
|
|
]
|
|
)
|
|
stats = type(
|
|
"Stats",
|
|
(),
|
|
{
|
|
"created": 0,
|
|
"updated": 0,
|
|
"deleted": 0,
|
|
"unchanged": 0,
|
|
"fetched": 0,
|
|
"full_sync": False,
|
|
},
|
|
)()
|
|
apply_caldav_report(
|
|
self.session,
|
|
source=self.source,
|
|
client=client,
|
|
report=report,
|
|
stats=stats,
|
|
user_id=None,
|
|
)
|
|
self.assertEqual(event.summary, "Remote different")
|
|
|
|
|
|
def different_event_ics() -> str:
|
|
return """BEGIN:VCALENDAR
|
|
VERSION:2.0
|
|
BEGIN:VEVENT
|
|
UID:event-1@example.test
|
|
DTSTART:20260708T090000Z
|
|
DTEND:20260708T100000Z
|
|
SUMMARY:Remote different
|
|
END:VEVENT
|
|
END:VCALENDAR
|
|
"""
|
|
|
|
|
|
def imported_event_ics(summary: str) -> str:
|
|
return f"""BEGIN:VCALENDAR
|
|
VERSION:2.0
|
|
BEGIN:VEVENT
|
|
UID:imported-event@example.test
|
|
DTSTART:20260710T090000Z
|
|
DTEND:20260710T100000Z
|
|
SUMMARY:{summary}
|
|
END:VEVENT
|
|
END:VCALENDAR
|
|
"""
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|