alexei.dolgolyov
898912f8b1
Bundle the remaining backend touch-ups that the production review landed individually as small surgical edits across many modules: - MQTT runtime: fire-and-forget task tracking + drain resilience. - mqtt_source + store + storage/color_strip_source: secret_box encryption for credentials with auto-migration of plaintext fields. - devices/discovery_watcher: task tracking on watcher start/stop. - devices/wled_client + wled_provider: URL scheme inference helper applied at the create/update boundary so bare hostnames stay valid. - core/capture/screen_capture: hardened error paths. - core/processing (mapped/processed/processor_manager/video/wled_target): smaller follow-throughs from the registry refactor that landed earlier on the branch. - utils/safe_source + utils/file_ops + utils/__init__: shared URL + IP classification helpers + larger streaming upload size caps. - api/auth: WebSocket Origin allow-list + /docs auth-gate. - api/dependencies: register the new HTTP-endpoint store. - api/routes (assets, backup, webhooks): streaming-upload caps + asyncio.gather return_exceptions on broadcast loops. - tests/test_api + tests/e2e/test_backup_flow: cover the new caps and the Origin allow-list.
113 lines
4.2 KiB
Python
113 lines
4.2 KiB
Python
"""E2E: Backup and restore flow.
|
|
|
|
Tests creating entities, backing up (ZIP containing SQLite .db + asset files),
|
|
deleting, then restoring.
|
|
"""
|
|
|
|
import io
|
|
import zipfile
|
|
|
|
|
|
class TestBackupRestoreFlow:
|
|
"""A user backs up their configuration and restores it."""
|
|
|
|
def _create_device(self, client, name="Backup Device") -> str:
|
|
resp = client.post(
|
|
"/api/v1/devices",
|
|
json={
|
|
"name": name,
|
|
"url": "mock://backup",
|
|
"device_type": "mock",
|
|
"led_count": 30,
|
|
},
|
|
)
|
|
assert resp.status_code == 201
|
|
return resp.json()["id"]
|
|
|
|
def _create_css(self, client, name="Backup CSS") -> str:
|
|
resp = client.post(
|
|
"/api/v1/color-strip-sources",
|
|
json={
|
|
"name": name,
|
|
"source_type": "single_color",
|
|
"color": [255, 0, 0],
|
|
"led_count": 30,
|
|
},
|
|
)
|
|
assert resp.status_code == 201
|
|
return resp.json()["id"]
|
|
|
|
def test_backup_and_restore_roundtrip(self, client):
|
|
# 1. Create some entities
|
|
device_id = self._create_device(client, "Device for Backup")
|
|
css_id = self._create_css(client, "CSS for Backup")
|
|
|
|
# Verify entities exist
|
|
resp = client.get("/api/v1/devices")
|
|
assert resp.json()["count"] == 1
|
|
resp = client.get("/api/v1/color-strip-sources")
|
|
assert resp.json()["count"] == 1
|
|
|
|
# 2. Create a backup (GET returns a ZIP containing ledgrab.db + assets)
|
|
resp = client.get("/api/v1/system/backup")
|
|
assert resp.status_code == 200
|
|
backup_bytes = resp.content
|
|
# Backup is a ZIP file (PK magic bytes)
|
|
assert backup_bytes[:4] == b"PK\x03\x04"
|
|
# ZIP should contain ledgrab.db
|
|
with zipfile.ZipFile(io.BytesIO(backup_bytes)) as zf:
|
|
assert "ledgrab.db" in zf.namelist()
|
|
db_data = zf.read("ledgrab.db")
|
|
assert db_data[:16].startswith(b"SQLite format 3")
|
|
|
|
# 3. Delete all created entities
|
|
resp = client.delete(f"/api/v1/color-strip-sources/{css_id}")
|
|
assert resp.status_code == 204
|
|
resp = client.delete(f"/api/v1/devices/{device_id}")
|
|
assert resp.status_code == 204
|
|
|
|
# Verify they're gone
|
|
resp = client.get("/api/v1/devices")
|
|
assert resp.json()["count"] == 0
|
|
resp = client.get("/api/v1/color-strip-sources")
|
|
assert resp.json()["count"] == 0
|
|
|
|
# 4. Restore from backup (POST with the .zip file upload)
|
|
resp = client.post(
|
|
"/api/v1/system/restore",
|
|
files={"file": ("backup.zip", io.BytesIO(backup_bytes), "application/zip")},
|
|
)
|
|
assert resp.status_code == 200, f"Restore failed: {resp.text}"
|
|
restore_result = resp.json()
|
|
assert restore_result["status"] == "restored"
|
|
assert restore_result["restart_scheduled"] is True
|
|
|
|
def test_backup_is_valid_zip(self, client):
|
|
"""Backup response is a valid ZIP containing a SQLite database."""
|
|
resp = client.get("/api/v1/system/backup")
|
|
assert resp.status_code == 200
|
|
assert resp.content[:4] == b"PK\x03\x04"
|
|
# Should have Content-Disposition header for download
|
|
assert "attachment" in resp.headers.get("content-disposition", "")
|
|
# ZIP should contain ledgrab.db with valid SQLite header
|
|
with zipfile.ZipFile(io.BytesIO(resp.content)) as zf:
|
|
assert "ledgrab.db" in zf.namelist()
|
|
assert zf.read("ledgrab.db")[:16].startswith(b"SQLite format 3")
|
|
|
|
def test_restore_rejects_invalid_format(self, client):
|
|
"""Uploading a non-SQLite file should fail validation."""
|
|
bad_data = b"not a database file at all, just random text content"
|
|
resp = client.post(
|
|
"/api/v1/system/restore",
|
|
files={"file": ("bad.db", io.BytesIO(bad_data), "application/octet-stream")},
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
def test_restore_rejects_empty_file(self, client):
|
|
"""A tiny file should fail validation."""
|
|
resp = client.post(
|
|
"/api/v1/system/restore",
|
|
files={"file": ("tiny.db", io.BytesIO(b"x" * 50), "application/octet-stream")},
|
|
)
|
|
assert resp.status_code == 400
|