diff --git a/.gitignore b/.gitignore index 3f58b27..6ebfe19 100644 --- a/.gitignore +++ b/.gitignore @@ -17,3 +17,6 @@ _todo/ pictures/ photos/ _IGNORE/ + +# Test failure evidence (US07-04) +.artifacts/ diff --git a/README.md b/README.md index 0d25d10..f65e826 100644 --- a/README.md +++ b/README.md @@ -252,6 +252,35 @@ twice to prove it does not drift. work_item/scripts/python -m pytest tests/integration/test_media_hardening.py tests/integration/test_exif_checkpoints.py -q ``` +## Concurrency and crash recovery (US07-04) + +Crash safety is proven by crashing. `photo_pipeline/faults.py` defines the control +points — the persisted transitions of the rename, archive, EXIF, upload, and job +lanes — and arms one only when `PHOTO_PIPELINE_FAULT_AFTER` names it, at which +point the process dies the way a `SIGKILL` does. There is no endpoint and no +configuration field that can reach a barrier; a deployment that never sets the +variable can never hit one. + +The race suite runs each scenario several times with a seed recorded on the test +result (`race_seed`) and asserts invariants rather than schedules: work is never +claimed or executed twice, a stale fencing token never commits, no file body is +lost or overwritten, and the database still passes `PRAGMA integrity_check`. + +```bash +work_item/scripts/python -m pytest tests/integration/test_concurrency_races.py \ + tests/integration/test_fault_matrix.py tests/e2e/test_crash_recovery.py -q + +# replay a failure, or soak for new interleavings +PHOTO_PIPELINE_RACE_SEED=1234 PHOTO_PIPELINE_RACE_REPEATS=50 \ + work_item/scripts/python -m pytest tests/integration/test_concurrency_races.py -q +``` + +Any failing test keeps its evidence: the temporary database (with its write-ahead +log), the journals, the logs, the recorded seed, and a SHA-256 manifest of every +file in the temporary library are copied to `.artifacts//` before pytest +deletes the directory. Point `PHOTO_PIPELINE_TEST_ARTIFACTS` elsewhere to collect +them from CI. + ## Legacy CLI archive The command-line tools this application was extracted from are frozen in diff --git a/photo_pipeline/faults.py b/photo_pipeline/faults.py new file mode 100644 index 0000000..6fda151 --- /dev/null +++ b/photo_pipeline/faults.py @@ -0,0 +1,45 @@ +"""Test-only fault control points (concept §18, US07-04). + +Crash safety can only be proven by crashing at the exact moment a transition has +been persisted but its consequence has not. That needs a barrier *inside* the +production code path — but not a production capability: there is no endpoint, no +service method, and no configuration file entry that can trigger one. The only +switch is an environment variable naming a single point, read at the moment it is +passed, and the only thing it does is kill the process. A deployment that never +sets it can never reach the barrier. + +``os._exit`` is deliberate: it skips atexit handlers, buffered flushes, and +``finally`` blocks, which is what a real ``SIGKILL`` or power loss does. A clean +shutdown would prove nothing. + +The points are the persisted transitions of the journalled stages: + + rename moving | moved | database_updated | verified | complete + archive transferring | verified | removing | source_removed | complete + exif exif:written — keywords on disk, checkpoint not yet recorded + upload upload:accepted — uploader exited, outcome not yet persisted + jobs job:item_done — item committed, job outcome not yet written + +Recovery for each is asserted in tests/integration/test_fault_matrix.py and +tests/e2e/test_crash_recovery.py. +""" + +from __future__ import annotations + +import os + +ENV_VAR = "PHOTO_PIPELINE_FAULT_AFTER" + +EXIF_WRITTEN = "exif:written" +UPLOAD_ACCEPTED = "upload:accepted" +JOB_ITEM_DONE = "job:item_done" + + +def maybe_fault(point: str) -> None: + """Die abruptly when ``PHOTO_PIPELINE_FAULT_AFTER`` names ``point``. + + Shared by the rename, archive, restore, EXIF, upload, and job lanes, each + passing its own state names. Never set the variable outside tests. + """ + if os.environ.get(ENV_VAR) == point: + os._exit(9) diff --git a/photo_pipeline/integrations/exiftool.py b/photo_pipeline/integrations/exiftool.py index 56aae86..780b04e 100644 --- a/photo_pipeline/integrations/exiftool.py +++ b/photo_pipeline/integrations/exiftool.py @@ -17,6 +17,20 @@ import os import subprocess from collections.abc import Iterable +# A hung exiftool must not hang the worker with it: every call is bounded, and a +# call that runs out of time is treated exactly like a failed one — no metadata +# answer, nothing marked verified (US07-04). The knob exists because "slow" is a +# property of the machine, not of the code: huge files on a slow network volume +# legitimately take longer than the default. +DEFAULT_TIMEOUT_SECONDS = 120.0 + + +def _timeout() -> float: + try: + return float(os.environ.get("PHOTO_PIPELINE_EXIFTOOL_TIMEOUT", DEFAULT_TIMEOUT_SECONDS)) + except ValueError: + return DEFAULT_TIMEOUT_SECONDS + def read_keyword_sets(paths: Iterable[str]) -> dict[str, set[str]]: """Map each path to its lowercased set of ``Keywords`` + ``Subject`` values. @@ -34,8 +48,9 @@ def read_keyword_sets(paths: Iterable[str]) -> dict[str, set[str]]: input="\n".join(paths), capture_output=True, text=True, + timeout=_timeout(), ) - except FileNotFoundError: + except (FileNotFoundError, subprocess.TimeoutExpired): return {} out: dict[str, set[str]] = {} try: @@ -67,9 +82,12 @@ def read_all(path: str) -> dict | None: """ try: result = subprocess.run( - ["exiftool", "-m", "-j", "-G0:1", path], capture_output=True, text=True + ["exiftool", "-m", "-j", "-G0:1", path], + capture_output=True, + text=True, + timeout=_timeout(), ) - except FileNotFoundError: + except (FileNotFoundError, subprocess.TimeoutExpired): return None try: records = json.loads(result.stdout or "[]") @@ -91,4 +109,10 @@ def apply_keywords(path: str, *, add: Iterable[str] = (), remove: Iterable[str] if len(args) == 3: return True args.append(path) - return subprocess.run(args, capture_output=True, text=True).returncode == 0 + try: + return subprocess.run( + args, capture_output=True, text=True, timeout=_timeout() + ).returncode == 0 + except (FileNotFoundError, subprocess.TimeoutExpired): + # A write that never returned is not a write that succeeded. + return False diff --git a/photo_pipeline/jobs/worker.py b/photo_pipeline/jobs/worker.py index 3d24835..159162d 100644 --- a/photo_pipeline/jobs/worker.py +++ b/photo_pipeline/jobs/worker.py @@ -16,6 +16,7 @@ from collections.abc import Mapping, Sequence from sqlalchemy import select from sqlalchemy.orm import sessionmaker +from photo_pipeline.faults import JOB_ITEM_DONE, maybe_fault from photo_pipeline.jobs.handlers import REGISTRY, Cancelled, Handler, JobContext from photo_pipeline.models import JobItem from photo_pipeline.services.jobs import ItemState, JobConflict, JobService, JobState @@ -86,6 +87,9 @@ class Worker: ) else: self.service.set_item(job_id, item_key, ItemState.SUCCEEDED, fencing_token=token) + # One item is durably done, the job outcome is not: the control point + # for a crash mid-batch (US07-04). Recovery must not re-run this item. + maybe_fault(JOB_ITEM_DONE) self.service.heartbeat(job_id, self.worker_id, lease_seconds=self.lease_seconds) self._finalize(job_id, token, cancelled=cancelled, any_failed=any_failed) diff --git a/photo_pipeline/services/analysis.py b/photo_pipeline/services/analysis.py index 3698d5d..5db2ec7 100644 --- a/photo_pipeline/services/analysis.py +++ b/photo_pipeline/services/analysis.py @@ -78,6 +78,11 @@ class AnalysisService: latest[review.asset_id] = review.decision return {aid for aid, decision in latest.items() if decision == SFW} + def _is_still_sfw(self, asset_id: str) -> bool: + """Re-read the current safety decision straight from the database.""" + with self._session_factory() as session: + return asset_id in self._sfw_asset_ids(session) + def eligible_asset_ids(self) -> list[str]: """Confirmed-SFW canonical active assets without a completed analysis.""" with self._session_factory() as session: @@ -161,6 +166,22 @@ class AnalysisService: self._store(asset_id, status="error", result=None, error=str(error), tokens=0, raw="") errors += 1 continue + # Third gate, after the call: a provider request takes seconds, and the + # reviewer may have flipped this asset to NSFW while it was in flight. + # The result describes an asset that is no longer analysable, so it is + # discarded — not stored, and above all not written into its EXIF + # (concept §18 scenario 7, US07-04). + if not self._is_still_sfw(asset_id): + self._store( + asset_id, + status="skipped_nsfw", + result=None, + error="the safety decision changed while analysis was in flight", + tokens=0, + raw="", + ) + skipped += 1 + continue self._store( asset_id, status="analyzed", diff --git a/photo_pipeline/services/archive_transfer.py b/photo_pipeline/services/archive_transfer.py index 754171b..301cd9f 100644 --- a/photo_pipeline/services/archive_transfer.py +++ b/photo_pipeline/services/archive_transfer.py @@ -52,6 +52,7 @@ from sqlalchemy import select from sqlalchemy.orm import sessionmaker from photo_pipeline.config import Config +from photo_pipeline.faults import maybe_fault from photo_pipeline.models import ArchiveLocation, ArchiveOperation, ArchivePlan, Asset, AssetPath from photo_pipeline.services.archive_journal import ( ARCHIVE, @@ -63,7 +64,7 @@ from photo_pipeline.services.archive_journal import ( from photo_pipeline.services.archives import MARKER_NAME, ArchiveError, ArchiveService from photo_pipeline.services.duplicates import DuplicateService from photo_pipeline.services.hashing import sha256_file -from photo_pipeline.services.rename_apply import PreconditionFailed, maybe_fault +from photo_pipeline.services.rename_apply import PreconditionFailed from photo_pipeline.services.thumbnails import ThumbnailService # The per-medium manifest: one JSON line per archived file, appended and fsynced diff --git a/photo_pipeline/services/exif_checkpoint.py b/photo_pipeline/services/exif_checkpoint.py index 17bf5b3..393e94f 100644 --- a/photo_pipeline/services/exif_checkpoint.py +++ b/photo_pipeline/services/exif_checkpoint.py @@ -25,6 +25,7 @@ import uuid from dataclasses import dataclass from datetime import datetime, timezone +from photo_pipeline.faults import EXIF_WRITTEN, maybe_fault from photo_pipeline.integrations import exiftool from photo_pipeline.models import ExifProjection from photo_pipeline.services import hashing @@ -120,6 +121,10 @@ def run( if not exiftool.apply_keywords(path, add=add, remove=remove): return CheckpointResult(FAILED, reason="write_failed") + # The file on disk has changed; nothing about it is recorded yet. A crash here + # is the worst case for metadata, so it is a fault control point (US07-04). + maybe_fault(EXIF_WRITTEN) + after = exiftool.read_all(path) if after is None: return CheckpointResult(FAILED, reason="readback_unreadable") diff --git a/photo_pipeline/services/jobs.py b/photo_pipeline/services/jobs.py index 71d7d50..521a9ca 100644 --- a/photo_pipeline/services/jobs.py +++ b/photo_pipeline/services/jobs.py @@ -236,17 +236,32 @@ class JobService: raise InvalidTransition(f"{job.state} -> {to_state}") if worker_id is not None and job.lease_owner not in (None, worker_id): raise JobConflict(f"job {job_id} owned by {job.lease_owner}, not {worker_id}") - job.state = to_state - job.version += 1 - job.updated_at = now + + # Compare-and-set on the version this decision was made against. Without + # it a transition validated against a row that has since been claimed, + # cancelled, or finished would overwrite that newer state (concept §16 + # database rule 6) — a cancel racing a claim used to un-claim a running + # job and leave the worker finalizing a job it no longer owned. + values = { + "state": to_state, + "version": job.version + 1, + "updated_at": now, + } if error: - job.error_code, job.error_message = error + values["error_code"], values["error_message"] = error if to_state in TERMINAL_STATES: - job.finished_at = now - job.lease_owner = None - job.lease_expires_at = None + values.update(finished_at=now, lease_owner=None, lease_expires_at=None) + result = session.execute( + update(Job).where(Job.id == job_id, Job.version == job.version).values(**values) + ) + if result.rowcount != 1: + session.rollback() + raise JobConflict( + f"job {job_id} changed while transitioning to {to_state}; retry" + ) self._event(session, job_id, f"state:{to_state}", error[1] if error else None) session.commit() + session.expire_all() # the core UPDATE bypassed the identity map return self._snapshot(session, job_id) def cancel(self, job_id: str) -> dict: diff --git a/photo_pipeline/services/rename_apply.py b/photo_pipeline/services/rename_apply.py index 68b884d..ca182eb 100644 --- a/photo_pipeline/services/rename_apply.py +++ b/photo_pipeline/services/rename_apply.py @@ -46,9 +46,11 @@ from pathlib import Path from sqlalchemy import select from sqlalchemy.orm import sessionmaker +from photo_pipeline.faults import maybe_fault from photo_pipeline.models import Asset, AssetPath, RenamePlan from photo_pipeline.services import hashing from photo_pipeline.services.rename_journal import ( + ALLOWED_TRANSITIONS, MANUAL, RESUMABLE, JournalState, @@ -83,18 +85,6 @@ def _now() -> datetime: return datetime.now(timezone.utc) -def maybe_fault(state: str) -> None: - """Test-only crash barrier (concept §18 fault injection). - - When ``PHOTO_PIPELINE_FAULT_AFTER`` names a journal state, the process dies - abruptly the moment that state has been persisted — modelling a real kill at - exactly that transition. Never set outside tests. Shared with the archive - transfer journal (US06-02), which uses the same env var and its own state names. - """ - if os.environ.get("PHOTO_PIPELINE_FAULT_AFTER") == state: - os._exit(9) - - class RenameApplyService: def __init__(self, session_factory: sessionmaker, *, library_roots: tuple = ()) -> None: self._session_factory = session_factory @@ -143,20 +133,10 @@ class RenameApplyService: self._apply_one(operation, token=token, worker_id=worker_id) applied += 1 except PreconditionFailed as error: - self.journal.transition( - operation["id"], - JournalState.FAILED, - fencing_token=token, - error=(error.code, str(error)), - ) + self._record_failure(operation["id"], token, error.code, str(error)) failed += 1 except Exception as error: # unexpected: record and stop touching disk - self.journal.transition( - operation["id"], - JournalState.FAILED, - fencing_token=token, - error=("apply_error", str(error)), - ) + self._record_failure(operation["id"], token, "apply_error", str(error)) failed += 1 state = self.journal.sync_plan_state(plan_id) return { @@ -167,6 +147,26 @@ class RenameApplyService: "state": state, } + def _record_failure(self, operation_id: str, token: int, code: str, message: str) -> None: + """Record a failed operation in a state its journal can actually reach. + + ``failed`` only makes sense while nothing has moved. Once the folder is at + its destination — a postcondition failure such as bytes edited during the + move — the operation is not "failed and forgotten": the disk changed and + the database followed, so it becomes ``rollback_required`` and waits for a + human (US07-04). Guessing an unreachable transition used to raise out of + ``apply`` and lose the record entirely. + """ + current = self.journal.get(operation_id)["journal_state"] + target = ( + JournalState.FAILED + if JournalState.FAILED in ALLOWED_TRANSITIONS.get(current, set()) + else JournalState.ROLLBACK_REQUIRED + ) + self.journal.transition( + operation_id, target, fencing_token=token, error=(code, message) + ) + def _apply_one(self, operation: dict, *, token: int, worker_id: str) -> None: source = Path(operation["source_path"]) destination = Path(operation["destination_path"]) diff --git a/photo_pipeline/services/rename_journal.py b/photo_pipeline/services/rename_journal.py index bf29bb1..e954e1f 100644 --- a/photo_pipeline/services/rename_journal.py +++ b/photo_pipeline/services/rename_journal.py @@ -81,7 +81,17 @@ ALLOWED_TRANSITIONS = { TERMINAL_STATES = frozenset({JournalState.COMPLETE, JournalState.ROLLED_BACK}) # States where the disk may already have been touched by this operation. -UNSAFE_STATES = frozenset({JournalState.MOVING, JournalState.MOVED, JournalState.DATABASE_UPDATED}) +# ``rollback_required`` belongs here too (US07-04): the move happened and someone +# has to decide what to do about it, so the library is not in a state another +# mutation may build on. +UNSAFE_STATES = frozenset( + { + JournalState.MOVING, + JournalState.MOVED, + JournalState.DATABASE_UPDATED, + JournalState.ROLLBACK_REQUIRED, + } +) RESUMABLE = "resumable" ROLLBACK_SAFE = "rollback_safe" diff --git a/photo_pipeline/services/restores.py b/photo_pipeline/services/restores.py index cd0acf8..e5cdcd7 100644 --- a/photo_pipeline/services/restores.py +++ b/photo_pipeline/services/restores.py @@ -53,6 +53,7 @@ from sqlalchemy import select from sqlalchemy.orm import sessionmaker from photo_pipeline.config import Config +from photo_pipeline.faults import maybe_fault from photo_pipeline.jobs.domain_handlers import ARCHIVE_LOCK, LIBRARY_WRITE_LOCK, UPLOAD_LOCK from photo_pipeline.models import ArchiveLocation, ArchiveOperation, ArchivePlan, Asset, AssetPath from photo_pipeline.path_policy import PathPolicyError, is_excluded, normalize_root, resolve_within @@ -73,7 +74,7 @@ from photo_pipeline.services.archive_transfer import ( from photo_pipeline.services.archives import ArchiveError from photo_pipeline.services.hashing import sha256_file from photo_pipeline.services.jobs import JobService -from photo_pipeline.services.rename_apply import PreconditionFailed, maybe_fault +from photo_pipeline.services.rename_apply import PreconditionFailed from photo_pipeline.services.rename_journal import RenameJournal PREFLIGHT_VERSION = 1 diff --git a/photo_pipeline/services/upload_batches.py b/photo_pipeline/services/upload_batches.py index a08af50..9a7f46a 100644 --- a/photo_pipeline/services/upload_batches.py +++ b/photo_pipeline/services/upload_batches.py @@ -37,6 +37,7 @@ from sqlalchemy import select, update from sqlalchemy.orm import sessionmaker from photo_pipeline.config import Config +from photo_pipeline.faults import UPLOAD_ACCEPTED, maybe_fault from photo_pipeline.integrations import immich_go from photo_pipeline.models import UploadBatch, UploadItem from photo_pipeline.services.hashing import sha1_file @@ -236,6 +237,10 @@ class UploadBatchService: error = ("uploader_failed", f"immich-go exited with {result['exit_code']}") item_state = ItemState.FAILED + # The uploader is done and Immich may already hold every file, but nothing + # about that is durable yet — the control point for "accepted, outcome not + # recorded" (US07-04). Recovery must answer ``unknown_requires_verification``. + maybe_fault(UPLOAD_ACCEPTED) self._finish(batch_id, token=token, state=state, error=error, result=result) if item_state: self._set_items(batch_id, item_state) diff --git a/tests/_artifacts.py b/tests/_artifacts.py new file mode 100644 index 0000000..8e0e0b2 --- /dev/null +++ b/tests/_artifacts.py @@ -0,0 +1,96 @@ +"""Failure artifacts for the fault and race suites (US07-04). + +A randomized concurrency failure that leaves nothing behind is a failure nobody +can diagnose: the temporary library is deleted, the database goes with it, and the +seed that produced the interleaving is gone. So when a test fails, everything +needed to reproduce and read it is copied out of the temporary directory: + + // + seeds.json recorded properties (``race_seed``) and the failing test id + manifest.json every file under the temporary directory: path, size, sha256 + files/... the databases (with -wal/-shm), journals, and logs themselves + +The manifest covers the whole tree — including files too large or too private to +copy — so a missing or unexpected file is still visible afterwards. Copying is +bounded by ``MAX_COPY_BYTES``: artifacts must not turn a failing CI run into a +disk-full one. + +Set ``PHOTO_PIPELINE_TEST_ARTIFACTS`` to choose the destination; the default is +``.artifacts/`` in the repository root. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import re +import shutil +from pathlib import Path + +REPO = Path(__file__).resolve().parents[1] +DEFAULT_DIR = REPO / ".artifacts" +MAX_COPY_BYTES = 25 * 1024 * 1024 +# Databases (and their write-ahead logs), journals exported as files, and logs. +COPY_SUFFIXES = (".db", ".db-wal", ".db-shm", ".sqlite", ".log", ".json", ".jsonl", ".argv") + + +def artifacts_dir() -> Path: + return Path(os.environ.get("PHOTO_PIPELINE_TEST_ARTIFACTS", DEFAULT_DIR)) + + +def _slug(test_id: str) -> str: + return re.sub(r"[^A-Za-z0-9._-]+", "_", test_id)[:120] + + +def manifest(root: Path) -> list[dict]: + """Every file under ``root``: relative path, byte size, and SHA-256. + + The filesystem state at the moment of failure — what was moved, what was left + behind, what was half-written. + """ + entries = [] + for path in sorted(root.rglob("*")): + if not path.is_file() or path.is_symlink(): + continue + try: + body = path.read_bytes() + except OSError as error: + entries.append({"path": str(path.relative_to(root)), "error": str(error)}) + continue + entries.append( + { + "path": str(path.relative_to(root)), + "bytes": len(body), + "sha256": hashlib.sha256(body).hexdigest(), + } + ) + return entries + + +def collect(root: Path, test_id: str, *, properties: dict | None = None) -> Path: + """Copy the evidence for one failed test out of ``root``. Returns its directory.""" + destination = artifacts_dir() / _slug(test_id) + files = destination / "files" + files.mkdir(parents=True, exist_ok=True) + + entries = manifest(root) + (destination / "manifest.json").write_text(json.dumps(entries, indent=1)) + (destination / "seeds.json").write_text( + json.dumps({"test": test_id, "properties": properties or {}}, indent=1) + ) + + budget = MAX_COPY_BYTES + for path in sorted(root.rglob("*")): + if not path.is_file() or path.is_symlink(): + continue + if not path.name.endswith(COPY_SUFFIXES): + continue + size = path.stat().st_size + if size > budget: + continue # the manifest still records it; the copy is what is skipped + target = files / path.relative_to(root) + target.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(path, target) + budget -= size + return destination diff --git a/tests/conftest.py b/tests/conftest.py index 96fbf01..f095287 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,5 +1,6 @@ -"""Make the repository root importable for the pipeline test suites, and give every -suite the application session the API requires since US07-02. +"""Make the repository root importable for the pipeline test suites, give every +suite the application session the API requires since US07-02, and keep the evidence +of any failure (US07-04). The suites drive the API the way the browser does — module-level ``httpx`` calls and ``TestClient`` — so instead of threading a cookie through several hundred call sites, @@ -99,3 +100,30 @@ def _api_session(): httpx._api.request, httpx.request = real_request, real_request httpx._api.stream, httpx.stream = real_stream, real_stream TestClient.request, TestClient.__init__ = real_client_request, real_client_init + + +@pytest.hookimpl(hookwrapper=True) +def pytest_runtest_makereport(item, call): + """On failure, keep the temporary library, database, journals, logs, and seed. + + ``tmp_path`` is deleted a few runs later and its contents are the only record + of what a randomized or fault-injected failure actually did, so they are copied + out while they still exist (US07-04). + """ + outcome = yield + report = outcome.get_result() + if report.when != "call" or not report.failed: + return + root = item.funcargs.get("tmp_path") + if root is None or not Path(root).exists(): + return + from tests._artifacts import collect + + try: + destination = collect( + Path(root), item.nodeid, properties=dict(getattr(report, "user_properties", [])) + ) + except OSError as error: # never let evidence collection mask the real failure + report.sections.append(("failure artifacts", f"could not be collected: {error}")) + else: + report.sections.append(("failure artifacts", str(destination))) diff --git a/tests/e2e/test_crash_recovery.py b/tests/e2e/test_crash_recovery.py new file mode 100644 index 0000000..f63f434 --- /dev/null +++ b/tests/e2e/test_crash_recovery.py @@ -0,0 +1,274 @@ +"""Process death at the newer control points (US07-04, concept §18). + +The rename and archive journals already prove crash safety at each of their +transitions (tests/integration/test_rename_recovery.py, +tests/integration/test_archive_recovery.py). The three transitions covered here +are the remaining ones where a kill leaves the world and the database disagreeing: + +- ``exif:written`` — keywords are on disk, nothing about them is recorded; +- ``upload:accepted``— the uploader finished, no outcome is stored; +- ``job:item_done`` — one item is durably done, the job is not finished. + +Each test kills a real child process at the barrier and then asserts what a +restart does: resume idempotently, or say plainly that a human has to look. Never +"assume it worked". +""" + +from __future__ import annotations + +import os +import shutil +import subprocess +import sys +import uuid +from datetime import datetime, timezone +from pathlib import Path + +import numpy as np +import pytest +from PIL import Image +from sqlalchemy import select + +from photo_pipeline.config import Config +from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations +from photo_pipeline.faults import EXIF_WRITTEN, JOB_ITEM_DONE, UPLOAD_ACCEPTED +from photo_pipeline.integrations import exiftool +from photo_pipeline.jobs.worker import Worker +from photo_pipeline.models import Asset, SafetyReview +from photo_pipeline.services import exif_checkpoint, hashing +from photo_pipeline.services.jobs import ItemState, JobService, JobState +from photo_pipeline.services.safety import SafetyService +from photo_pipeline.services.upload_batches import BatchState, UploadBatchService +from photo_pipeline.services.upload_verification import retry_blockers +from photo_pipeline.services.uploads import UploadService +from tests.e2e._pipeline_harness import ( + SILENT_UPLOADER, + FakeImmich, + fake_uploader, + mark_upload_ready, + seed_album, +) + +REPO = Path(__file__).resolve().parents[2] +NOW = datetime(2026, 1, 1, tzinfo=timezone.utc) + + +def _child(script: str, *args: str, barrier: str, tmp_path: Path) -> None: + """Run ``script`` in a child that dies at ``barrier``; assert it really died.""" + path = tmp_path / f"child_{barrier.replace(':', '_')}.py" + path.write_text(script.format(repo=str(REPO))) + env = dict(os.environ) + env["PHOTO_PIPELINE_FAULT_AFTER"] = barrier + result = subprocess.run( + [sys.executable, str(path), *args], env=env, capture_output=True + ) + assert result.returncode in (9, -9), ( + f"child should have been killed at {barrier}, got {result.returncode}: " + f"{result.stderr.decode(errors='replace')[-400:]}" + ) + + +def _env(tmp_path, **extra): + (tmp_path / "data").mkdir(exist_ok=True) + lib = tmp_path / "lib" + lib.mkdir(exist_ok=True) + config = Config.from_env( + { + "PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data"), + "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib), + **extra, + } + ) + run_migrations(config.database_url) + return config, create_session_factory(create_db_engine(config.database_url)), lib + + +def _image(path: Path, seed: int = 3) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + pixels = np.random.default_rng(seed).integers(0, 256, (64, 96, 3), dtype=np.uint8) + Image.fromarray(pixels).save(path, quality=90) + + +def _register(sf, path: Path) -> str: + asset_id = str(uuid.uuid4()) + with sf() as session: + 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=hashing.sha256_file(path), + ) + ) + session.commit() + return asset_id + + +# ── EXIF written, checkpoint not recorded ──────────────────────────────────── + +DECIDE_SCRIPT = """ +import sys +sys.path.insert(0, {repo!r}) +from photo_pipeline.db import create_db_engine, create_session_factory +from photo_pipeline.services.safety import SafetyService + +db_url, asset_id = sys.argv[1], sys.argv[2] +sf = create_session_factory(create_db_engine(db_url)) +SafetyService(sf).decide(asset_id, "nsfw") +""" + + +@pytest.mark.skipif(shutil.which("exiftool") is None, reason="exiftool not installed") +def test_a_crash_after_the_exif_write_leaves_nothing_verified_and_re_runs_cleanly(tmp_path): + config, sf, lib = _env(tmp_path) + path = lib / "a.jpg" + _image(path) + asset_id = _register(sf, path) + original_sha = hashing.sha256_file(path) + + _child( + DECIDE_SCRIPT, config.database_url, asset_id, barrier=EXIF_WRITTEN, tmp_path=tmp_path + ) + + # The file changed, but the application claims nothing about it: no decision, + # no projection, and the stored hash is still the pre-write one. + assert "nsfw" in exiftool.read_keyword_sets([str(path)])[str(path)] + assert hashing.sha256_file(path) != original_sha + with sf() as session: + assert session.scalars(select(SafetyReview)).all() == [] + assert session.get(Asset, asset_id).current_sha256 == original_sha + assert exif_checkpoint.state_for(sf, asset_id, "safety") is None + + # Re-running is the recovery: the write is idempotent, so the second attempt + # verifies and records what the first one only did to the file. + review = SafetyService(sf).decide(asset_id, "nsfw") + assert review["exif_verified"] is True + assert exif_checkpoint.state_for(sf, asset_id, "safety") == exif_checkpoint.VERIFIED + keywords = exiftool.read_keyword_sets([str(path)])[str(path)] + assert "nsfw" in keywords and "sfw" not in keywords + with sf() as session: + asset = session.get(Asset, asset_id) + assert asset.current_sha256 == hashing.sha256_file(path) + + +# ── uploader accepted, outcome not persisted ───────────────────────────────── + +UPLOAD_SCRIPT = """ +import sys +sys.path.insert(0, {repo!r}) +from photo_pipeline.config import Config +from photo_pipeline.db import create_db_engine, create_session_factory +from photo_pipeline.services.upload_batches import UploadBatchService + +db_url, data_dir, lib, binary, server, batch_id = sys.argv[1:7] +config = Config.from_env( + {{ + "PHOTO_PIPELINE_DATA_DIR": data_dir, + "PHOTO_PIPELINE_LIBRARY_ROOTS": lib, + "PHOTO_PIPELINE_IMMICH_GO_BINARY": binary, + "PHOTO_PIPELINE_IMMICH_SERVER_URL": server, + "PHOTO_PIPELINE_IMMICH_API_KEY": "sentinel", + }} +) +sf = create_session_factory(create_db_engine(db_url)) +UploadBatchService(sf, config=config).run(batch_id) +""" + + +def test_a_crash_after_the_uploader_accepted_requires_verification(tmp_path): + """immich-go exited cleanly and the server may hold every file, but nothing was + written down. Recovery must not guess success — and must not blindly retry.""" + seeded = seed_album(tmp_path) + mark_upload_ready(seeded) + immich = FakeImmich() + binary = fake_uploader(tmp_path, SILENT_UPLOADER) + config = Config.from_env( + { + "PHOTO_PIPELINE_DATA_DIR": str(seeded.data), + "PHOTO_PIPELINE_LIBRARY_ROOTS": str(seeded.lib), + "PHOTO_PIPELINE_IMMICH_GO_BINARY": str(binary), + "PHOTO_PIPELINE_IMMICH_SERVER_URL": immich.url, + "PHOTO_PIPELINE_IMMICH_API_KEY": "sentinel", + } + ) + engine = create_db_engine(config.database_url) + sf = create_session_factory(engine) + service = UploadBatchService(sf, config=config) + report = UploadService(sf, config=config).preflight(["rome"]) + assert report["state"] == "ready", report["blockers"] + batch_id = service.create(["rome"], token=report["token"])[0]["id"] + + try: + # The batch is claimed by the child, which dies once the uploader has run. + _child( + UPLOAD_SCRIPT, + config.database_url, + str(seeded.data), + str(seeded.lib), + str(binary), + immich.url, + batch_id, + barrier=UPLOAD_ACCEPTED, + tmp_path=tmp_path, + ) + finally: + immich.stop() + assert service.get(batch_id)["state"] == BatchState.RUNNING # lane still held + + recovered = service.recover() + + assert recovered == {"interrupted": 1} + batch = service.get(batch_id) + assert batch["state"] == BatchState.UNKNOWN and batch["error_code"] == "interrupted" + assert [b["code"] for b in retry_blockers(batch)] == ["requires_verification"] + assert all(item["state"] == "pending" for item in batch["items"]), "nothing claimed as sent" + + +# ── one item done, the job unfinished ──────────────────────────────────────── + +WORKER_SCRIPT = """ +import sys +sys.path.insert(0, {repo!r}) +from pathlib import Path +from photo_pipeline.db import create_db_engine, create_session_factory +from photo_pipeline.jobs.worker import Worker + +db_url, log = sys.argv[1], Path(sys.argv[2]) + +def handler(item_key, ctx): + with log.open("a") as handle: + handle.write(item_key + "\\n") + +sf = create_session_factory(create_db_engine(db_url)) +Worker(sf, {{"scan": handler}}, "killable", lease_seconds=1).run_once() +""" + + +def test_a_crash_between_items_resumes_without_running_the_done_item_twice(tmp_path): + config, sf, lib = _env(tmp_path) + service = JobService(sf) + job = service.enqueue("scan", items=["a", "b", "c"]) + log = tmp_path / "handled.log" + + _child( + WORKER_SCRIPT, config.database_url, str(log), barrier=JOB_ITEM_DONE, tmp_path=tmp_path + ) + + crashed = log.read_text().split() + assert crashed == ["a"], "the child should have died right after its first item" + assert service.get(job["id"])["state"] == JobState.RUNNING + + # A fresh worker takes over once the dead lease expires. + import time + + time.sleep(1.1) # the child's lease is one second long + fresh = Worker(sf, {"scan": lambda item, ctx: log.open("a").write(item + "\n")}, "alive") + fresh.run_once() + + assert service.get(job["id"])["state"] == JobState.SUCCEEDED + handled = log.read_text().split() + assert sorted(handled) == ["a", "b", "c"], f"an item ran twice or not at all: {handled}" + assert service.progress(job["id"])["by_state"] == {ItemState.SUCCEEDED: 3} diff --git a/tests/integration/test_concurrency_races.py b/tests/integration/test_concurrency_races.py new file mode 100644 index 0000000..2437708 --- /dev/null +++ b/tests/integration/test_concurrency_races.py @@ -0,0 +1,535 @@ +"""Randomized concurrency and race tests (US07-04, concept §16 and §18). + +Every test here runs several times with a *recorded* seed: the seed decides the +jitter injected around each racing operation, it is attached to the test result +(``race_seed``), and a failing run can be replayed exactly with + + PHOTO_PIPELINE_RACE_SEED= pytest tests/integration/test_concurrency_races.py + +``PHOTO_PIPELINE_RACE_REPEATS`` raises the repeat count for a soak run; the +default is small enough to belong in the ordinary suite. + +The assertions are invariants, not schedules — a race whose interleaving decides +the *outcome* is fine, one whose interleaving decides whether the database still +makes sense is not: + +- no work is claimed, executed, or completed twice; +- no commit from a stale fencing token lands; +- no file is lost, overwritten, or left with foreign content; +- a decision that changed mid-flight is never overwritten by the older answer; +- the database passes ``PRAGMA integrity_check`` afterwards. +""" + +from __future__ import annotations + +import os +import random +import subprocess +import sys +import threading +import time +import uuid +from collections import Counter +from concurrent.futures import ThreadPoolExecutor +from datetime import datetime, timedelta, timezone +from pathlib import Path + +import numpy as np +import pytest +from PIL import Image +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.faults import JOB_ITEM_DONE +from photo_pipeline.jobs.worker import Worker +from photo_pipeline.models import ( + AlbumProposal, + AnalysisResult, + Asset, + ExifProjection, + SafetyReview, +) +from photo_pipeline.services import hashing +from photo_pipeline.services.analysis import AnalysisService +from photo_pipeline.services.jobs import ( + InvalidTransition, + ItemState, + JobConflict, + JobService, + JobState, + _now, +) +from photo_pipeline.services.rename_apply import RenameApplyService +from photo_pipeline.services.renames import RenameService +from photo_pipeline.services.thumbnails import ThumbnailError, ThumbnailService + +REPO = Path(__file__).resolve().parents[2] +NOW = datetime(2026, 1, 1, tzinfo=timezone.utc) +REPEATS = int(os.environ.get("PHOTO_PIPELINE_RACE_REPEATS", "3")) +MAX_JITTER_SECONDS = 0.01 + + +# ── seeded randomness ──────────────────────────────────────────────────────── + + +@pytest.fixture(params=range(REPEATS), ids=lambda run: f"run{run}") +def rng(request, record_property): + """A seeded RNG whose seed is recorded on the test result. + + Without a pinned seed the run is genuinely random — which is the point, a + fixed schedule stops finding new interleavings after the first green run — + so the seed is reported for replay instead. + """ + pinned = os.environ.get("PHOTO_PIPELINE_RACE_SEED") + seed = int(pinned) + request.param if pinned else random.SystemRandom().randrange(2**32) + record_property("race_seed", seed) + print(f"race seed: {seed}") # visible with -s and in the failure report + return random.Random(seed) + + +def jitter(rng: random.Random) -> None: + """Sleep a random sliver so racing threads interleave differently each run.""" + time.sleep(rng.uniform(0, MAX_JITTER_SECONDS)) + + +# ── fixtures ───────────────────────────────────────────────────────────────── + + +def _config(tmp_path) -> tuple[Config, Path]: + (tmp_path / "data").mkdir(exist_ok=True) + lib = tmp_path / "lib" + lib.mkdir(exist_ok=True) + config = Config.from_env( + { + "PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data"), + "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib), + } + ) + run_migrations(config.database_url) + return config, lib + + +@pytest.fixture +def stack(tmp_path): + """Config, library root, and a factory for *independent* session factories — + each thread must own its session (concept §16 database rule 1).""" + config, lib = _config(tmp_path) + engines = [] + + def factory(): + engine = create_db_engine(config.database_url) + engines.append(engine) + return create_session_factory(engine) + + yield config, lib, factory + for engine in engines: + engine.dispose() + + +def image(path: Path, seed: int) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + pixels = np.random.default_rng(seed).integers(0, 256, (64, 96, 3), dtype=np.uint8) + Image.fromarray(pixels).save(path, quality=90) + + +def register(sf, path: Path) -> str: + asset_id = str(uuid.uuid4()) + with sf() as session: + 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=hashing.sha256_file(path), + ) + ) + session.commit() + return asset_id + + +def album(sf, lib: Path, name: str, *, approved_name: str, count: int = 2) -> list[str]: + """A real folder of real images with an approved proposal renaming it.""" + folder = lib / name + ids = [] + for index in range(count): + path = folder / f"{name}-{index}.jpg" + image(path, seed=abs(hash((name, index))) % 10_000) + ids.append(register(sf, path)) + with sf() as session: + session.add( + AlbumProposal( + id=str(uuid.uuid4()), + album=name, + proposed_name=approved_name, + final_name=approved_name, + status="approved", + version=2, + ) + ) + session.commit() + return ids + + +def contents(lib: Path) -> Counter: + """Every file body under the library — what may never be lost or duplicated.""" + return Counter( + path.read_bytes() + for path in lib.rglob("*") + if path.is_file() and ".rename-" not in path.name + ) + + +def integrity_ok(sf) -> bool: + with sf() as session: + return session.execute(text("PRAGMA integrity_check")).scalar() == "ok" + + +# ── database writer pressure ───────────────────────────────────────────────── + + +def test_many_writers_finish_without_losing_a_row(stack, rng): + """Eight lanes committing at once: SQLite has one writer, so this either works + through the busy timeout or loses data. Nothing may be lost.""" + _, _, factory = stack + service = JobService(factory()) + job_ids = [service.enqueue("scan", items=[f"i{n}" for n in range(4)])["id"] for n in range(8)] + + def write(index: int) -> None: + own = JobService(factory()) + job_id = job_ids[index] + token = own.claim(["scan"], f"w{index}") + if token is None: + return + for item in [f"i{n}" for n in range(4)]: + jitter(rng) + own.set_item(token["id"], item, ItemState.RUNNING, fencing_token=token["fencing_token"]) + own.set_item( + token["id"], item, ItemState.SUCCEEDED, fencing_token=token["fencing_token"] + ) + assert job_id # the claim order is racy; every job is claimed by someone + + with ThreadPoolExecutor(max_workers=8) as pool: + list(pool.map(write, range(8))) + + for job_id in job_ids: + progress = service.progress(job_id) + assert progress["by_state"].get(ItemState.SUCCEEDED) == 4, progress + assert integrity_ok(factory()) + + +# ── worker claim races ─────────────────────────────────────────────────────── + + +def test_a_job_is_claimed_by_exactly_one_worker(stack, rng): + _, _, factory = stack + service = JobService(factory()) + job_ids = {service.enqueue("scan", items=["a"])["id"] for _ in range(10)} + + claimed: list[str] = [] + lock = threading.Lock() + + def claim_all(index: int) -> None: + own = JobService(factory()) + while True: + jitter(rng) + job = own.claim(["scan"], f"w{index}") + if job is None: + return + with lock: + claimed.append(job["id"]) + + with ThreadPoolExecutor(max_workers=4) as pool: + list(pool.map(claim_all, range(4))) + + assert sorted(claimed) == sorted(job_ids), "every job claimed exactly once" + assert len(set(claimed)) == len(claimed) + + +def test_a_late_commit_from_an_expired_lease_is_refused(stack, rng): + """Lease expiry then a late write from the old owner: the fencing token, not + timing, decides who may commit.""" + _, _, factory = stack + service = JobService(factory()) + job = service.enqueue("scan", items=["a"]) + old = service.claim(["scan"], "old") + service.recover_stale(now=_now() + timedelta(hours=1)) + new = service.claim(["scan"], "new") + + jitter(rng) + with pytest.raises(JobConflict): + service.set_item(job["id"], "a", ItemState.RUNNING, fencing_token=old["fencing_token"]) + with pytest.raises(JobConflict): + service.transition(job["id"], JobState.SUCCEEDED, fencing_token=old["fencing_token"]) + + # The current owner finishes the same work without interference. + service.set_item(job["id"], "a", ItemState.RUNNING, fencing_token=new["fencing_token"]) + service.set_item(job["id"], "a", ItemState.SUCCEEDED, fencing_token=new["fencing_token"]) + service.transition(job["id"], JobState.SUCCEEDED, fencing_token=new["fencing_token"]) + assert service.get(job["id"])["state"] == JobState.SUCCEEDED + assert integrity_ok(factory()) + + +def test_cancellation_arriving_at_a_random_moment_leaves_a_consistent_job(stack, rng): + _, _, factory = stack + service = JobService(factory()) + job = service.enqueue("scan", items=[f"i{n}" for n in range(6)]) + handled: list[str] = [] + + def handler(item, ctx): + jitter(rng) + handled.append(item) + + def cancel() -> None: + try: + service.cancel(job["id"]) + except (JobConflict, InvalidTransition): + # The request lost its race with the worker's own transition; the job + # simply finishes. What must never happen is a *silent* overwrite. + pass + + canceller = threading.Timer(rng.uniform(0, 0.03), cancel) + canceller.start() + Worker(factory(), {"scan": handler}, "w1").run_once() + canceller.join() + + state = service.get(job["id"])["state"] + assert state in {JobState.SUCCEEDED, JobState.CANCELLED, JobState.CANCELLING} + by_state = service.progress(job["id"])["by_state"] + assert by_state.get(ItemState.RUNNING, 0) == 0, "no item left mid-flight" + assert len(handled) == len(set(handled)), "no item ran twice" + if state == JobState.CANCELLED: + # Whatever did not run is resumable, never silently dropped. + assert by_state.get(ItemState.SUCCEEDED, 0) + by_state.get(ItemState.QUEUED, 0) == 6 + assert integrity_ok(factory()) + + +KILLABLE_WORKER = """ +import sys +sys.path.insert(0, {repo!r}) +from pathlib import Path +from photo_pipeline.db import create_db_engine, create_session_factory +from photo_pipeline.jobs.worker import Worker + +db_url, log = sys.argv[1], Path(sys.argv[2]) + +def handler(item_key, ctx): + with log.open("a") as handle: + handle.write(item_key + "\\n") + +sf = create_session_factory(create_db_engine(db_url)) +Worker(sf, {{"scan": handler}}, sys.argv[3], lease_seconds=1).run_once() +""" + + +def test_a_worker_killed_at_a_random_item_resumes_exactly_once(stack, rng, tmp_path): + """Kill a real worker mid-batch, restart it, and assert the invariant across + the restart: every item ran exactly once and the job ends succeeded.""" + config, _, factory = stack + service = JobService(factory()) + items = [f"i{n}" for n in range(rng.randint(2, 5))] + job = service.enqueue("scan", items=items) + log = tmp_path / "handled.log" + script = tmp_path / "killable_worker.py" + script.write_text(KILLABLE_WORKER.format(repo=str(REPO))) + + def run_worker(worker_id: str, barrier: str | None) -> None: + env = dict(os.environ) + env.pop("PHOTO_PIPELINE_FAULT_AFTER", None) + if barrier: + env["PHOTO_PIPELINE_FAULT_AFTER"] = barrier + subprocess.run( + [sys.executable, str(script), config.database_url, str(log), worker_id], + env=env, + capture_output=True, + ) + + run_worker("killable", JOB_ITEM_DONE) # dies after its first completed item + assert log.read_text().split() == items[:1] + time.sleep(1.1) # let the dead worker's one-second lease expire + jitter(rng) + run_worker("survivor", None) + + assert service.get(job["id"])["state"] == JobState.SUCCEEDED + handled = log.read_text().split() + assert sorted(handled) == sorted(items), f"an item ran twice or not at all: {handled}" + assert integrity_ok(factory()) + + +# ── filesystem races ───────────────────────────────────────────────────────── + + +def test_a_file_changed_during_apply_never_loses_content(stack, rng): + """The user edits a photo while its folder is being renamed. The rename may + win or be refused, but no body may disappear or be overwritten.""" + config, lib, factory = stack + sf = factory() + album(sf, lib, "rome", approved_name="2019 Rome") + plan = RenameService(sf, library_roots=(lib,)).build_plan() + before = contents(lib) + + edited = b"the user saved over this file" + + def edit() -> None: + jitter(rng) + target = next((lib / "rome").glob("*.jpg"), None) + if target is None: + return # the rename won the race; the folder already moved + try: + target.write_bytes(edited) + except OSError: + pass # ...or it moved between the glob and the write + + thread = threading.Thread(target=edit) + thread.start() + RenameApplyService(sf, library_roots=(lib,)).apply(plan["id"], expected_version=plan["version"]) + thread.join() + + after = contents(lib) + assert sum(after.values()) == sum(before.values()), "a file was lost or duplicated" + survived = (before - Counter({edited: 1})) & after + assert sum(survived.values()) >= sum(before.values()) - 1, "unrelated content was destroyed" + assert integrity_ok(sf) + + +def test_two_folders_claiming_one_destination_never_merge(stack, rng): + """Two approved albums want the same name. The plan must refuse rather than + move one folder into the other.""" + config, lib, factory = stack + sf = factory() + album(sf, lib, "rome-a", approved_name="2019 Rome") + album(sf, lib, "rome-b", approved_name="2019 Rome") + jitter(rng) + + plan = RenameService(sf, library_roots=(lib,)).build_plan() + codes = {issue["code"] for op in plan["operations"] for issue in op["issues"]} + + assert "duplicate_target" in codes + assert plan["applicable"] is False + assert (lib / "rome-a").is_dir() and (lib / "rome-b").is_dir() + + +def test_thumbnail_requests_racing_a_rename_never_serve_a_wrong_file(stack, rng): + """Previews are keyed by pixels, not paths, so a rename must not make a request + fail loudly *or* return another asset's picture.""" + config, lib, factory = stack + sf = factory() + asset_ids = album(sf, lib, "rome", approved_name="2019 Rome") + plan = RenameService(sf, library_roots=(lib,)).build_plan() + + thumbnails = ThumbnailService(factory(), config) + expected = {aid: thumbnails.generate(aid, 256).read_bytes() for aid in asset_ids} + served: dict[str, set[bytes]] = {aid: set() for aid in asset_ids} + errors: list[str] = [] + stop = threading.Event() + + def serve() -> None: + own = ThumbnailService(factory(), config) + while not stop.is_set(): + for asset_id in asset_ids: + jitter(rng) + try: + served[asset_id].add(own.generate(asset_id, 256).read_bytes()) + except ThumbnailError as error: + errors.append(error.code) # precise, never an unhandled crash + + reader = threading.Thread(target=serve) + reader.start() + try: + RenameApplyService(sf, library_roots=(lib,)).apply( + plan["id"], expected_version=plan["version"] + ) + finally: + stop.set() + reader.join() + + for asset_id, bodies in served.items(): + assert bodies <= {expected[asset_id]}, "a request served another asset's picture" + assert integrity_ok(sf) + + +# ── stage races ────────────────────────────────────────────────────────────── + + +def test_an_analysis_result_racing_a_safety_flip_is_discarded(stack, rng): + """Concept §18 scenario 7: the reviewer marks an asset NSFW while the provider + call is in flight. The answer that comes back describes an asset that may no + longer be analysed, so it is dropped — and no analysis EXIF is written.""" + _, lib, factory = stack + sf = factory() + path = lib / "beach.jpg" + image(path, seed=7) + asset_id = register(sf, path) + with sf() as session: + session.add( + SafetyReview(id=str(uuid.uuid4()), asset_id=asset_id, decision="sfw", created_at=NOW) + ) + session.commit() + before_sha = hashing.sha256_file(path) + + class FlippingProvider: + """Records the call, then the reviewer's decision lands mid-flight.""" + + def __init__(self) -> None: + self.calls: list[str] = [] + + def analyze(self, path, *, album_hint): + self.calls.append(path) + jitter(rng) + with sf() as session: + session.add( + SafetyReview( + id=str(uuid.uuid4()), + asset_id=asset_id, + decision="nsfw", + prior_decision="sfw", + created_at=NOW + timedelta(minutes=1), + ) + ) + session.commit() + return {"description": "a beach", "tags": ["beach", "sand"]} + + provider = FlippingProvider() + result = AnalysisService(sf, provider=provider, library_roots=(lib,)).run([asset_id]) + + assert provider.calls == [str(path)], "the call was legitimate when it started" + assert result == {"analyzed": 0, "skipped": 1, "errors": 0} + with sf() as session: + row = session.get(AnalysisResult, asset_id) + assert row.status == "skipped_nsfw" and row.description is None + assert session.get(ExifProjection, (asset_id, "analysis")) is None + assert hashing.sha256_file(path) == before_sha, "the file was written after the flip" + + +def test_a_safety_decision_taken_twice_at_once_keeps_one_history(stack, rng): + """Two windows, one asset, the same decision: the audit trail may record both + attempts, but the effective state must be a single coherent decision.""" + _, lib, factory = stack + sf = factory() + path = lib / "twice.jpg" + image(path, seed=11) + asset_id = register(sf, path) + + from photo_pipeline.services.safety import SafetyService + + def decide(decision: str) -> None: + jitter(rng) + SafetyService(factory()).decide(asset_id, decision, write_exif=False) + + with ThreadPoolExecutor(max_workers=2) as pool: + list(pool.map(decide, ["sfw", "sfw"])) + + with sf() as session: + decisions = [ + review.decision + for review in session.scalars( + select(SafetyReview) + .where(SafetyReview.asset_id == asset_id) + .order_by(SafetyReview.created_at) + ) + ] + assert decisions and set(decisions) == {"sfw"} + assert integrity_ok(sf) diff --git a/tests/integration/test_fault_matrix.py b/tests/integration/test_fault_matrix.py new file mode 100644 index 0000000..c494600 --- /dev/null +++ b/tests/integration/test_fault_matrix.py @@ -0,0 +1,527 @@ +"""The fault matrix (US07-04, concept §18 "crash/fault-injection tests"). + +Process death at each persisted transition lives in tests/e2e/test_crash_recovery.py +and the per-stage recovery suites. This file covers the *environmental* faults — +the ones that are not a crash but are just as good at corrupting a library if the +code guesses: + + disk full · read-only path · database busy · database corruption · + network failure · malformed provider output · GPU exhaustion · + subprocess hang · missing external tool + +Every case asserts the same shape of outcome: the operation fails visibly, the +failure names what happened, and nothing irreversible was done on the way — no +source removed, no metadata marked verified, no decision invented. +""" + +from __future__ import annotations + +import errno +import os +import sqlite3 +import stat +import threading +import time +import uuid +from contextlib import contextmanager +from datetime import datetime, timezone +from pathlib import Path + +import numpy as np +import pytest +from PIL import Image +from sqlalchemy import select, text +from sqlalchemy.exc import DatabaseError, OperationalError + +from photo_pipeline import faults +from photo_pipeline.config import Config +from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations +from photo_pipeline.jobs.worker import Worker +from photo_pipeline.models import ( + AlbumProposal, + AnalysisResult, + Asset, + ExifProjection, + SafetyReview, + UploadBatch, + UploadItem, +) +from photo_pipeline.services import archive_transfer, exif_checkpoint, hashing, rename_apply +from photo_pipeline.services.analysis import AnalysisService +from photo_pipeline.services.archive_transfer import ArchiveTransferService +from photo_pipeline.services.archives import ArchiveService +from photo_pipeline.services.jobs import ItemState, JobService +from photo_pipeline.services.rename_apply import ApplyError, RenameApplyService +from photo_pipeline.services.rename_journal import JournalState, RenameJournal +from photo_pipeline.services.renames import RenameService +from photo_pipeline.services.safety import SafetyService +from photo_pipeline.services.uploads import UploadService + +NOW = datetime(2026, 1, 1, tzinfo=timezone.utc) + + +# ── environment ────────────────────────────────────────────────────────────── + + +def _env(tmp_path, **extra): + (tmp_path / "data").mkdir(exist_ok=True) + lib = tmp_path / "lib" + lib.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", + **extra, + } + ) + run_migrations(config.database_url) + return config, create_session_factory(create_db_engine(config.database_url)), lib + + +def image(path: Path, seed: int = 1) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + pixels = np.random.default_rng(seed).integers(0, 256, (64, 96, 3), dtype=np.uint8) + Image.fromarray(pixels).save(path, quality=90) + + +def register(sf, path: Path) -> str: + asset_id = str(uuid.uuid4()) + with sf() as session: + 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=hashing.sha256_file(path), + ) + ) + session.commit() + return asset_id + + +def uploaded_album(sf, lib, album="rome", names=("a.jpg",)): + """An album with verified upload evidence — what archiving requires.""" + folder = lib / album + folder.mkdir(parents=True, exist_ok=True) + 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 name in names: + path = folder / name + path.write_bytes(f"{album}/{name} content".encode() * 8) + asset_id = str(uuid.uuid4()) + 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=hashing.sha256_file(path), + ) + ) + session.add( + UploadItem( + batch_id=batch_id, + asset_id=asset_id, + path=str(path), + sha256=hashing.sha256_file(path), + sha1="0" * 40, + state="sent", + outcome="uploaded", + ) + ) + session.commit() + return folder + + +def archive_plan(sf, config, archive, albums=None): + location = ArchiveService(sf, config=config).register("external", str(archive)) + token = ArchiveService(sf, config=config).preflight(location["id"], albums)["token"] + service = ArchiveTransferService(sf, config=config) + return service, service.create(location["id"], albums, token=token) + + +def fake_tool(directory: Path, name: str, body: str) -> Path: + """A real executable on a directory a test can put in front of PATH.""" + directory.mkdir(parents=True, exist_ok=True) + path = directory / name + path.write_text(f"#!/bin/sh\n{body}") + path.chmod(path.stat().st_mode | stat.S_IEXEC | stat.S_IXGRP | stat.S_IXOTH) + return path + + +# ── control points ─────────────────────────────────────────────────────────── + + +def test_the_fault_barrier_does_nothing_unless_its_variable_names_the_point(monkeypatch): + monkeypatch.delenv(faults.ENV_VAR, raising=False) + for point in (faults.EXIF_WRITTEN, faults.UPLOAD_ACCEPTED, faults.JOB_ITEM_DONE, "moving"): + faults.maybe_fault(point) # would kill the process if it were armed + monkeypatch.setenv(faults.ENV_VAR, faults.EXIF_WRITTEN) + faults.maybe_fault(faults.UPLOAD_ACCEPTED) # a different point stays inert + + +def test_no_route_or_configuration_can_arm_a_fault(): + """The control points are reachable only through an environment variable read + inside ``photo_pipeline.faults`` — never through the API, and never through + configuration a browser or a config file could set.""" + from photo_pipeline.api.app import create_app + + app = create_app() + assert not [route for route in app.routes if "fault" in getattr(route, "path", "")] + assert not [field for field in Config.model_fields if "fault" in field] + + package = Path(__file__).resolve().parents[2] / "photo_pipeline" + sources = { + path.relative_to(package.parent) + for path in package.rglob("*.py") + if faults.ENV_VAR in path.read_text() + } + assert sources == {Path("photo_pipeline/faults.py")} + + +# ── disk full ──────────────────────────────────────────────────────────────── + + +def test_a_full_disk_during_an_archive_never_removes_the_source(tmp_path, monkeypatch): + config, sf, lib = _env(tmp_path) + archive = tmp_path / "archive" + archive.mkdir() + folder = uploaded_album(sf, lib) + original = {path: path.read_bytes() for path in folder.iterdir()} + service, plan = archive_plan(sf, config, archive) + + def no_space(*args, **kwargs): + raise OSError(errno.ENOSPC, "No space left on device") + + # Force the cross-filesystem path (a real archive medium) and fill it up. + monkeypatch.setattr(archive_transfer, "_same_filesystem", lambda *a: False) + monkeypatch.setattr(archive_transfer.shutil, "copyfileobj", no_space) + + result = service.apply(plan["id"]) + + assert result["archived"] == 0 and result["failed"] == 1 + for path, body in original.items(): + assert path.read_bytes() == body, "the source was touched despite the failure" + with sf() as session: + assert all(a.availability_state == "active" for a in session.scalars(select(Asset))) + assert [p for p in archive.rglob("*") if p.is_file() and not p.name.startswith(".")] == [] + + +# ── read-only paths ────────────────────────────────────────────────────────── + + +@pytest.mark.skipif(os.geteuid() == 0, reason="root ignores directory permissions") +def test_a_read_only_library_refuses_the_rename_and_keeps_the_source(tmp_path): + config, sf, lib = _env(tmp_path) + folder = lib / "rome" + image(folder / "a.jpg") + register(sf, folder / "a.jpg") + with sf() as session: + session.add( + AlbumProposal( + id=str(uuid.uuid4()), + album="rome", + proposed_name="2019 Rome", + final_name="2019 Rome", + status="approved", + version=2, + ) + ) + session.commit() + plan = RenameService(sf, library_roots=(lib,)).build_plan() + + mode = lib.stat().st_mode + lib.chmod(0o500) # readable, traversable, not writable + try: + result = RenameApplyService(sf, library_roots=(lib,)).apply( + plan["id"], expected_version=plan["version"] + ) + finally: + lib.chmod(mode) + + assert result["applied"] == 0 and result["failed"] == 1 + assert (folder / "a.jpg").exists() and not (lib / "2019 Rome").exists() + with sf() as session: + asset = session.scalars(select(Asset)).one() + assert asset.current_path == str(folder / "a.jpg") + + +# ── database faults ────────────────────────────────────────────────────────── + + +@contextmanager +def exclusive_lock(database_url: str): + """Hold SQLite's write lock from another thread, the way a second process would. + + A sqlite3 connection belongs to the thread that opened it, so the holder thread + opens, locks, waits, and releases entirely on its own. + """ + path = database_url.replace("sqlite:///", "") + locked, release = threading.Event(), threading.Event() + + def hold() -> None: + connection = sqlite3.connect(path, timeout=10) + connection.execute("BEGIN EXCLUSIVE") + locked.set() + release.wait(30) + connection.rollback() + connection.close() + + thread = threading.Thread(target=hold) + thread.start() + assert locked.wait(10), "the holder never acquired the lock" + try: + yield release.set # callers may release early; exiting releases anyway + finally: + release.set() + thread.join(10) + + +def test_a_busy_database_waits_rather_than_failing(tmp_path): + """SQLite has one writer. A short conflict must resolve through the busy + timeout instead of surfacing as an error.""" + config, sf, lib = _env(tmp_path) + service = JobService(sf) + job = service.enqueue("scan", items=["a"]) + + with exclusive_lock(config.database_url) as release: + threading.Timer(0.3, release).start() + started = time.monotonic() + claimed = service.claim(["scan"], "w1") # blocks until the lock is gone + waited = time.monotonic() - started + + assert claimed["id"] == job["id"] and claimed["state"] == "running" + assert waited >= 0.25, "the claim did not actually wait for the writer" + with sf() as session: + assert session.execute(text("PRAGMA busy_timeout")).scalar() >= 1000 + + +def test_a_database_locked_beyond_the_timeout_is_an_error_not_a_silent_skip(tmp_path): + config, sf, lib = _env(tmp_path) + with sf() as session: + session.execute(text("SELECT 1")) # connect first: the lock comes after + with exclusive_lock(config.database_url): + session.execute(text("PRAGMA busy_timeout=50")) # do not wait five seconds + with pytest.raises(OperationalError, match="locked"): + session.execute( + text("INSERT INTO jobs (id, job_type, state) VALUES ('x','scan','queued')") + ) + session.commit() + session.rollback() + + # The refusal left nothing behind, and the database is still sound. + with sf() as session: + assert session.execute(text("PRAGMA integrity_check")).scalar() == "ok" + assert session.execute(text("SELECT count(*) FROM jobs")).scalar() == 0 + + +def test_a_corrupt_database_fails_loudly_instead_of_answering_wrongly(tmp_path): + config, sf, lib = _env(tmp_path) + for index in range(50): # enough rows to fill several pages + JobService(sf).enqueue("scan", items=[f"item-{index}-{n}" for n in range(20)]) + with sf() as session: + session.execute(text("PRAGMA wal_checkpoint(TRUNCATE)")) # WAL into the file + session.commit() + source = Path(config.database_url.replace("sqlite:///", "")) + + corrupt = tmp_path / "corrupt.db" + body = bytearray(source.read_bytes()) + body[4096 : 4096 + 2048] = b"\xde\xad\xbe\xef" * 512 # shred pages, keep the header + corrupt.write_bytes(bytes(body)) + + engine = create_db_engine(f"sqlite:///{corrupt}") + factory = create_session_factory(engine) + try: + with factory() as session: + assert session.execute(text("PRAGMA integrity_check")).scalar() != "ok" + # Reading the shredded pages must raise, never return half a table. + with pytest.raises(DatabaseError): + session.execute(text("SELECT * FROM job_items")).all() + session.execute(text("SELECT * FROM job_events")).all() + session.execute(text("REINDEX")).all() + finally: + engine.dispose() + + +# ── external services ──────────────────────────────────────────────────────── + + +def test_an_unreachable_immich_blocks_upload_instead_of_starting_one(tmp_path): + # Port 9 (discard) refuses connections deterministically. + config, sf, lib = _env( + tmp_path, + PHOTO_PIPELINE_IMMICH_SERVER_URL="http://127.0.0.1:9", + PHOTO_PIPELINE_IMMICH_API_KEY="sentinel", + ) + report = UploadService(sf, config=config).preflight() + codes = {blocker["code"] for blocker in report["blockers"]} + assert "server_unreachable" in codes + assert report["state"] != "ready" + + +def test_a_missing_uploader_blocks_upload_with_the_binary_named(tmp_path): + config, sf, lib = _env( + tmp_path, + PHOTO_PIPELINE_IMMICH_SERVER_URL="http://127.0.0.1:9", + PHOTO_PIPELINE_IMMICH_API_KEY="sentinel", + PHOTO_PIPELINE_IMMICH_GO_BINARY=str(tmp_path / "no-such-immich-go"), + ) + report = UploadService(sf, config=config).preflight() + assert "immich_go_missing" in {blocker["code"] for blocker in report["blockers"]} + + +def test_a_malformed_provider_answer_is_a_per_asset_error(tmp_path): + config, sf, lib = _env(tmp_path) + path = lib / "a.jpg" + image(path) + asset_id = register(sf, path) + with sf() as session: + session.add( + SafetyReview(id=str(uuid.uuid4()), asset_id=asset_id, decision="sfw", created_at=NOW) + ) + session.commit() + + class MalformedProvider: + def analyze(self, path, *, album_hint): + raise ValueError("Expecting value: line 1 column 1 (char 0)") + + result = AnalysisService(sf, provider=MalformedProvider(), library_roots=(lib,)).run([asset_id]) + + assert result == {"analyzed": 0, "skipped": 0, "errors": 1} + with sf() as session: + row = session.get(AnalysisResult, asset_id) + assert row.status == "error" and "Expecting value" in row.error_message + assert session.get(ExifProjection, (asset_id, "analysis")) is None + + +def test_gpu_exhaustion_fails_the_item_without_inventing_a_score(tmp_path): + config, sf, lib = _env(tmp_path) + path = lib / "a.jpg" + image(path) + asset_id = register(sf, path) + + class ExhaustedModel: + def score(self, paths): + raise RuntimeError("MPS backend out of memory (MPS allocated: 9.00 GB)") + + service = SafetyService(sf, model=ExhaustedModel()) + jobs = JobService(sf) + job = jobs.enqueue("safety_score", items=[asset_id]) + Worker(sf, {"safety_score": lambda item, ctx: service.score_assets([item])}, "w1").run_once() + + progress = jobs.progress(job["id"]) + assert progress["by_state"] == {ItemState.FAILED: 1} + with sf() as session: + assert session.scalars(select(SafetyReview)).all() == [], "no score was invented" + + +# ── external tools ─────────────────────────────────────────────────────────── + + +def test_a_hanging_exiftool_times_out_and_verifies_nothing(tmp_path, monkeypatch): + config, sf, lib = _env(tmp_path) + path = lib / "a.jpg" + image(path) + asset_id = register(sf, path) + before = hashing.sha256_file(path) + + fake_tool(tmp_path / "bin", "exiftool", "sleep 30\n") + monkeypatch.setenv("PATH", f"{tmp_path / 'bin'}:{os.environ['PATH']}") + monkeypatch.setenv("PHOTO_PIPELINE_EXIFTOOL_TIMEOUT", "1") + + review = SafetyService(sf).decide(asset_id, "nsfw") + + # The decision is durable; the metadata claim is not made. + assert review["decision"] == "nsfw" and review["exif_verified"] is False + assert exif_checkpoint.state_for(sf, asset_id, "safety") == exif_checkpoint.FAILED + assert hashing.sha256_file(path) == before + + +def test_a_missing_exiftool_is_a_failed_checkpoint_not_a_verified_one(tmp_path, monkeypatch): + config, sf, lib = _env(tmp_path) + path = lib / "a.jpg" + image(path) + asset_id = register(sf, path) + + empty_bin = tmp_path / "empty-bin" + empty_bin.mkdir() + monkeypatch.setenv("PATH", str(empty_bin)) # nothing on PATH at all + + review = SafetyService(sf).decide(asset_id, "sfw") + + assert review["exif_verified"] is False + assert exif_checkpoint.state_for(sf, asset_id, "safety") == exif_checkpoint.FAILED + with sf() as session: + # Upload eligibility depends on a verified checkpoint, so it stays blocked. + assert session.scalars(select(SafetyReview)).all()[-1].exif_verified_at is None + + +def test_a_file_edited_during_the_move_is_left_for_a_human(tmp_path, monkeypatch): + """The user saves over a photo in the instant between the move and its + verification. The move already happened and the database already followed it, + so the operation cannot simply be "failed": it becomes ``rollback_required`` + and blocks further mutation until someone decides (US07-04).""" + config, sf, lib = _env(tmp_path) + folder = lib / "rome" + image(folder / "a.jpg") + register(sf, folder / "a.jpg") + with sf() as session: + session.add( + AlbumProposal( + id=str(uuid.uuid4()), + album="rome", + proposed_name="2019 Rome", + final_name="2019 Rome", + status="approved", + version=2, + ) + ) + session.commit() + plan = RenameService(sf, library_roots=(lib,)).build_plan() + + real_rename = rename_apply.os.rename + + def rename_then_edit(source, destination): + real_rename(source, destination) + for path in Path(destination).glob("*.jpg"): + path.write_bytes(b"the user saved over this file") + + monkeypatch.setattr(rename_apply.os, "rename", rename_then_edit) + + result = RenameApplyService(sf, library_roots=(lib,)).apply( + plan["id"], expected_version=plan["version"] + ) + + assert result["applied"] == 0 and result["failed"] == 1 + journal = RenameJournal(sf) + operation = journal.incomplete()[0] + assert operation["journal_state"] == JournalState.ROLLBACK_REQUIRED + assert operation["error_code"] == "verify_bytes" + assert journal.blocks_mutation() is True, "the unresolved rename blocks the library" + + # Recovery offers the rollback the evidence supports, and the rollback itself + # refuses the edited bytes rather than putting the user's newer file back as if + # it were the old one. + service = RenameApplyService(sf, library_roots=(lib,)) + with pytest.raises(ApplyError, match="manual recovery"): + service.rollback_operation(operation["id"]) + # Nothing was lost: the edited file is at its new home, not deleted. + assert (lib / "2019 Rome" / "a.jpg").read_bytes() == b"the user saved over this file" diff --git a/tests/story_traceability.json b/tests/story_traceability.json index cd1f570..272c609 100644 --- a/tests/story_traceability.json +++ b/tests/story_traceability.json @@ -156,6 +156,12 @@ "US07-03": [ "tests/integration/test_media_hardening.py", "tests/integration/test_exif_checkpoints.py" + ], + "US07-04": [ + "tests/unit/test_fault_artifacts.py", + "tests/integration/test_concurrency_races.py", + "tests/integration/test_fault_matrix.py", + "tests/e2e/test_crash_recovery.py" ] } } diff --git a/tests/unit/test_fault_artifacts.py b/tests/unit/test_fault_artifacts.py new file mode 100644 index 0000000..3246379 --- /dev/null +++ b/tests/unit/test_fault_artifacts.py @@ -0,0 +1,69 @@ +"""The failure-artifact collector (US07-04). + +CI must be able to explain a randomized failure after the temporary library is +gone, which means the seed, the database, the journals, the logs, and a complete +filesystem manifest have to leave the temporary directory before pytest cleans it. +""" + +from __future__ import annotations + +import hashlib +import json + +from tests._artifacts import collect, manifest + + +def _library(root): + (root / "data").mkdir(parents=True) + (root / "data" / "photo_pipeline.db").write_bytes(b"database bytes") + (root / "data" / "photo_pipeline.db-wal").write_bytes(b"write ahead log") + (root / "data" / "uploads").mkdir() + (root / "data" / "uploads" / "batch.log").write_text("INFO uploaded a.jpg\n") + (root / "lib").mkdir() + (root / "lib" / "a.jpg").write_bytes(b"pixels") + return root + + +def test_the_manifest_covers_every_file_with_its_hash(tmp_path): + root = _library(tmp_path / "run") + entries = {entry["path"]: entry for entry in manifest(root)} + + assert set(entries) == { + "data/photo_pipeline.db", + "data/photo_pipeline.db-wal", + "data/uploads/batch.log", + "lib/a.jpg", + } + assert entries["lib/a.jpg"]["sha256"] == hashlib.sha256(b"pixels").hexdigest() + assert entries["lib/a.jpg"]["bytes"] == 6 + + +def test_collect_keeps_the_database_journals_logs_and_seed(tmp_path, monkeypatch): + root = _library(tmp_path / "run") + monkeypatch.setenv("PHOTO_PIPELINE_TEST_ARTIFACTS", str(tmp_path / "artifacts")) + + destination = collect(root, "tests/x.py::test_races[run1]", properties={"race_seed": 1234}) + + seeds = json.loads((destination / "seeds.json").read_text()) + assert seeds["properties"]["race_seed"] == 1234 + assert seeds["test"].endswith("test_races[run1]") + + kept = {str(p.relative_to(destination / "files")) for p in (destination / "files").rglob("*") if p.is_file()} + assert kept == { + "data/photo_pipeline.db", + "data/photo_pipeline.db-wal", + "data/uploads/batch.log", + }, "databases, write-ahead logs, and logs are the diagnosable evidence" + # The photo itself is never copied out of the library — but it is in the + # manifest, so a file that went missing is still provable. + assert any(entry["path"] == "lib/a.jpg" for entry in json.loads((destination / "manifest.json").read_text())) + + +def test_collecting_twice_for_one_test_is_safe(tmp_path, monkeypatch): + root = _library(tmp_path / "run") + monkeypatch.setenv("PHOTO_PIPELINE_TEST_ARTIFACTS", str(tmp_path / "artifacts")) + + first = collect(root, "tests/x.py::test_a") + second = collect(root, "tests/x.py::test_a") + + assert first == second and (second / "manifest.json").exists() diff --git a/tests/unit/test_rename_journal_states.py b/tests/unit/test_rename_journal_states.py index 7ca2f13..43b9f6f 100644 --- a/tests/unit/test_rename_journal_states.py +++ b/tests/unit/test_rename_journal_states.py @@ -77,6 +77,8 @@ def test_unsafe_states_are_the_ones_where_disk_may_have_changed(): JournalState.MOVING, JournalState.MOVED, JournalState.DATABASE_UPDATED, + # The move happened and a human still has to decide about it (US07-04). + JournalState.ROLLBACK_REQUIRED, } # planned has not touched anything; complete/rolled_back are settled. assert JournalState.PLANNED not in UNSAFE_STATES