418 lines
17 KiB
Python
418 lines
17 KiB
Python
"""Planning and executing safe restores (US06-04).
|
|
|
|
Restoring is the one archive operation that can *add* a file to the library, so
|
|
every case here asks two questions: did the right bytes come back under the right
|
|
identity, and did anything already in the library get touched? The media are real
|
|
directories, the hashes are real, and the failure paths assert that the archived
|
|
copy is still exactly where it was — a restore that fails must cost nothing.
|
|
"""
|
|
|
|
import shutil
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
|
|
import numpy as np
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from PIL import Image
|
|
from sqlalchemy import select
|
|
|
|
from photo_pipeline.api.app import create_app
|
|
from photo_pipeline.config import Config
|
|
from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
|
|
from photo_pipeline.models import Asset, AssetPath, SafetyReview, UploadBatch, UploadItem
|
|
from photo_pipeline.services import availability
|
|
from photo_pipeline.services.archive_journal import ArchiveState
|
|
from photo_pipeline.services.archive_transfer import ArchiveTransferService
|
|
from photo_pipeline.services.archives import MARKER_NAME, ArchiveError, ArchiveService
|
|
from photo_pipeline.services.hashing import sha256_file
|
|
from photo_pipeline.services.inventory import InventoryService
|
|
from photo_pipeline.services.restores import RestoreService
|
|
|
|
pytestmark = pytest.mark.phase_f # part of the Phase F acceptance gate (US06-06)
|
|
|
|
NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
|
|
|
|
|
# ── environment ──────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _env(tmp_path):
|
|
(tmp_path / "data").mkdir(exist_ok=True)
|
|
lib = tmp_path / "lib"
|
|
lib.mkdir(exist_ok=True)
|
|
archive = tmp_path / "archive"
|
|
archive.mkdir(exist_ok=True)
|
|
config = Config.from_env(
|
|
{
|
|
"PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data"),
|
|
"PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib),
|
|
"PHOTO_PIPELINE_ARCHIVE_FREE_SPACE_RESERVE_BYTES": "0",
|
|
}
|
|
)
|
|
run_migrations(config.database_url)
|
|
return config, create_session_factory(create_db_engine(config.database_url)), lib, archive
|
|
|
|
|
|
def structured(path, seed, size=(192, 144)):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
rng = np.random.default_rng(seed)
|
|
w, h = size
|
|
base = np.zeros((h, w, 3), dtype=np.uint8)
|
|
for _ in range(5):
|
|
x0 = int(rng.integers(0, w - 40))
|
|
y0 = int(rng.integers(0, h - 40))
|
|
base[y0 : y0 + 40, x0 : x0 + 40] = rng.integers(0, 256, 3)
|
|
Image.fromarray(base).save(path, quality=95)
|
|
return path
|
|
|
|
|
|
def _archived(sf, config, lib, archive, album="rome", seeds=(1, 2)):
|
|
"""A real album taken all the way through archiving, ready to be restored."""
|
|
folder = lib / album
|
|
for index, seed in enumerate(seeds):
|
|
structured(folder / f"{index}.jpg", seed)
|
|
scan = InventoryService(sf).scan(lib)
|
|
with sf() as session:
|
|
batch_id = str(uuid.uuid4())
|
|
session.add(
|
|
UploadBatch(
|
|
id=batch_id,
|
|
album=album,
|
|
folder=str(folder),
|
|
album_name=album,
|
|
state="succeeded",
|
|
preflight_token="v1:test",
|
|
outcome_state="verified",
|
|
created_at=NOW,
|
|
)
|
|
)
|
|
for path, asset_id in scan.asset_ids.items():
|
|
session.add(
|
|
UploadItem(
|
|
batch_id=batch_id,
|
|
asset_id=asset_id,
|
|
path=path,
|
|
sha256=sha256_file(path),
|
|
sha1="0" * 40,
|
|
state="sent",
|
|
outcome="uploaded",
|
|
)
|
|
)
|
|
# A decision that must survive the whole round trip.
|
|
session.add(
|
|
SafetyReview(
|
|
id=str(uuid.uuid4()),
|
|
asset_id=asset_id,
|
|
decision="sfw",
|
|
score=0.01,
|
|
reviewer="test",
|
|
)
|
|
)
|
|
session.commit()
|
|
|
|
service = ArchiveService(sf, config=config)
|
|
location = service.register("external", str(archive))
|
|
token = service.preflight(location["id"])["token"]
|
|
transfers = ArchiveTransferService(sf, config=config)
|
|
plan = transfers.create(location["id"], None, token=token)
|
|
transfers.apply(plan["id"])
|
|
return location, scan.asset_ids
|
|
|
|
|
|
def _restore(sf, config, location_id, asset_ids=None):
|
|
service = RestoreService(sf, config=config)
|
|
token = service.preflight(location_id, asset_ids)["token"]
|
|
plan = service.create(location_id, asset_ids, token=token)
|
|
return service, plan, service.apply(plan["id"])
|
|
|
|
|
|
def _unmount(archive):
|
|
(archive / MARKER_NAME).rename(archive / f"{MARKER_NAME}.away")
|
|
|
|
|
|
def _assets(sf):
|
|
with sf() as session:
|
|
return {asset.id: asset for asset in session.scalars(select(Asset))}
|
|
|
|
|
|
def _codes(report):
|
|
return {issue["code"] for issue in report["blockers"]} | {
|
|
issue["code"] for item in report["items"] for issue in item["blockers"]
|
|
}
|
|
|
|
|
|
# ── preflight ────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_preflight_blocks_offline_medium(tmp_path):
|
|
config, sf, lib, archive = _env(tmp_path)
|
|
location, _ = _archived(sf, config, lib, archive)
|
|
_unmount(archive)
|
|
|
|
report = RestoreService(sf, config=config).preflight(location["id"])
|
|
assert report["state"] == "blocked"
|
|
assert "location_offline" in _codes(report)
|
|
|
|
|
|
def test_preflight_blocks_wrong_volume(tmp_path):
|
|
config, sf, lib, archive = _env(tmp_path)
|
|
location, _ = _archived(sf, config, lib, archive)
|
|
(archive / MARKER_NAME).write_text('{"media_id": "someone-elses-disk"}', encoding="utf-8")
|
|
|
|
report = RestoreService(sf, config=config).preflight(location["id"])
|
|
assert report["state"] == "blocked"
|
|
assert "wrong_volume" in _codes(report)
|
|
|
|
|
|
def test_preflight_blocks_changed_archive_bytes(tmp_path):
|
|
config, sf, lib, archive = _env(tmp_path)
|
|
location, ids = _archived(sf, config, lib, archive, seeds=(1,))
|
|
asset_id = next(iter(ids.values()))
|
|
with sf() as session:
|
|
archived_file = archive / session.get(Asset, asset_id).archive_path
|
|
archived_file.write_bytes(b"not the photo that was archived")
|
|
|
|
report = RestoreService(sf, config=config).preflight(location["id"])
|
|
assert report["state"] == "blocked"
|
|
assert "bytes_changed" in _codes(report)
|
|
with pytest.raises(ArchiveError) as error:
|
|
_restore(sf, config, location["id"])
|
|
assert error.value.code == "blocked"
|
|
|
|
|
|
def test_preflight_blocks_insufficient_capacity(tmp_path):
|
|
config, sf, lib, archive = _env(tmp_path)
|
|
location, _ = _archived(sf, config, lib, archive)
|
|
greedy = config.model_copy(
|
|
update={"archive_free_space_reserve_bytes": 1 << 62} # more than any disk has
|
|
)
|
|
|
|
report = RestoreService(sf, config=greedy).preflight(location["id"])
|
|
assert report["state"] == "blocked"
|
|
assert "insufficient_capacity" in _codes(report)
|
|
|
|
|
|
def test_token_changes_with_the_scope(tmp_path):
|
|
config, sf, lib, archive = _env(tmp_path)
|
|
location, ids = _archived(sf, config, lib, archive)
|
|
service = RestoreService(sf, config=config)
|
|
|
|
whole = service.preflight(location["id"])["token"]
|
|
partial = service.preflight(location["id"], [sorted(ids.values())[0]])["token"]
|
|
assert whole != partial
|
|
assert service.verify_token(whole, location["id"])
|
|
assert not service.verify_token(partial, location["id"])
|
|
|
|
|
|
# ── restore ──────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_restore_returns_bytes_identity_and_decisions(tmp_path):
|
|
config, sf, lib, archive = _env(tmp_path)
|
|
location, ids = _archived(sf, config, lib, archive)
|
|
archived_hashes = {
|
|
asset_id: asset.current_sha256 for asset_id, asset in _assets(sf).items()
|
|
}
|
|
|
|
service, plan, result = _restore(sf, config, location["id"])
|
|
assert (result["restored"], result["failed"], result["state"]) == (2, 0, "complete")
|
|
|
|
for asset_id, asset in _assets(sf).items():
|
|
assert asset.availability_state == availability.ACTIVE
|
|
assert asset.current_path == str(lib / asset.archive_path)
|
|
assert sha256_file(asset.current_path) == archived_hashes[asset_id]
|
|
# The archived copy is a copy: restoring never empties the medium.
|
|
assert (archive / asset.archive_path).exists()
|
|
assert asset.archive_location_id == location["id"]
|
|
|
|
with sf() as session:
|
|
# Identity and decisions survived: same ids, same reviews, new occurrence.
|
|
assert set(ids.values()) == {a.id for a in session.scalars(select(Asset))}
|
|
assert {r.decision for r in session.scalars(select(SafetyReview))} == {"sfw"}
|
|
occurrences = [
|
|
row.reason
|
|
for row in session.scalars(
|
|
select(AssetPath).where(AssetPath.asset_id == sorted(ids.values())[0])
|
|
)
|
|
]
|
|
assert "restore" in occurrences
|
|
|
|
|
|
def test_restore_never_overwrites_a_collision(tmp_path):
|
|
config, sf, lib, archive = _env(tmp_path)
|
|
location, ids = _archived(sf, config, lib, archive, seeds=(1,))
|
|
asset_id = next(iter(ids.values()))
|
|
with sf() as session:
|
|
archive_path = session.get(Asset, asset_id).archive_path
|
|
occupied = lib / archive_path
|
|
occupied.parent.mkdir(parents=True, exist_ok=True)
|
|
occupied.write_bytes(b"a different photo already lives here")
|
|
before = occupied.read_bytes()
|
|
|
|
service, plan, result = _restore(sf, config, location["id"])
|
|
assert result["failed"] == 0
|
|
|
|
assert occupied.read_bytes() == before # untouched
|
|
restored = _assets(sf)[asset_id].current_path
|
|
assert restored != str(occupied)
|
|
assert "(restored)" in restored
|
|
assert sha256_file(restored) == sha256_file(archive / archive_path)
|
|
|
|
|
|
def test_apply_refuses_a_destination_taken_after_planning(tmp_path):
|
|
config, sf, lib, archive = _env(tmp_path)
|
|
location, ids = _archived(sf, config, lib, archive, seeds=(1,))
|
|
service = RestoreService(sf, config=config)
|
|
token = service.preflight(location["id"])["token"]
|
|
plan = service.create(location["id"], None, token=token)
|
|
|
|
# Someone drops a file exactly where the plan intends to publish.
|
|
destination = plan["operations"][0]["destination_path"]
|
|
from pathlib import Path
|
|
|
|
Path(destination).parent.mkdir(parents=True, exist_ok=True)
|
|
Path(destination).write_bytes(b"squatter")
|
|
|
|
result = service.apply(plan["id"])
|
|
assert result["failed"] == 1
|
|
operation = service.journal.operations(plan["id"])[0]
|
|
assert operation["journal_state"] == ArchiveState.FAILED
|
|
assert operation["error_code"] == "destination_exists"
|
|
assert Path(destination).read_bytes() == b"squatter"
|
|
assert _assets(sf)[next(iter(ids.values()))].availability_state == availability.ARCHIVED_ONLINE
|
|
|
|
|
|
def test_changed_archive_bytes_mark_the_asset_divergent(tmp_path):
|
|
config, sf, lib, archive = _env(tmp_path)
|
|
location, ids = _archived(sf, config, lib, archive, seeds=(1,))
|
|
asset_id = next(iter(ids.values()))
|
|
service = RestoreService(sf, config=config)
|
|
token = service.preflight(location["id"])["token"]
|
|
plan = service.create(location["id"], None, token=token)
|
|
|
|
# The medium's copy is edited after the plan was approved.
|
|
with sf() as session:
|
|
archived_file = archive / session.get(Asset, asset_id).archive_path
|
|
archived_file.write_bytes(b"edited on the shelf")
|
|
|
|
result = service.apply(plan["id"])
|
|
assert result["failed"] == 1
|
|
operation = service.journal.operations(plan["id"])[0]
|
|
assert operation["error_code"] == "bytes_changed"
|
|
asset = _assets(sf)[asset_id]
|
|
assert asset.archive_divergent_at is not None # durable divergence
|
|
assert asset.availability_state == availability.ARCHIVED_ONLINE
|
|
assert asset.current_path is None # nothing was published
|
|
|
|
|
|
# ── interruption and idempotency ─────────────────────────────────────────────
|
|
|
|
|
|
def test_interrupted_before_publishing_is_resumable(tmp_path):
|
|
config, sf, lib, archive = _env(tmp_path)
|
|
location, ids = _archived(sf, config, lib, archive, seeds=(1,))
|
|
service = RestoreService(sf, config=config)
|
|
token = service.preflight(location["id"])["token"]
|
|
plan = service.create(location["id"], None, token=token)
|
|
operation = service.journal.operations(plan["id"])[0]
|
|
|
|
# Model a kill right after the intent was written: nothing published yet.
|
|
service.journal.begin(operation["id"], worker_id="killed", fencing_token=1)
|
|
|
|
status = service.recovery_status()
|
|
assert status["operations"][0]["classification"] == "resumable"
|
|
assert service.recover() == {"resumed": 1, "completed": 0, "manual": 0}
|
|
assert service.journal.operations(plan["id"])[0]["journal_state"] == ArchiveState.PLANNED
|
|
|
|
result = service.apply(plan["id"])
|
|
assert result["failed"] == 0
|
|
assert _assets(sf)[next(iter(ids.values()))].availability_state == availability.ACTIVE
|
|
|
|
|
|
def test_interrupted_after_publishing_is_finished_by_recovery(tmp_path):
|
|
config, sf, lib, archive = _env(tmp_path)
|
|
location, ids = _archived(sf, config, lib, archive, seeds=(1,))
|
|
asset_id = next(iter(ids.values()))
|
|
service = RestoreService(sf, config=config)
|
|
token = service.preflight(location["id"])["token"]
|
|
plan = service.create(location["id"], None, token=token)
|
|
operation = service.journal.operations(plan["id"])[0]
|
|
|
|
# Model a kill between the published copy and the database update.
|
|
from pathlib import Path
|
|
|
|
destination = Path(operation["destination_path"])
|
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(operation["source_path"], destination)
|
|
service.journal.begin(operation["id"], worker_id="killed", fencing_token=1)
|
|
service.journal.transition(operation["id"], ArchiveState.VERIFIED, fencing_token=1)
|
|
|
|
assert service.recovery_status()["operations"][0]["classification"] == "forward"
|
|
assert service.recover()["completed"] == 1
|
|
asset = _assets(sf)[asset_id]
|
|
assert asset.availability_state == availability.ACTIVE
|
|
assert asset.current_path == str(destination)
|
|
|
|
# Repeated recovery and a repeated apply converge on the same state.
|
|
assert service.recover() == {"resumed": 0, "completed": 0, "manual": 0}
|
|
again = service.apply(plan["id"])
|
|
assert (again["skipped"], again["failed"]) == (1, 0)
|
|
assert _assets(sf)[asset_id].current_path == str(destination)
|
|
|
|
|
|
def test_restored_state_survives_restart_and_rescan(tmp_path):
|
|
config, sf, lib, archive = _env(tmp_path)
|
|
location, ids = _archived(sf, config, lib, archive)
|
|
_restore(sf, config, location["id"])
|
|
|
|
restarted = create_session_factory(create_db_engine(config.database_url))
|
|
InventoryService(restarted).scan(lib)
|
|
|
|
assets = _assets(restarted)
|
|
assert set(assets) == set(ids.values()) # no new identities from the rescan
|
|
for asset in assets.values():
|
|
assert asset.availability_state == availability.ACTIVE
|
|
assert asset.missing_at is None
|
|
|
|
# Nothing is archived at that location any more, so there is nothing to restore.
|
|
again = RestoreService(restarted, config=config).preflight(location["id"])
|
|
assert _codes(again) == {"empty_scope"}
|
|
|
|
|
|
def test_restore_api_round_trip(tmp_path):
|
|
config, sf, lib, archive = _env(tmp_path)
|
|
location, ids = _archived(sf, config, lib, archive, seeds=(1,))
|
|
|
|
with TestClient(create_app(config)) as client:
|
|
report = client.post(
|
|
"/api/v1/restore-preflight", json={"location_id": location["id"]}
|
|
).json()
|
|
assert report["state"] == "ready"
|
|
|
|
stale = client.post(
|
|
"/api/v1/restore-plans",
|
|
json={"location_id": location["id"], "token": "r1:not-the-token"},
|
|
)
|
|
assert stale.status_code == 409
|
|
|
|
created = client.post(
|
|
"/api/v1/restore-plans",
|
|
json={"location_id": location["id"], "token": report["token"]},
|
|
)
|
|
assert created.status_code == 201
|
|
plan_id = created.json()["id"]
|
|
assert created.json()["direction"] == "restore"
|
|
|
|
# The plan is visible and applying it queues work on the archiver lane.
|
|
assert client.get(f"/api/v1/restore-plans/{plan_id}").status_code == 200
|
|
queued = client.post(f"/api/v1/restore-plans/{plan_id}/apply")
|
|
assert queued.status_code == 200
|
|
assert queued.json()["job"]["job_type"] == "restore_plan"
|
|
assert queued.json()["job"]["lock_key"] == "archive"
|
|
assert client.get("/api/v1/restore-recovery").json()["manual"] == []
|
|
|
|
assert _assets(sf)[next(iter(ids.values()))].availability_state == (
|
|
availability.ARCHIVED_ONLINE # the worker, not the request, does the work
|
|
)
|