[verified] feat: add external calendar subscriptions

2026-09-20 12:29:21 +08:00
parent d7014c4f4f
commit 583b6a8a9d
11 changed files with 849 additions and 2 deletions
@@ -0,0 +1,60 @@
+import asyncio
+import json
+import zipfile
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import async_sessionmaker
+
+from backend.backup.archive import parse_archive_path
+from backend.backup.service import export_v2, validate_archive
+from backend.db import get_engine
+from backend.models import CalendarSubscription, User
+
+
+def initialized(client):
+    response = client.post(
+        "/api/v1/setup/initialize",
+        json={"username": "owner", "password": "correct horse battery staple"},
+    )
+    assert response.status_code == 201
+    return client
+
+
+def test_backup_exports_calendar_subscriptions_and_accepts_old_archive(client, tmp_path):
+    client = initialized(client)
+
+    async def prepare_and_export():
+        factory = async_sessionmaker(get_engine(), expire_on_commit=False)
+        async with factory() as db:
+            user = await db.scalar(select(User))
+            db.add(CalendarSubscription(
+                user_id=user.id,
+                name="Work",
+                url="https://example.com/work.ics",
+                color="#123abc",
+                ics_cache="BEGIN:VCALENDAR\r\nVERSION:2.0\r\nEND:VCALENDAR\r\n",
+            ))
+            await db.commit()
+            path = await export_v2(db, user)
+            await db.commit()
+            return path
+
+    path = asyncio.run(prepare_and_export())
+    try:
+        parsed = parse_archive_path(path)
+        assert parsed.entities["calendar_subscriptions"][0]["name"] == "Work"
+        validate_archive(parsed)
+
+        old_path = tmp_path / "old.zip"
+        with zipfile.ZipFile(path) as source, zipfile.ZipFile(old_path, "w") as target:
+            manifest = json.loads(source.read("manifest.json"))
+            manifest["entities"].pop("calendar_subscriptions")
+            manifest["checksums"].pop("data/calendar_subscriptions.json")
+            for name in source.namelist():
+                if name not in {"manifest.json", "data/calendar_subscriptions.json"}:
+                    target.writestr(name, source.read(name))
+            from backend.backup.archive import canonical_json
+            target.writestr("manifest.json", canonical_json(manifest))
+        validate_archive(parse_archive_path(old_path))
+    finally:
+        path.unlink(missing_ok=True)