61 lines
2.2 KiB
Python
61 lines
2.2 KiB
Python
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)
|