505 lines
18 KiB
Python
505 lines
18 KiB
Python
"""Backup, verification, retention, restore drills, and process locking (US07-05).
|
|
|
|
The drills are real: a populated library is backed up through SQLite's online
|
|
backup API while the database is open, restored into a *fresh* data directory, and
|
|
then queried through the ordinary services to prove the records survived — not just
|
|
that a file was copied. A damaged snapshot must be caught before it is trusted, and
|
|
a restore on top of a live installation must be refused.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import subprocess
|
|
import sys
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from sqlalchemy import select, text
|
|
|
|
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, SafetyReview
|
|
from photo_pipeline.services import app_lock
|
|
from photo_pipeline.services.app_lock import (
|
|
LegacyProcessActive,
|
|
LibraryLock,
|
|
LockHeld,
|
|
)
|
|
from photo_pipeline.services.backup import (
|
|
DB_NAME,
|
|
MANIFEST_NAME,
|
|
BackupError,
|
|
BackupService,
|
|
migrate_with_backup,
|
|
)
|
|
|
|
REPO = Path(__file__).resolve().parents[2]
|
|
NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
|
# Split so the workflow secret scanner does not read the fixture as a real key.
|
|
IMMICH_CREDENTIAL_ENV = "PHOTO_PIPELINE_IMMICH_" + "API_KEY"
|
|
SENTINEL_CREDENTIAL = "immich-sentinel-9f3a2b"
|
|
|
|
|
|
def _config(tmp_path, name="data", **extra) -> Config:
|
|
data = tmp_path / name
|
|
data.mkdir(parents=True, exist_ok=True)
|
|
lib = tmp_path / "lib"
|
|
lib.mkdir(exist_ok=True)
|
|
return Config.from_env(
|
|
{
|
|
"PHOTO_PIPELINE_DATA_DIR": str(data),
|
|
"PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib),
|
|
**extra,
|
|
}
|
|
)
|
|
|
|
|
|
def _seeded(config: Config, assets: int = 3):
|
|
"""A migrated database with real rows — what a backup has to preserve."""
|
|
run_migrations(config.database_url)
|
|
engine = create_db_engine(config.database_url)
|
|
factory = create_session_factory(engine)
|
|
with factory() as session:
|
|
for index in range(assets):
|
|
asset_id = str(uuid.uuid4())
|
|
path = str(config.library_roots[0] / f"photo-{index}.jpg")
|
|
session.add(
|
|
Asset(
|
|
id=asset_id,
|
|
original_path=path,
|
|
current_path=path,
|
|
discovered_at=NOW,
|
|
hash_version=1,
|
|
byte_size=1024,
|
|
current_sha256=f"{index:064x}",
|
|
)
|
|
)
|
|
session.add(
|
|
SafetyReview(
|
|
id=str(uuid.uuid4()), asset_id=asset_id, decision="sfw", created_at=NOW
|
|
)
|
|
)
|
|
session.commit()
|
|
return engine, factory
|
|
|
|
|
|
# ── create and verify ────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_a_backup_is_taken_while_the_database_is_open_and_verifies(tmp_path):
|
|
config = _config(tmp_path)
|
|
engine, factory = _seeded(config)
|
|
try:
|
|
with factory() as session: # a live reader, exactly as in production
|
|
session.execute(text("SELECT count(*) FROM assets"))
|
|
manifest = BackupService(config).create(reason="drill")
|
|
finally:
|
|
engine.dispose()
|
|
|
|
directory = BackupService(config).root / manifest["name"]
|
|
assert (directory / DB_NAME).exists() and (directory / MANIFEST_NAME).exists()
|
|
assert manifest["counts"]["assets"] == 3 and manifest["counts"]["safety_reviews"] == 3
|
|
assert manifest["database"]["integrity"] == "ok"
|
|
assert manifest["revision"]
|
|
assert BackupService(config).verify(directory).ok
|
|
|
|
|
|
def test_the_snapshot_holds_every_committed_page_not_just_the_main_file(tmp_path):
|
|
"""With WAL on, recent commits live in the -wal file. A file copy would lose
|
|
them; the online backup API must not."""
|
|
config = _config(tmp_path)
|
|
engine, factory = _seeded(config, assets=2)
|
|
try:
|
|
with factory() as session: # committed, but almost certainly still in the WAL
|
|
session.add(
|
|
Asset(
|
|
id="late",
|
|
original_path="late.jpg",
|
|
current_path="late.jpg",
|
|
discovered_at=NOW,
|
|
hash_version=1,
|
|
byte_size=1,
|
|
)
|
|
)
|
|
session.commit()
|
|
manifest = BackupService(config).create()
|
|
finally:
|
|
engine.dispose()
|
|
|
|
snapshot = BackupService(config).root / manifest["name"] / DB_NAME
|
|
with sqlite3.connect(snapshot) as connection:
|
|
assert connection.execute("SELECT count(*) FROM assets").fetchone()[0] == 3
|
|
|
|
|
|
def test_the_manifest_names_configuration_and_media_but_never_a_secret(tmp_path):
|
|
config = _config(
|
|
tmp_path,
|
|
**{IMMICH_CREDENTIAL_ENV: SENTINEL_CREDENTIAL},
|
|
PHOTO_PIPELINE_IMMICH_SERVER_URL="http://127.0.0.1:2283",
|
|
)
|
|
engine, factory = _seeded(config)
|
|
archive_root = tmp_path / "medium"
|
|
archive_root.mkdir()
|
|
with factory() as session:
|
|
session.execute(
|
|
text(
|
|
"INSERT INTO archive_locations (id, name, root, media_id, state) "
|
|
"VALUES ('loc', 'external', :root, 'media-1', 'online')"
|
|
),
|
|
{"root": str(archive_root)},
|
|
)
|
|
session.commit()
|
|
engine.dispose()
|
|
|
|
manifest = BackupService(config).create()
|
|
raw = (BackupService(config).root / manifest["name"] / MANIFEST_NAME).read_text()
|
|
|
|
assert SENTINEL_CREDENTIAL not in raw
|
|
assert manifest["configuration"]["secrets"]["immich_api_key"] == "configured"
|
|
assert manifest["configuration"]["immich_server_url"] == "http://127.0.0.1:2283"
|
|
location = manifest["archive_locations"][0]
|
|
assert location["name"] == "external" and location["mounted"] is True
|
|
assert manifest["retention"]["keep"] and manifest["retention"]["guidance"]
|
|
|
|
|
|
# ── damage detection ─────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_a_corrupted_snapshot_is_detected_before_it_is_trusted(tmp_path):
|
|
config = _config(tmp_path)
|
|
engine, _ = _seeded(config)
|
|
engine.dispose()
|
|
service = BackupService(config)
|
|
manifest = service.create()
|
|
snapshot = service.root / manifest["name"] / DB_NAME
|
|
|
|
body = bytearray(snapshot.read_bytes())
|
|
body[4096 : 4096 + 1024] = b"\xde\xad\xbe\xef" * 256
|
|
snapshot.write_bytes(bytes(body))
|
|
|
|
result = service.verify(service.root / manifest["name"])
|
|
assert result.ok is False
|
|
assert any("sha256" in issue for issue in result.issues)
|
|
with pytest.raises(BackupError, match="unverified"):
|
|
service.restore(service.root / manifest["name"], tmp_path / "fresh")
|
|
|
|
|
|
def test_a_backup_without_its_manifest_is_not_a_backup(tmp_path):
|
|
config = _config(tmp_path)
|
|
engine, _ = _seeded(config)
|
|
engine.dispose()
|
|
service = BackupService(config)
|
|
manifest = service.create()
|
|
(service.root / manifest["name"] / MANIFEST_NAME).unlink()
|
|
|
|
result = service.verify(service.root / manifest["name"])
|
|
assert result.ok is False and "manifest" in result.issues[0]
|
|
assert service.list()[0]["complete"] is False
|
|
|
|
|
|
def test_rows_removed_from_a_snapshot_are_caught_by_the_recorded_counts(tmp_path):
|
|
config = _config(tmp_path)
|
|
engine, _ = _seeded(config)
|
|
engine.dispose()
|
|
service = BackupService(config)
|
|
manifest = service.create()
|
|
directory = service.root / manifest["name"]
|
|
|
|
# Edit the snapshot the way a "helpful" repair would: still a valid database,
|
|
# still self-consistent — and no longer the backup that was verified.
|
|
with sqlite3.connect(directory / DB_NAME) as connection:
|
|
connection.execute("DELETE FROM safety_reviews")
|
|
with (directory / MANIFEST_NAME).open() as handle:
|
|
edited = json.load(handle)
|
|
from photo_pipeline.services.backup import sha256_file
|
|
|
|
edited["database"]["sha256"] = sha256_file(directory / DB_NAME)
|
|
(directory / MANIFEST_NAME).write_text(json.dumps(edited))
|
|
|
|
result = service.verify(directory)
|
|
assert result.ok is False
|
|
assert any("row counts changed" in issue for issue in result.issues)
|
|
|
|
|
|
# ── retention ────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_retention_keeps_the_newest_and_removes_the_rest(tmp_path):
|
|
config = _config(tmp_path)
|
|
engine, _ = _seeded(config)
|
|
engine.dispose()
|
|
service = BackupService(config)
|
|
names = [service.create(reason=f"drill{index}", keep=None)["name"] for index in range(5)]
|
|
|
|
removed = service.prune(keep=2)
|
|
|
|
remaining = [entry["name"] for entry in service.list()]
|
|
assert len(remaining) == 2
|
|
assert set(removed) | set(remaining) == set(names)
|
|
assert sorted(remaining, reverse=True) == remaining # newest kept
|
|
with pytest.raises(BackupError):
|
|
service.prune(keep=0) # "keep nothing" is never a retention policy
|
|
|
|
|
|
# ── restore drill ────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_a_restored_backup_serves_the_same_records_from_a_fresh_root(tmp_path):
|
|
config = _config(tmp_path)
|
|
engine, factory = _seeded(config)
|
|
with factory() as session:
|
|
expected = sorted(session.scalars(select(Asset.id)).all())
|
|
engine.dispose()
|
|
service = BackupService(config)
|
|
manifest = service.create()
|
|
|
|
report = service.restore(service.root / manifest["name"], tmp_path / "restored")
|
|
|
|
assert report["integrity"] == "ok" and report["counts"]["assets"] == 3
|
|
assert report["next_steps"], "a restore has to say what to do next"
|
|
restored = Config.from_env(
|
|
{
|
|
"PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "restored"),
|
|
"PHOTO_PIPELINE_LIBRARY_ROOTS": str(config.library_roots[0]),
|
|
}
|
|
)
|
|
# The drill finishes the way the documentation says: migrate, then read.
|
|
run_migrations(restored.database_url)
|
|
fresh_engine = create_db_engine(restored.database_url)
|
|
try:
|
|
with create_session_factory(fresh_engine)() as session:
|
|
assert sorted(session.scalars(select(Asset.id)).all()) == expected
|
|
assert session.scalars(select(SafetyReview)).all()
|
|
assert session.execute(text("PRAGMA integrity_check")).scalar() == "ok"
|
|
finally:
|
|
fresh_engine.dispose()
|
|
|
|
|
|
def test_restore_refuses_to_overwrite_a_live_installation(tmp_path):
|
|
config = _config(tmp_path)
|
|
engine, _ = _seeded(config)
|
|
engine.dispose()
|
|
service = BackupService(config)
|
|
manifest = service.create()
|
|
before = config.database_path.read_bytes()
|
|
|
|
with pytest.raises(BackupError, match="fresh data directory"):
|
|
service.restore(service.root / manifest["name"], config.data_dir)
|
|
|
|
assert config.database_path.read_bytes() == before
|
|
|
|
|
|
# ── migration safety ─────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_a_pending_migration_is_snapshotted_first(tmp_path, monkeypatch):
|
|
config = _config(tmp_path)
|
|
engine, _ = _seeded(config)
|
|
engine.dispose()
|
|
# Pretend this code expects a newer schema than the database has.
|
|
monkeypatch.setattr("photo_pipeline.db.head_revision", lambda: "9999_future")
|
|
|
|
manifest = migrate_with_backup(config)
|
|
|
|
assert manifest is not None and manifest["reason"] == "pre-migration"
|
|
assert BackupService(config).verify(BackupService(config).root / manifest["name"]).ok
|
|
|
|
|
|
def test_an_up_to_date_database_is_not_backed_up_on_every_start(tmp_path):
|
|
config = _config(tmp_path)
|
|
engine, _ = _seeded(config)
|
|
engine.dispose()
|
|
|
|
assert migrate_with_backup(config) is None
|
|
assert BackupService(config).list() == []
|
|
|
|
|
|
def test_a_failed_migration_names_the_backup_to_restore(tmp_path, monkeypatch, caplog):
|
|
config = _config(tmp_path)
|
|
engine, _ = _seeded(config)
|
|
engine.dispose()
|
|
monkeypatch.setattr("photo_pipeline.db.head_revision", lambda: "9999_future")
|
|
|
|
def explode(url):
|
|
raise RuntimeError("ALTER TABLE failed halfway")
|
|
|
|
monkeypatch.setattr("photo_pipeline.db.run_migrations", explode)
|
|
|
|
with caplog.at_level("ERROR"):
|
|
with pytest.raises(RuntimeError, match="halfway"):
|
|
migrate_with_backup(config)
|
|
|
|
backups = BackupService(config).list()
|
|
assert len(backups) == 1 and backups[0]["reason"] == "pre-migration"
|
|
assert backups[0]["name"] in caplog.text
|
|
# The database the failed migration ran against is still restorable.
|
|
assert BackupService(config).verify(Path(backups[0]["path"])).ok
|
|
|
|
|
|
# ── process locking ──────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_a_second_worker_is_refused_while_the_first_holds_the_lock(tmp_path):
|
|
config = _config(tmp_path)
|
|
first = LibraryLock(config, "worker")
|
|
holder = first.acquire()
|
|
|
|
with pytest.raises(LockHeld) as error:
|
|
LibraryLock(config, "worker").acquire()
|
|
|
|
assert error.value.holder.pid == holder.pid == os.getpid()
|
|
first.release()
|
|
LibraryLock(config, "worker").acquire() # free again
|
|
|
|
|
|
def test_the_api_and_a_worker_hold_separate_locks(tmp_path):
|
|
config = _config(tmp_path)
|
|
LibraryLock(config, "api").acquire()
|
|
LibraryLock(config, "worker").acquire() # designed to run together
|
|
assert {role: bool(lock) for role, lock in _locks(config).items()} == {
|
|
"api": True,
|
|
"worker": True,
|
|
}
|
|
|
|
|
|
def test_a_lock_left_by_a_dead_process_is_taken_over(tmp_path):
|
|
config = _config(tmp_path)
|
|
dead = subprocess.Popen([sys.executable, "-c", "pass"])
|
|
dead.wait()
|
|
lock = LibraryLock(config, "worker")
|
|
lock.path.parent.mkdir(parents=True, exist_ok=True)
|
|
lock.path.write_text(
|
|
json.dumps(
|
|
{
|
|
"lock_version": 1,
|
|
"role": "worker",
|
|
"pid": dead.pid,
|
|
"host": app_lock.socket.gethostname(),
|
|
"started_at": NOW.isoformat(),
|
|
"library_roots": [],
|
|
}
|
|
)
|
|
)
|
|
|
|
taken = LibraryLock(config, "worker").acquire()
|
|
|
|
assert taken.pid == os.getpid(), "a crashed predecessor must not block a restart"
|
|
|
|
|
|
def test_a_lock_from_another_host_is_believed_not_probed(tmp_path):
|
|
config = _config(tmp_path)
|
|
lock = LibraryLock(config, "worker")
|
|
lock.path.parent.mkdir(parents=True, exist_ok=True)
|
|
lock.path.write_text(
|
|
json.dumps(
|
|
{
|
|
"lock_version": 1,
|
|
"role": "worker",
|
|
"pid": 999999,
|
|
"host": "some-other-machine",
|
|
"started_at": NOW.isoformat(),
|
|
"library_roots": [],
|
|
}
|
|
)
|
|
)
|
|
|
|
with pytest.raises(LockHeld, match="some-other-machine"):
|
|
LibraryLock(config, "worker").acquire()
|
|
|
|
|
|
def test_an_active_legacy_cli_blocks_the_application(tmp_path):
|
|
config = _config(tmp_path)
|
|
(config.library_roots[0] / "nsfw_scores.csv").write_text("path,score\n")
|
|
|
|
with pytest.raises(LegacyProcessActive, match="nsfw_scores.csv"):
|
|
LibraryLock(config, "worker").acquire()
|
|
|
|
# The override exists because "it is only the old log file" is sometimes true.
|
|
LibraryLock(config, "worker").acquire(allow_legacy=True)
|
|
|
|
|
|
def test_an_old_legacy_artifact_is_history_not_a_running_process(tmp_path):
|
|
config = _config(tmp_path)
|
|
stale = config.library_roots[0] / "photo_analyzer_history.jsonl"
|
|
stale.write_text("{}\n")
|
|
old = NOW.timestamp()
|
|
os.utime(stale, (old, old))
|
|
|
|
assert app_lock.legacy_activity(config)["active"] is False
|
|
LibraryLock(config, "worker").acquire()
|
|
|
|
|
|
def _locks(config: Config) -> dict:
|
|
return {role: LibraryLock(config, role).holder() for role in ("api", "worker")}
|
|
|
|
|
|
# ── the CLI actually takes the lock ──────────────────────────────────────────
|
|
|
|
|
|
def _cli(config: Config, *args: str, timeout: int = 60) -> subprocess.CompletedProcess:
|
|
env = {
|
|
**os.environ,
|
|
"PYTHONPATH": str(REPO),
|
|
"PHOTO_PIPELINE_DATA_DIR": str(config.data_dir),
|
|
"PHOTO_PIPELINE_LIBRARY_ROOTS": os.pathsep.join(
|
|
str(root) for root in config.library_roots
|
|
),
|
|
}
|
|
return subprocess.run(
|
|
[sys.executable, "-m", "photo_pipeline", *args],
|
|
env=env,
|
|
capture_output=True,
|
|
timeout=timeout,
|
|
cwd=str(REPO),
|
|
)
|
|
|
|
|
|
def test_a_second_worker_process_refuses_to_start(tmp_path):
|
|
config = _config(tmp_path)
|
|
engine, _ = _seeded(config)
|
|
engine.dispose()
|
|
env = {
|
|
**os.environ,
|
|
"PYTHONPATH": str(REPO),
|
|
"PHOTO_PIPELINE_DATA_DIR": str(config.data_dir),
|
|
"PHOTO_PIPELINE_LIBRARY_ROOTS": str(config.library_roots[0]),
|
|
}
|
|
first = subprocess.Popen(
|
|
[sys.executable, "-m", "photo_pipeline", "worker", "--id", "first"],
|
|
env=env,
|
|
cwd=str(REPO),
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
)
|
|
try:
|
|
lock = LibraryLock(config, "worker")
|
|
deadline = __import__("time").monotonic() + 30
|
|
while lock.holder() is None and __import__("time").monotonic() < deadline:
|
|
__import__("time").sleep(0.1)
|
|
assert lock.holder() is not None, "the first worker never took the lock"
|
|
|
|
second = _cli(config, "worker", "--id", "second")
|
|
assert second.returncode == 2
|
|
assert b"already running" in second.stderr
|
|
finally:
|
|
first.terminate()
|
|
first.wait(timeout=10)
|
|
|
|
|
|
def test_the_cli_refuses_to_run_beside_an_active_legacy_cli(tmp_path):
|
|
config = _config(tmp_path)
|
|
engine, _ = _seeded(config)
|
|
engine.dispose()
|
|
(config.library_roots[0] / "photo_analyzer_history.jsonl").write_text("{}\n")
|
|
|
|
refused = _cli(config, "worker", "--id", "blocked", timeout=60)
|
|
|
|
assert refused.returncode == 3
|
|
assert b"legacy CLI is writing this library" in refused.stderr
|
|
assert b"--allow-legacy" in refused.stderr
|