alexei.dolgolyov
7380b33b9b
Some checks failed
Lint & Test / test (push) Failing after 9s
Auto-fixed 138 unused imports and f-string issues. Manually fixed: ambiguous variable names (l→layer), availability-check imports using importlib.util.find_spec, unused Color import, ImagePool forward ref via TYPE_CHECKING, multi-statement semicolons, and E402 suppression.
121 lines
4.7 KiB
Python
121 lines
4.7 KiB
Python
"""E2E: Backup and restore flow.
|
|
|
|
Tests creating entities, backing up, deleting, then restoring from backup.
|
|
"""
|
|
|
|
import io
|
|
import json
|
|
|
|
|
|
|
|
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": "static",
|
|
"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 JSON file)
|
|
resp = client.get("/api/v1/system/backup")
|
|
assert resp.status_code == 200
|
|
backup_data = resp.json()
|
|
assert backup_data["meta"]["format"] == "ledgrab-backup"
|
|
assert "stores" in backup_data
|
|
assert "devices" in backup_data["stores"]
|
|
assert "color_strip_sources" in backup_data["stores"]
|
|
|
|
# Verify device is in the backup.
|
|
# Store files have structure: {"version": "...", "devices": {id: {...}}}
|
|
devices_store = backup_data["stores"]["devices"]
|
|
assert "devices" in devices_store
|
|
assert len(devices_store["devices"]) == 1
|
|
|
|
# 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 backup JSON as a file upload)
|
|
backup_bytes = json.dumps(backup_data).encode("utf-8")
|
|
resp = client.post(
|
|
"/api/v1/system/restore",
|
|
files={"file": ("backup.json", io.BytesIO(backup_bytes), "application/json")},
|
|
)
|
|
assert resp.status_code == 200, f"Restore failed: {resp.text}"
|
|
restore_result = resp.json()
|
|
assert restore_result["status"] == "restored"
|
|
assert restore_result["stores_written"] > 0
|
|
|
|
# 5. After restore, stores are written to disk but the in-memory
|
|
# stores haven't been re-loaded (normally a server restart does that).
|
|
# Verify the backup file was written correctly by reading it back.
|
|
# The restore endpoint writes JSON files; we check the response confirms success.
|
|
assert restore_result["restart_scheduled"] is True
|
|
|
|
def test_backup_contains_all_store_keys(self, client):
|
|
"""Backup response includes entries for all known store types."""
|
|
resp = client.get("/api/v1/system/backup")
|
|
assert resp.status_code == 200
|
|
stores = resp.json()["stores"]
|
|
# At minimum, these critical stores should be present
|
|
expected_keys = {
|
|
"devices", "output_targets", "color_strip_sources",
|
|
"capture_templates", "value_sources",
|
|
}
|
|
assert expected_keys.issubset(set(stores.keys()))
|
|
|
|
def test_restore_rejects_invalid_format(self, client):
|
|
"""Uploading a non-backup JSON file should fail validation."""
|
|
bad_data = json.dumps({"not": "a backup"}).encode("utf-8")
|
|
resp = client.post(
|
|
"/api/v1/system/restore",
|
|
files={"file": ("bad.json", io.BytesIO(bad_data), "application/json")},
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
def test_restore_rejects_empty_stores(self, client):
|
|
"""A backup with no recognized stores should fail."""
|
|
bad_backup = {
|
|
"meta": {"format": "ledgrab-backup", "format_version": 1},
|
|
"stores": {"unknown_store": {}},
|
|
}
|
|
resp = client.post(
|
|
"/api/v1/system/restore",
|
|
files={"file": ("bad.json", io.BytesIO(json.dumps(bad_backup).encode()), "application/json")},
|
|
)
|
|
assert resp.status_code == 400
|