247 lines
8.9 KiB
Python
247 lines
8.9 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
import pytest
|
|
from backup_tool.config import Settings
|
|
from backup_tool.db.engine import create_engine
|
|
from backup_tool.db.models import Backup, Execution, Repository, Restore
|
|
from backup_tool.snapshot import verify_published_snapshot
|
|
from backup_tool.worker import Worker
|
|
from sqlalchemy import select
|
|
|
|
PASSWORD = "correct-horse-battery-staple"
|
|
|
|
|
|
async def login(client: httpx.AsyncClient) -> dict[str, str]:
|
|
response = await client.post("/api/v2/setup", json={"username": "admin", "password": PASSWORD})
|
|
assert response.status_code == 201
|
|
return {"X-CSRF-Token": client.cookies["backup_tool_csrf"]}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_worker_publishes_a_verified_signed_full_backup_and_atomic_restore(
|
|
app_client: tuple[httpx.AsyncClient, Settings],
|
|
) -> None:
|
|
client, settings = app_client
|
|
source_root = settings.local_source_roots[0] / "project"
|
|
source_root.mkdir()
|
|
(source_root / "nested").mkdir()
|
|
(source_root / "nested" / "hello.txt").write_text("hello backup\n", encoding="utf-8")
|
|
headers = await login(client)
|
|
repository_response = await client.post(
|
|
"/api/v2/repositories",
|
|
json={
|
|
"name": "primary",
|
|
"relative_path": "primary",
|
|
"compression": "none",
|
|
"encryption": "none",
|
|
},
|
|
headers=headers,
|
|
)
|
|
assert repository_response.status_code == 201
|
|
source_response = await client.post(
|
|
"/api/v2/sources",
|
|
json={
|
|
"name": "local",
|
|
"kind": "local",
|
|
"public_config": {"root": str(source_root)},
|
|
},
|
|
headers=headers,
|
|
)
|
|
assert source_response.status_code == 201
|
|
job_response = await client.post(
|
|
"/api/v2/jobs",
|
|
json={
|
|
"name": "full-backup",
|
|
"source_id": source_response.json()["id"],
|
|
"repository_id": repository_response.json()["id"],
|
|
"requested_mode": "full",
|
|
"exclusions": [],
|
|
"retention": {},
|
|
"enabled": True,
|
|
"allow_empty": False,
|
|
},
|
|
headers=headers,
|
|
)
|
|
assert job_response.status_code == 201
|
|
execution_response = await client.post(
|
|
f"/api/v2/jobs/{job_response.json()['id']}/executions", headers=headers
|
|
)
|
|
assert execution_response.status_code == 202
|
|
execution_id = execution_response.json()["id"]
|
|
|
|
worker = Worker(settings, owner="snapshot-worker")
|
|
try:
|
|
assert await worker.run_once()
|
|
finally:
|
|
await worker.engine.dispose()
|
|
|
|
engine = create_engine(settings)
|
|
try:
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker
|
|
|
|
sessions = async_sessionmaker(engine, expire_on_commit=False)
|
|
async with sessions() as db:
|
|
execution = await db.get(Execution, execution_id)
|
|
backup = await db.scalar(select(Backup).where(Backup.execution_id == execution_id))
|
|
repository = await db.get(Repository, repository_response.json()["id"])
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
assert execution is not None
|
|
assert execution.state == "committed"
|
|
assert backup is not None
|
|
assert backup.integrity == "verified"
|
|
assert repository is not None
|
|
root = Path(repository.root)
|
|
manifest_path = root / "manifests" / f"{backup.manifest_id}.json"
|
|
manifest = verify_published_snapshot(root, manifest_path, repository.signing_public_key)
|
|
file_entry = next(entry for entry in manifest["entries"] if entry["type"] == "file")
|
|
assert file_entry["path"] == "nested/hello.txt"
|
|
assert (root / "blobs" / "sha256" / file_entry["blob_digest"]).read_text() == "hello backup\n"
|
|
|
|
dry_run_destination = settings.restore_roots[0] / "dry-run-backup"
|
|
dry_run_response = await client.post(
|
|
f"/api/v2/backups/{backup.id}/restores",
|
|
json={
|
|
"destination": str(dry_run_destination),
|
|
"selection": ["nested"],
|
|
"dry_run": True,
|
|
"overwrite_policy": "fail",
|
|
},
|
|
headers=headers,
|
|
)
|
|
assert dry_run_response.status_code == 202
|
|
dry_run_worker = Worker(settings, owner="dry-run-worker")
|
|
try:
|
|
assert await dry_run_worker.run_once()
|
|
finally:
|
|
await dry_run_worker.engine.dispose()
|
|
dry_run = await client.get(f"/api/v2/restores/{dry_run_response.json()['id']}", headers=headers)
|
|
assert dry_run.json()["state"] == "committed"
|
|
assert dry_run.json()["result"]["dry_run"]
|
|
assert dry_run.json()["result"]["entry_count"] == 2
|
|
assert not dry_run_destination.exists()
|
|
|
|
destination = settings.restore_roots[0] / "restored-backup"
|
|
restore_response = await client.post(
|
|
f"/api/v2/backups/{backup.id}/restores",
|
|
json={
|
|
"destination": str(destination),
|
|
"selection": [],
|
|
"overwrite_policy": "fail",
|
|
},
|
|
headers=headers,
|
|
)
|
|
assert restore_response.status_code == 202
|
|
restore_id = restore_response.json()["id"]
|
|
assert restore_response.json()["state"] == "queued"
|
|
source_root.rename(settings.data_dir / "removed-source")
|
|
|
|
restore_worker = Worker(settings, owner="restore-worker")
|
|
try:
|
|
assert await restore_worker.run_once()
|
|
finally:
|
|
await restore_worker.engine.dispose()
|
|
|
|
restored = await client.get(f"/api/v2/restores/{restore_id}", headers=headers)
|
|
assert restored.status_code == 200
|
|
assert restored.json()["state"] == "committed"
|
|
assert restored.json()["result"]["manifest_digest"] == backup.manifest_digest
|
|
assert (destination / "nested" / "hello.txt").read_text(encoding="utf-8") == "hello backup\n"
|
|
|
|
engine = create_engine(settings)
|
|
try:
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker
|
|
|
|
sessions = async_sessionmaker(engine, expire_on_commit=False)
|
|
async with sessions() as db:
|
|
interrupted_restore = await db.get(Restore, restore_id)
|
|
assert interrupted_restore is not None
|
|
interrupted_restore.state = "running"
|
|
interrupted_restore.result = None
|
|
await db.commit()
|
|
finally:
|
|
await engine.dispose()
|
|
recovery_worker = Worker(settings, owner="recovery-worker")
|
|
try:
|
|
assert await recovery_worker.startup() == 1
|
|
finally:
|
|
await recovery_worker.engine.dispose()
|
|
recovered = await client.get(f"/api/v2/restores/{restore_id}", headers=headers)
|
|
assert recovered.json()["state"] == "committed"
|
|
|
|
skipped_restore = await client.post(
|
|
f"/api/v2/backups/{backup.id}/restores",
|
|
json={
|
|
"destination": str(destination),
|
|
"selection": [],
|
|
"overwrite_policy": "skip",
|
|
},
|
|
headers=headers,
|
|
)
|
|
assert skipped_restore.status_code == 202
|
|
skip_worker = Worker(settings, owner="skip-worker")
|
|
try:
|
|
assert await skip_worker.run_once()
|
|
finally:
|
|
await skip_worker.engine.dispose()
|
|
skipped = await client.get(f"/api/v2/restores/{skipped_restore.json()['id']}", headers=headers)
|
|
assert skipped.json()["result"]["skipped"]
|
|
|
|
(destination / "nested" / "hello.txt").write_text("replaced", encoding="utf-8")
|
|
replaced_restore = await client.post(
|
|
f"/api/v2/backups/{backup.id}/restores",
|
|
json={
|
|
"destination": str(destination),
|
|
"selection": [],
|
|
"overwrite_policy": "replace",
|
|
},
|
|
headers=headers,
|
|
)
|
|
assert replaced_restore.status_code == 202
|
|
replace_worker = Worker(settings, owner="replace-worker")
|
|
try:
|
|
assert await replace_worker.run_once()
|
|
finally:
|
|
await replace_worker.engine.dispose()
|
|
assert (destination / "nested" / "hello.txt").read_text(encoding="utf-8") == "hello backup\n"
|
|
|
|
(root / "blobs" / "sha256" / file_entry["blob_digest"]).write_text("tampered")
|
|
corrupt_destination = settings.restore_roots[0] / "corrupt-restore"
|
|
corrupt_restore = await client.post(
|
|
f"/api/v2/backups/{backup.id}/restores",
|
|
json={
|
|
"destination": str(corrupt_destination),
|
|
"selection": [],
|
|
"overwrite_policy": "fail",
|
|
},
|
|
headers=headers,
|
|
)
|
|
assert corrupt_restore.status_code == 202
|
|
|
|
corrupt_worker = Worker(settings, owner="corrupt-restore-worker")
|
|
try:
|
|
assert await corrupt_worker.run_once()
|
|
finally:
|
|
await corrupt_worker.engine.dispose()
|
|
|
|
corrupt_status = await client.get(
|
|
f"/api/v2/restores/{corrupt_restore.json()['id']}", headers=headers
|
|
)
|
|
assert corrupt_status.json()["state"] == "failed"
|
|
assert not corrupt_destination.exists()
|
|
engine = create_engine(settings)
|
|
try:
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker
|
|
|
|
sessions = async_sessionmaker(engine, expire_on_commit=False)
|
|
async with sessions() as db:
|
|
corrupted_backup = await db.get(Backup, backup.id)
|
|
finally:
|
|
await engine.dispose()
|
|
assert corrupted_backup is not None
|
|
assert corrupted_backup.integrity == "corrupt"
|