MrSphay/odysseus
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: CalDAV write-back — push local event create/update/delete to the remote (#800) (#1282)

Browse Source 
* feat: CalDAV write-back — push local event create/update/delete to the remote (#800)

CalDAV sync was pull-only (src/caldav_sync.py), so events created, edited, or
deleted in Odysseus on a CalDAV-backed calendar only changed local SQLite and
never reached the server — they silently vanished on the next pull and never
appeared on the user's phone (iCloud, etc.).

This adds the missing write half:
- src/caldav_writeback.py builds the VEVENT, re-discovers the remote calendar by
  the same URL-hash the local id was derived from (the remote URL isn't stored),
  and PUTs/DELETEs the event by UID via the caldav lib. The pure pieces
  (build_event_ical, find_remote_calendar, push_event) take inputs by argument so
  they unit-test against a fake client with no network.
- create/update/delete event handlers (routes/calendar_routes.py) call it
  best-effort for caldav-sourced calendars only: the local DB stays the source of
  truth, a remote failure is logged, never fatal, and local calendars are untouched.

Tests: tests/test_caldav_writeback.py (9, pure logic incl. iCal serialization,
hash discovery, create/update/delete orchestration) and
tests/test_caldav_writeback_route.py (3, route-level: a caldav calendar pushes,
a local one does not, delete pushes a delete). 12 passed.

Note: write-back re-discovers the remote calendar per write (the URL isn't
persisted locally); a follow-up could cache it. Live-iCloud verification needs a
real account — flagging for a maintainer pass.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* test: drive #800 route regression without TestClient (fixes local hang)

Same fix as the document route test: the CalDAV write-back route regression used
Starlette TestClient (middleware app + threadpool) which hung in the maintainer's
environment. Rework it to call the async create/delete calendar handlers directly
— extracted from the router — with a minimal fake request, temp-SQLite-patched
SessionLocal, and writeback_event stubbed to record calls. Same coverage (a
caldav calendar pushes, a local one does not, delete pushes a delete), completes
in ~0.3s with no TestClient.

Verified the maintainer's exact batch:
  pytest tests/test_caldav_writeback.py tests/test_caldav_writeback_route.py -> 12 passed

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
lekt8
2026-06-03 00:44:02 +08:00
committed by GitHub
parent 7504fedb17
commit 1507d140b8
4 changed files with 415 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
routes/calendar_routes.py
View File

@@ -758,6 +758,16 @@ def setup_calendar_routes() -> APIRouter:
)
db.add(ev)
db.commit()
if cal.source == "caldav":
# Push the new event to the remote so it appears on the user's
# other devices — the sync is otherwise pull-only (#800).
from src.caldav_writeback import writeback_event
await writeback_event(owner, cal.source, cal.id, {
"uid": uid, "summary": data.summary, "description": data.description,
"location": data.location, "dtstart": dtstart, "dtend": dtend,
"all_day": data.all_day, "is_utc": _is_utc and not data.all_day,
"rrule": data.rrule or "",
})
return {"ok": True, "uid": uid}
except HTTPException:
raise
@@ -804,6 +814,14 @@ def setup_calendar_routes() -> APIRouter:
if data.color is not None:
ev.color = data.color if data.color else None
db.commit()
cal = db.query(CalendarCal).filter(CalendarCal.id == ev.calendar_id).first()
if cal and cal.source == "caldav":
from src.caldav_writeback import writeback_event
await writeback_event(owner, cal.source, cal.id, {
"uid": ev.uid, "summary": ev.summary, "description": ev.description,
"location": ev.location, "dtstart": ev.dtstart, "dtend": ev.dtend,
"all_day": ev.all_day, "is_utc": ev.is_utc, "rrule": ev.rrule or "",
})
return {"ok": True}
except HTTPException:
raise
@@ -824,8 +842,15 @@ def setup_calendar_routes() -> APIRouter:
db = SessionLocal()
try:
ev = _get_or_404_event(db, base_uid, owner)
# Capture what the remote push needs BEFORE the row is gone.
_cal = db.query(CalendarCal).filter(CalendarCal.id == ev.calendar_id).first()
_is_caldav = bool(_cal and _cal.source == "caldav")
_cal_id, _ev_uid = ev.calendar_id, ev.uid
db.delete(ev)
db.commit()
if _is_caldav:
from src.caldav_writeback import writeback_event
await writeback_event(owner, "caldav", _cal_id, {"uid": _ev_uid}, delete=True)
return {"ok": True}
except HTTPException:
raise

169
src/caldav_writeback.py Normal file
View File

@@ -0,0 +1,169 @@
"""CalDAV write-back: push local create/update/delete out to the remote (#800).
``src/caldav_sync.py`` is a one-way pull (remote → local). So events created,
edited, or deleted in Odysseus on a CalDAV-backed calendar only changed the local
SQLite copy and never reached the server (iCloud/Nextcloud/Radicale/Fastmail) —
they'd silently disappear on the next pull and never show on the user's phone.
This adds the missing write half. The remote calendar URL isn't stored locally
(the local calendar id is a one-way hash of it), so we re-discover the remote
calendar by matching that same hash, then PUT/DELETE the VEVENT by its UID via
the `caldav` lib. Writes are best-effort: the local DB stays the source of truth,
and a remote failure is reported, never fatal to the local operation.
The pure pieces (``build_event_ical``, ``find_remote_calendar``, ``push_event``)
take their inputs by argument so they unit-test against a fake client with no
network.
"""
import asyncio
import logging
from datetime import timezone
logger = logging.getLogger(__name__)
def _stable_cal_id(remote_url: str) -> str:
# Reuse the sync module's hashing so a local CalDAV calendar id maps back to
# the same remote URL it was pulled from.
from src.caldav_sync import _stable_cal_id as _sync_id
return _sync_id(remote_url)
def build_event_ical(ev: dict) -> str:
"""Serialize a local event dict to a VCALENDAR/VEVENT iCalendar string.
``ev`` keys: uid, summary, description, location, dtstart (datetime),
dtend (datetime), all_day (bool), is_utc (bool), rrule (str).
Mirrors how the pull path interprets is_utc/all_day so a round-trip is stable.
"""
from icalendar import Calendar, Event as iEvent
from icalendar.prop import vRecur
cal = Calendar()
cal.add("prodid", "-//Odysseus//CalDAV write-back//EN")
cal.add("version", "2.0")
ve = iEvent()
ve.add("uid", ev["uid"])
ve.add("summary", ev.get("summary") or "")
if ev.get("description"):
ve.add("description", ev["description"])
if ev.get("location"):
ve.add("location", ev["location"])
dtstart = ev["dtstart"]
dtend = ev["dtend"]
if ev.get("all_day"):
ve.add("dtstart", dtstart.date())
ve.add("dtend", dtend.date())
elif ev.get("is_utc"):
# Stored as naive-UTC instants — re-attach UTC so the server gets a Z time.
ve.add("dtstart", dtstart.replace(tzinfo=timezone.utc))
ve.add("dtend", dtend.replace(tzinfo=timezone.utc))
else:
# Legacy naive-local ("floating") time — emit without a TZ.
ve.add("dtstart", dtstart)
ve.add("dtend", dtend)
if ev.get("rrule"):
try:
ve.add("rrule", vRecur.from_ical(ev["rrule"]))
except Exception:
logger.debug("CalDAV write-back: skipping unparseable rrule %r", ev.get("rrule"))
cal.add_component(ve)
return cal.to_ical().decode("utf-8")
def find_remote_calendar(calendars, local_cal_id: str):
"""Find the remote calendar whose URL hashes to ``local_cal_id``, or None."""
for cal in calendars:
try:
if _stable_cal_id(str(cal.url)) == local_cal_id:
return cal
except Exception:
continue
return None
def push_event(calendars, local_cal_id: str, ev: dict, *, delete: bool = False) -> dict:
"""Create/update (or delete) ``ev`` on the matching remote calendar.
Returns ``{"ok": bool, ...}``. ``calendars`` is the discovered caldav
calendar list (injected so this is unit-testable with fakes).
"""
remote = find_remote_calendar(calendars, local_cal_id)
if remote is None:
return {"ok": False, "error": "remote calendar not found"}
uid = ev["uid"]
try:
existing = remote.event_by_uid(uid)
except Exception:
existing = None
if delete:
if existing is None:
return {"ok": True, "note": "already absent on remote"}
existing.delete()
return {"ok": True}
ical = build_event_ical(ev)
if existing is not None:
existing.data = ical
existing.save()
return {"ok": True, "updated": True}
remote.save_event(ical)
return {"ok": True, "created": True}
def _discover_calendars(client):
"""Discover the principal's calendars, falling back to the URL itself —
same strategy as the pull path."""
from caldav.lib.error import AuthorizationError, NotFoundError
try:
return client.principal().calendars()
except (AuthorizationError, NotFoundError):
raise
except Exception:
try:
return [client.calendar(url=str(client.url))]
except Exception:
return []
def _writeback_blocking(local_cal_id, ev, delete, url, username, password) -> dict:
import caldav
client = caldav.DAVClient(url=url, username=username, password=password)
calendars = _discover_calendars(client)
if not calendars:
return {"ok": False, "error": "no remote calendars discovered"}
return push_event(calendars, local_cal_id, ev, delete=delete)
async def writeback_event(owner: str, calendar_source: str, calendar_id: str,
ev: dict, *, delete: bool = False) -> dict:
"""Best-effort push of a local change to the remote CalDAV server.
No-ops (``{"skipped": ...}``) when the calendar isn't CalDAV-backed or no
credentials are configured. Never raises — a remote failure is logged and
returned, the local DB remaining the source of truth.
"""
if calendar_source != "caldav":
return {"skipped": "not a caldav calendar"}
try:
from routes.prefs_routes import _load_for_user
cfg = (_load_for_user(owner) or {}).get("caldav", {}) or {}
url = (cfg.get("url") or "").strip()
user = (cfg.get("username") or "").strip()
pw = cfg.get("password") or ""
if not (url and user and pw):
return {"skipped": "caldav not configured"}
result = await asyncio.to_thread(_writeback_blocking, calendar_id, ev, delete, url, user, pw)
if not result.get("ok"):
logger.warning("CalDAV write-back did not apply: %s", result.get("error") or result)
return result
except Exception as e:
logger.exception("CalDAV write-back raised")
return {"ok": False, "error": str(e)[:200]}

118
tests/test_caldav_writeback.py Normal file
View File

@@ -0,0 +1,118 @@
"""Issue #800 — CalDAV write-back pushes local changes to the remote server.
Unit-tests the pure pieces against a fake caldav calendar (no network): the
iCalendar serialization, hash-based remote-calendar discovery, and the
create/update/delete orchestration.
"""
from datetime import datetime
from src.caldav_writeback import (
build_event_ical,
find_remote_calendar,
push_event,
_stable_cal_id,
)
REMOTE_URL = "https://p69-caldav.icloud.com/123/calendars/home/"
CAL_ID = _stable_cal_id(REMOTE_URL)
class FakeEvent:
def __init__(self):
self.data = "OLD"
self.saved = False
self.deleted = False
def save(self):
self.saved = True
def delete(self):
self.deleted = True
class FakeCalendar:
def __init__(self, url, existing=None):
self.url = url
self._existing = existing
self.saved_ical = None
def event_by_uid(self, uid):
if self._existing is None:
raise Exception("not found")
return self._existing
def save_event(self, ical):
self.saved_ical = ical
def _ev(**over):
base = dict(
uid="evt-1", summary="Dentist", description="bring x-rays",
location="Clinic", dtstart=datetime(2026, 6, 10, 14, 0),
dtend=datetime(2026, 6, 10, 15, 0), all_day=False, is_utc=True, rrule="",
)
base.update(over)
return base
def test_build_ical_timed_event_has_core_fields():
ical = build_event_ical(_ev())
assert "BEGIN:VEVENT" in ical and "END:VEVENT" in ical
assert "UID:evt-1" in ical
assert "SUMMARY:Dentist" in ical
# is_utc -> UTC instant (Z suffix)
assert "DTSTART:20260610T140000Z" in ical
assert "DTEND:20260610T150000Z" in ical
def test_build_ical_all_day_uses_date_values():
ical = build_event_ical(_ev(all_day=True, is_utc=False))
assert "DTSTART;VALUE=DATE:20260610" in ical
def test_build_ical_includes_rrule():
ical = build_event_ical(_ev(rrule="FREQ=WEEKLY;BYDAY=MO"))
assert "RRULE:FREQ=WEEKLY" in ical
def test_find_remote_calendar_matches_by_hash():
cals = [FakeCalendar("https://other/x/"), FakeCalendar(REMOTE_URL)]
found = find_remote_calendar(cals, CAL_ID)
assert found is cals[1]
assert find_remote_calendar([FakeCalendar("https://nope/")], CAL_ID) is None
def test_push_create_calls_save_event():
cal = FakeCalendar(REMOTE_URL, existing=None) # event_by_uid raises -> create
res = push_event([cal], CAL_ID, _ev(), delete=False)
assert res["ok"] and res.get("created")
assert cal.saved_ical and "UID:evt-1" in cal.saved_ical
def test_push_update_overwrites_existing():
existing = FakeEvent()
cal = FakeCalendar(REMOTE_URL, existing=existing)
res = push_event([cal], CAL_ID, _ev(summary="Moved"), delete=False)
assert res["ok"] and res.get("updated")
assert existing.saved and "SUMMARY:Moved" in existing.data
assert cal.saved_ical is None # used update path, not create
def test_push_delete_removes_existing():
existing = FakeEvent()
cal = FakeCalendar(REMOTE_URL, existing=existing)
res = push_event([cal], CAL_ID, _ev(), delete=True)
assert res["ok"] and existing.deleted
def test_push_delete_absent_is_ok():
cal = FakeCalendar(REMOTE_URL, existing=None)
res = push_event([cal], CAL_ID, _ev(), delete=True)
assert res["ok"] and "absent" in res.get("note", "")
def test_push_unknown_calendar_reports_not_found():
cal = FakeCalendar("https://different/")
res = push_event([cal], CAL_ID, _ev())
assert res["ok"] is False and "not found" in res["error"]

103
tests/test_caldav_writeback_route.py Normal file
View File

@@ -0,0 +1,103 @@
"""Issue #800 — the calendar write handlers actually trigger CalDAV write-back.
Route-level: proves POST/DELETE /api/calendar/events fire writeback_event for a
CalDAV-backed calendar and not for a local one.
Calls the async route handlers DIRECTLY (extracted from the router) rather than
through Starlette's TestClient — the TestClient middleware-app + threadpool could
hang in some environments; a direct call with a minimal fake request keeps the
same coverage and completes reliably.
"""
import tempfile
import uuid
from types import SimpleNamespace
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import NullPool
import core.database as cdb
import routes.calendar_routes as croutes
import src.caldav_writeback as wb
from core.database import CalendarCal
from routes.calendar_routes import EventCreate
_TMPDB = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
_ENGINE = create_engine(
f"sqlite:///{_TMPDB.name}",
connect_args={"check_same_thread": False},
poolclass=NullPool,
)
cdb.Base.metadata.create_all(_ENGINE)
_TS = sessionmaker(bind=_ENGINE, autoflush=False, autocommit=False)
croutes.SessionLocal = _TS
@pytest.fixture
def calls(monkeypatch):
recorded = []
async def _fake_writeback(owner, source, cal_id, ev, *, delete=False):
recorded.append({"source": source, "cal_id": cal_id, "uid": ev.get("uid"), "delete": delete})
return {"ok": True}
monkeypatch.setattr(wb, "writeback_event", _fake_writeback)
return recorded
def _req():
return SimpleNamespace(state=SimpleNamespace(current_user="tester"))
def _endpoint(method, suffix):
router = croutes.setup_calendar_routes()
for r in router.routes:
if getattr(r, "path", "").endswith(suffix) and method in getattr(r, "methods", set()):
return r.endpoint
raise RuntimeError(f"{method} *{suffix} not found")
def _make_cal(source):
cid = ("caldav-" if source == "caldav" else "loc-") + uuid.uuid4().hex[:10]
db = _TS()
try:
db.add(CalendarCal(id=cid, owner="tester", name="C", source=source))
db.commit()
return cid
finally:
db.close()
async def test_create_on_caldav_calendar_pushes_to_remote(calls):
create_event = _endpoint("POST", "/events")
cal_id = _make_cal("caldav")
res = await create_event(_req(), EventCreate(
summary="Dentist", dtstart="2026-06-10T14:00:00Z", calendar_href=cal_id))
assert res["ok"] is True
assert len(calls) == 1
assert calls[0]["source"] == "caldav" and calls[0]["cal_id"] == cal_id
assert calls[0]["delete"] is False
async def test_create_on_local_calendar_does_not_push(calls):
create_event = _endpoint("POST", "/events")
cal_id = _make_cal("local")
res = await create_event(_req(), EventCreate(
summary="Local", dtstart="2026-06-10T14:00:00Z", calendar_href=cal_id))
assert res["ok"] is True
assert calls == []
async def test_delete_on_caldav_calendar_pushes_delete(calls):
create_event = _endpoint("POST", "/events")
delete_event = _endpoint("DELETE", "/events/{uid}")
cal_id = _make_cal("caldav")
res = await create_event(_req(), EventCreate(
summary="Temp", dtstart="2026-06-10T14:00:00Z", calendar_href=cal_id))
uid = res["uid"]
calls.clear()
rd = await delete_event(_req(), uid)
assert rd["ok"] is True
assert len(calls) == 1 and calls[0]["delete"] is True and calls[0]["uid"] == uid