593 lines
22 KiB
Python
593 lines
22 KiB
Python
"""Archive destinations and preflight (US06-01).
|
|
|
|
Archive is the only stage that removes originals, so every case here asks the same
|
|
question: would this preflight let an album leave active storage when it should
|
|
not? The destinations are real directories on real filesystems — mounted, missing,
|
|
swapped for another medium, read-only, or full — and preflight itself must stay
|
|
non-destructive: the library snapshot is asserted unchanged.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import stat
|
|
import uuid
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
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.jobs.domain_handlers import ARCHIVE_LOCK, UPLOAD_LOCK
|
|
from photo_pipeline.models import (
|
|
Asset,
|
|
RenameOperation,
|
|
RenamePlan,
|
|
UploadBatch,
|
|
UploadItem,
|
|
)
|
|
from photo_pipeline.services.archives import MARKER_NAME, ArchiveError, ArchiveService
|
|
from photo_pipeline.services.hashing import sha256_file
|
|
from photo_pipeline.services.jobs import JobService
|
|
|
|
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, *, reserve=0):
|
|
(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": str(reserve),
|
|
}
|
|
)
|
|
run_migrations(config.database_url)
|
|
return config, create_session_factory(create_db_engine(config.database_url)), lib, archive
|
|
|
|
|
|
def _album(
|
|
sf,
|
|
lib,
|
|
album="rome",
|
|
names=("a.jpg", "b.jpg"),
|
|
*,
|
|
uploaded=True,
|
|
outcome="uploaded",
|
|
outcome_state="verified",
|
|
stale_bytes=False,
|
|
):
|
|
"""A real album folder whose assets carry their upload evidence."""
|
|
folder = lib / album
|
|
folder.mkdir(parents=True, exist_ok=True)
|
|
ids = []
|
|
with sf() as session:
|
|
batch_id = str(uuid.uuid4())
|
|
if uploaded:
|
|
session.add(
|
|
UploadBatch(
|
|
id=batch_id,
|
|
album=album,
|
|
folder=str(folder),
|
|
album_name=album,
|
|
state="succeeded",
|
|
preflight_token="v1:test",
|
|
outcome_state=outcome_state,
|
|
stale_bytes=stale_bytes,
|
|
created_at=NOW,
|
|
)
|
|
)
|
|
for name in names:
|
|
path = folder / name
|
|
path.write_bytes(name.encode() * 16)
|
|
asset_id = str(uuid.uuid4())
|
|
ids.append(asset_id)
|
|
session.add(
|
|
Asset(
|
|
id=asset_id,
|
|
original_path=str(path),
|
|
current_path=str(path),
|
|
discovered_at=NOW,
|
|
hash_version=1,
|
|
byte_size=path.stat().st_size,
|
|
current_sha256=sha256_file(path),
|
|
)
|
|
)
|
|
if uploaded:
|
|
session.add(
|
|
UploadItem(
|
|
batch_id=batch_id,
|
|
asset_id=asset_id,
|
|
path=str(path),
|
|
sha256=sha256_file(path),
|
|
sha1="0" * 40,
|
|
state="sent",
|
|
outcome=outcome,
|
|
)
|
|
)
|
|
session.commit()
|
|
return folder, ids
|
|
|
|
|
|
def _service(sf, config):
|
|
return ArchiveService(sf, config=config)
|
|
|
|
|
|
def _location(sf, config, archive, name="external"):
|
|
return _service(sf, config).register(name, str(archive))
|
|
|
|
|
|
def _snapshot(lib):
|
|
return {
|
|
str(p.relative_to(lib)): (p.read_bytes() if p.is_file() else None)
|
|
for p in sorted(lib.rglob("*"))
|
|
}
|
|
|
|
|
|
def _codes(report):
|
|
return (
|
|
{issue["code"] for issue in report["blockers"]}
|
|
| {issue["code"] for album in report["albums"] for issue in album["blockers"]}
|
|
| {
|
|
issue["code"]
|
|
for album in report["albums"]
|
|
for asset in album["assets"]
|
|
for issue in asset["blockers"]
|
|
}
|
|
)
|
|
|
|
|
|
# ── locations ────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_registering_a_location_stamps_the_medium_with_its_identity(tmp_path):
|
|
config, sf, _, archive = _env(tmp_path)
|
|
|
|
location = _location(sf, config, archive)
|
|
|
|
marker = json.loads((archive / MARKER_NAME).read_text())
|
|
assert marker["media_id"] == location["media_id"]
|
|
assert location["state"] == "online" and location["writable"] is True
|
|
assert location["root"] == str(archive.resolve())
|
|
listed = _service(sf, config).locations()
|
|
assert [(row["id"], row["media_id"], row["state"]) for row in listed] == [
|
|
(location["id"], location["media_id"], "online")
|
|
]
|
|
|
|
|
|
def test_a_second_location_cannot_claim_the_same_medium(tmp_path):
|
|
config, sf, _, archive = _env(tmp_path)
|
|
_location(sf, config, archive)
|
|
|
|
with pytest.raises(ArchiveError) as error:
|
|
_location(sf, config, archive, name="second")
|
|
|
|
assert error.value.code == "already_registered"
|
|
|
|
|
|
@pytest.mark.parametrize("inside", ["", "sub"])
|
|
def test_a_destination_inside_the_library_is_refused(tmp_path, inside):
|
|
"""The library may never archive into itself: the 'reclaimed' bytes would still
|
|
be in the active tree, and a later scan would rediscover them."""
|
|
config, sf, lib, _ = _env(tmp_path)
|
|
root = lib / inside if inside else lib
|
|
root.mkdir(exist_ok=True)
|
|
|
|
with pytest.raises(ArchiveError) as error:
|
|
_service(sf, config).register("bad", str(root))
|
|
|
|
assert error.value.code == "unsafe_destination"
|
|
|
|
|
|
def test_an_ignored_destination_is_refused(tmp_path):
|
|
config, sf, _, _ = _env(tmp_path)
|
|
root = tmp_path / "_IGNORE" / "archive"
|
|
root.mkdir(parents=True)
|
|
|
|
with pytest.raises(ArchiveError) as error:
|
|
_service(sf, config).register("ignored", str(root))
|
|
|
|
assert error.value.code == "unsafe_destination"
|
|
|
|
|
|
def test_listing_reports_an_unmounted_medium_as_offline(tmp_path):
|
|
config, sf, _, archive = _env(tmp_path)
|
|
_location(sf, config, archive)
|
|
(archive / MARKER_NAME).unlink()
|
|
|
|
assert [row["state"] for row in _service(sf, config).locations()] == ["offline"]
|
|
|
|
|
|
# ── happy path ───────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_ready_preflight_previews_scope_method_and_reclaimable_bytes(tmp_path):
|
|
config, sf, lib, archive = _env(tmp_path)
|
|
folder, ids = _album(sf, lib)
|
|
location = _location(sf, config, archive)
|
|
before = _snapshot(lib)
|
|
|
|
report = _service(sf, config).preflight(location["id"])
|
|
|
|
assert report["state"] == "ready" and report["blockers"] == []
|
|
album = report["albums"][0]
|
|
assert album["album"] == "rome" and album["folder"] == str(folder)
|
|
assert album["destination"] == str(archive.resolve() / "rome")
|
|
assert album["transfer_method"] in ("move", "copy_verify_remove")
|
|
assert album["reclaimable_bytes"] == sum(p.stat().st_size for p in folder.iterdir())
|
|
assert sorted(a["asset_id"] for a in album["assets"]) == sorted(ids)
|
|
assert report["totals"]["bytes"] == album["reclaimable_bytes"]
|
|
assert report["capacity"]["sufficient"] is True
|
|
# Both must be proven by writing, not assumed.
|
|
assert report["backup"]["ok"] is True and report["backup"]["bytes"] > 0
|
|
assert report["manifest"]["ok"] is True
|
|
assert report["token"].startswith("v1:")
|
|
assert _snapshot(lib) == before, "preflight must not touch the library"
|
|
assert not list(archive.glob("*probe*")), "probe files must be cleaned up"
|
|
|
|
|
|
def test_same_filesystem_destination_is_previewed_as_a_move(tmp_path):
|
|
config, sf, lib, archive = _env(tmp_path)
|
|
_album(sf, lib)
|
|
location = _location(sf, config, archive)
|
|
|
|
report = _service(sf, config).preflight(location["id"])
|
|
|
|
# tmp_path is one filesystem, so this is the same-filesystem case by construction.
|
|
assert report["albums"][0]["transfer_method"] == "move"
|
|
|
|
|
|
def test_scoping_to_one_album_excludes_the_others(tmp_path):
|
|
config, sf, lib, archive = _env(tmp_path)
|
|
_album(sf, lib, "rome")
|
|
_album(sf, lib, "paris", names=("c.jpg",))
|
|
location = _location(sf, config, archive)
|
|
|
|
report = _service(sf, config).preflight(location["id"], ["paris"])
|
|
|
|
assert [album["album"] for album in report["albums"]] == ["paris"]
|
|
assert report["totals"]["assets"] == 1
|
|
|
|
|
|
def test_unknown_album_and_unknown_location_are_refused(tmp_path):
|
|
config, sf, lib, archive = _env(tmp_path)
|
|
_album(sf, lib)
|
|
location = _location(sf, config, archive)
|
|
service = _service(sf, config)
|
|
|
|
with pytest.raises(ArchiveError) as unknown_album:
|
|
service.preflight(location["id"], ["atlantis"])
|
|
with pytest.raises(ArchiveError) as unknown_location:
|
|
service.preflight("nope")
|
|
|
|
assert unknown_album.value.code == "unknown_album"
|
|
assert unknown_location.value.code == "unknown_location"
|
|
|
|
|
|
def test_empty_scope_is_a_blocker(tmp_path):
|
|
config, sf, _, archive = _env(tmp_path)
|
|
location = _location(sf, config, archive)
|
|
|
|
report = _service(sf, config).preflight(location["id"])
|
|
|
|
assert report["state"] == "blocked" and "empty_scope" in _codes(report)
|
|
|
|
|
|
# ── destination ──────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_offline_medium_blocks(tmp_path):
|
|
config, sf, lib, archive = _env(tmp_path)
|
|
_album(sf, lib)
|
|
location = _location(sf, config, archive)
|
|
(archive / MARKER_NAME).unlink() # the disk went away
|
|
|
|
report = _service(sf, config).preflight(location["id"])
|
|
|
|
assert report["state"] == "blocked" and "location_offline" in _codes(report)
|
|
assert report["location"]["state"] == "offline"
|
|
|
|
|
|
def test_a_different_medium_at_the_same_mountpoint_blocks(tmp_path):
|
|
"""The mountpoint is right, the disk is not — never write the archive here."""
|
|
config, sf, lib, archive = _env(tmp_path)
|
|
_album(sf, lib)
|
|
location = _location(sf, config, archive)
|
|
(archive / MARKER_NAME).write_text(json.dumps({"media_id": "some-other-disk"}))
|
|
|
|
report = _service(sf, config).preflight(location["id"])
|
|
|
|
assert "wrong_volume" in _codes(report)
|
|
assert report["location"]["state"] == "wrong_volume"
|
|
|
|
|
|
def test_read_only_destination_blocks_and_cannot_write_the_manifest(tmp_path):
|
|
config, sf, lib, archive = _env(tmp_path)
|
|
_album(sf, lib)
|
|
location = _location(sf, config, archive)
|
|
mode = archive.stat().st_mode
|
|
archive.chmod(mode & ~stat.S_IWUSR & ~stat.S_IWGRP & ~stat.S_IWOTH)
|
|
try:
|
|
report = _service(sf, config).preflight(location["id"])
|
|
finally:
|
|
archive.chmod(mode)
|
|
|
|
assert {"destination_not_writable", "manifest_unwritable"} <= _codes(report)
|
|
assert report["manifest"]["ok"] is False
|
|
|
|
|
|
@pytest.mark.skipif(os.geteuid() == 0, reason="root ignores directory permissions")
|
|
def test_read_only_destination_is_detected_by_a_real_write(tmp_path):
|
|
config, sf, lib, archive = _env(tmp_path)
|
|
_album(sf, lib)
|
|
location = _location(sf, config, archive)
|
|
mode = archive.stat().st_mode
|
|
archive.chmod(stat.S_IRUSR | stat.S_IXUSR)
|
|
try:
|
|
report = _service(sf, config).preflight(location["id"])
|
|
finally:
|
|
archive.chmod(mode)
|
|
|
|
assert report["location"]["writable"] is False
|
|
|
|
|
|
def test_insufficient_capacity_blocks(tmp_path):
|
|
"""The reserve is what stops an archive from filling its own destination."""
|
|
config, sf, lib, archive = _env(tmp_path, reserve=10**15)
|
|
_album(sf, lib)
|
|
location = _location(sf, config, archive)
|
|
|
|
report = _service(sf, config).preflight(location["id"])
|
|
|
|
assert report["state"] == "blocked" and "insufficient_capacity" in _codes(report)
|
|
assert report["capacity"]["sufficient"] is False
|
|
assert report["capacity"]["reserve_bytes"] == 10**15
|
|
|
|
|
|
def test_an_occupied_destination_blocks_that_album(tmp_path):
|
|
config, sf, lib, archive = _env(tmp_path)
|
|
_album(sf, lib)
|
|
location = _location(sf, config, archive)
|
|
(archive / "rome").mkdir()
|
|
(archive / "rome" / "a.jpg").write_bytes(b"something already here")
|
|
|
|
report = _service(sf, config).preflight(location["id"])
|
|
|
|
assert "destination_collision" in _codes(report)
|
|
assert report["albums"][0]["state"] == "blocked"
|
|
|
|
|
|
def test_a_destination_moved_into_the_library_blocks_even_though_it_registered(tmp_path):
|
|
"""Registration validated the root once; preflight validates it again, because a
|
|
mountpoint can be moved after the fact."""
|
|
config, sf, lib, archive = _env(tmp_path)
|
|
_album(sf, lib)
|
|
location = _location(sf, config, archive)
|
|
with sf() as session:
|
|
from photo_pipeline.models import ArchiveLocation
|
|
|
|
session.get(ArchiveLocation, location["id"]).root = str(lib / "inside")
|
|
session.commit()
|
|
(lib / "inside").mkdir()
|
|
(lib / "inside" / MARKER_NAME).write_text(json.dumps({"media_id": location["media_id"]}))
|
|
|
|
report = _service(sf, config).preflight(location["id"])
|
|
|
|
assert "unsafe_destination" in _codes(report)
|
|
|
|
|
|
# ── source readiness ─────────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"kwargs,code",
|
|
[
|
|
({"uploaded": False}, "upload_unverified"),
|
|
({"outcome_state": "requires_verification"}, "upload_unverified"),
|
|
({"outcome": "failed"}, "upload_unverified"),
|
|
({"outcome": "skipped"}, "upload_unverified"),
|
|
({"stale_bytes": True}, "upload_unverified"),
|
|
],
|
|
)
|
|
def test_an_unverified_upload_blocks_the_album(tmp_path, kwargs, code):
|
|
config, sf, lib, archive = _env(tmp_path)
|
|
_album(sf, lib, **kwargs)
|
|
location = _location(sf, config, archive)
|
|
|
|
report = _service(sf, config).preflight(location["id"])
|
|
|
|
assert report["state"] == "blocked" and code in _codes(report)
|
|
assert report["albums"][0]["blocked_count"] == 2
|
|
|
|
|
|
@pytest.mark.parametrize("outcome", ["upgraded", "duplicate"])
|
|
def test_upgraded_and_duplicate_uploads_are_evidence_enough(tmp_path, outcome):
|
|
"""Immich already holds these exact bytes; that is what archiving requires."""
|
|
config, sf, lib, archive = _env(tmp_path)
|
|
_album(sf, lib, outcome=outcome)
|
|
location = _location(sf, config, archive)
|
|
|
|
assert _service(sf, config).preflight(location["id"])["state"] == "ready"
|
|
|
|
|
|
def test_bytes_changed_since_upload_block_the_album(tmp_path):
|
|
config, sf, lib, archive = _env(tmp_path)
|
|
folder, _ = _album(sf, lib)
|
|
location = _location(sf, config, archive)
|
|
(folder / "a.jpg").write_bytes(b"edited after the upload")
|
|
|
|
report = _service(sf, config).preflight(location["id"])
|
|
|
|
assert {"bytes_changed", "partial_scope"} <= _codes(report)
|
|
assert report["albums"][0]["blocked_count"] == 1
|
|
|
|
|
|
def test_a_missing_source_file_blocks_the_album(tmp_path):
|
|
config, sf, lib, archive = _env(tmp_path)
|
|
folder, _ = _album(sf, lib)
|
|
location = _location(sf, config, archive)
|
|
(folder / "a.jpg").unlink()
|
|
|
|
assert "file_missing" in _codes(_service(sf, config).preflight(location["id"]))
|
|
|
|
|
|
# ── leases ───────────────────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.mark.parametrize("lock", [UPLOAD_LOCK, ARCHIVE_LOCK])
|
|
def test_a_held_lease_blocks_archiving(tmp_path, lock):
|
|
config, sf, lib, archive = _env(tmp_path)
|
|
_album(sf, lib)
|
|
location = _location(sf, config, archive)
|
|
JobService(sf).enqueue("upload_batch", lock=lock, items=["x"])
|
|
|
|
report = _service(sf, config).preflight(location["id"])
|
|
|
|
assert report["state"] == "blocked" and "lock_conflict" in _codes(report)
|
|
|
|
|
|
def test_a_half_applied_rename_blocks_archiving(tmp_path):
|
|
config, sf, lib, archive = _env(tmp_path)
|
|
folder, _ = _album(sf, lib)
|
|
location = _location(sf, config, archive)
|
|
with sf() as session:
|
|
plan_id = str(uuid.uuid4())
|
|
session.add(RenamePlan(id=plan_id, state="applying", operation_count=1))
|
|
session.flush()
|
|
session.add(
|
|
RenameOperation(
|
|
id=str(uuid.uuid4()),
|
|
plan_id=plan_id,
|
|
sequence=0,
|
|
operation="move_folder",
|
|
source_path=str(folder),
|
|
destination_path=str(lib / "2019 Rome"),
|
|
journal_state="moving",
|
|
)
|
|
)
|
|
session.commit()
|
|
|
|
report = _service(sf, config).preflight(location["id"])
|
|
|
|
assert report["state"] == "blocked" and "rename_pending" in _codes(report)
|
|
|
|
|
|
# ── token ────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_token_is_stable_while_nothing_relevant_changes(tmp_path):
|
|
config, sf, lib, archive = _env(tmp_path)
|
|
_album(sf, lib)
|
|
location = _location(sf, config, archive)
|
|
service = _service(sf, config)
|
|
|
|
first = service.preflight(location["id"])["token"]
|
|
|
|
assert service.preflight(location["id"])["token"] == first
|
|
assert service.verify_token(first, location["id"]) is True
|
|
|
|
|
|
def test_an_edited_source_makes_the_token_stale(tmp_path):
|
|
config, sf, lib, archive = _env(tmp_path)
|
|
folder, _ = _album(sf, lib)
|
|
location = _location(sf, config, archive)
|
|
service = _service(sf, config)
|
|
token = service.preflight(location["id"])["token"]
|
|
|
|
(folder / "b.jpg").write_bytes(b"edited outside the app")
|
|
|
|
assert service.verify_token(token, location["id"]) is False
|
|
|
|
|
|
def test_a_changed_destination_makes_the_token_stale(tmp_path):
|
|
config, sf, lib, archive = _env(tmp_path)
|
|
_album(sf, lib)
|
|
location = _location(sf, config, archive)
|
|
service = _service(sf, config)
|
|
token = service.preflight(location["id"])["token"]
|
|
|
|
(archive / "rome").mkdir()
|
|
(archive / "rome" / "a.jpg").write_bytes(b"appeared after approval")
|
|
|
|
assert service.verify_token(token, location["id"]) is False
|
|
|
|
|
|
def test_a_token_from_another_scope_or_medium_is_rejected(tmp_path):
|
|
config, sf, lib, archive = _env(tmp_path)
|
|
_album(sf, lib, "rome")
|
|
_album(sf, lib, "paris", names=("c.jpg",))
|
|
other = tmp_path / "archive2"
|
|
other.mkdir()
|
|
location = _location(sf, config, archive)
|
|
second = _location(sf, config, other, name="second")
|
|
service = _service(sf, config)
|
|
|
|
rome = service.preflight(location["id"], ["rome"])["token"]
|
|
|
|
assert service.verify_token(rome, location["id"], ["paris"]) is False
|
|
assert service.verify_token(rome, second["id"], ["rome"]) is False
|
|
assert service.verify_token("v1:not-a-real-token", location["id"]) is False
|
|
assert service.verify_token("", location["id"]) is False
|
|
|
|
|
|
def test_the_token_survives_free_space_and_timestamp_drift(tmp_path):
|
|
"""Free space changes constantly on a live disk; a token that expired on every
|
|
byte written elsewhere would train users to ignore it."""
|
|
config, sf, lib, archive = _env(tmp_path)
|
|
_album(sf, lib)
|
|
location = _location(sf, config, archive)
|
|
service = _service(sf, config)
|
|
token = service.preflight(location["id"])["token"]
|
|
|
|
(tmp_path / "unrelated.bin").write_bytes(b"0" * 100_000)
|
|
with sf() as session:
|
|
from photo_pipeline.models import ArchiveLocation
|
|
|
|
session.get(ArchiveLocation, location["id"]).last_seen_at = NOW - timedelta(days=5)
|
|
session.commit()
|
|
|
|
assert service.verify_token(token, location["id"]) is True
|
|
|
|
|
|
# ── API surface ──────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_api_registers_a_location_and_returns_a_preflight_report(tmp_path):
|
|
config, sf, lib, archive = _env(tmp_path)
|
|
_album(sf, lib)
|
|
|
|
with TestClient(create_app(config)) as client:
|
|
created = client.post(
|
|
"/api/v1/archive-locations", json={"name": "external", "root": str(archive)}
|
|
)
|
|
listed = client.get("/api/v1/archive-locations")
|
|
report = client.post(
|
|
"/api/v1/archive-preflight", json={"location_id": created.json()["id"]}
|
|
)
|
|
|
|
assert created.status_code == 201
|
|
assert [row["name"] for row in listed.json()["locations"]] == ["external"]
|
|
assert report.status_code == 200
|
|
assert report.json()["state"] == "ready" and report.json()["token"].startswith("v1:")
|
|
|
|
|
|
def test_api_rejects_an_unknown_location_and_an_unsafe_root(tmp_path):
|
|
config, sf, lib, _ = _env(tmp_path)
|
|
|
|
with TestClient(create_app(config)) as client:
|
|
unknown = client.post("/api/v1/archive-preflight", json={"location_id": "nope"})
|
|
unsafe = client.post("/api/v1/archive-locations", json={"name": "bad", "root": str(lib)})
|
|
|
|
assert unknown.status_code == 404 and unknown.json()["error"]["code"] == "unknown_location"
|
|
assert unsafe.status_code == 422 and unsafe.json()["error"]["code"] == "unsafe_destination"
|