diff --git a/migrations/versions/0010_upload_verification.py b/migrations/versions/0010_upload_verification.py new file mode 100644 index 0000000..17205e0 --- /dev/null +++ b/migrations/versions/0010_upload_verification.py @@ -0,0 +1,70 @@ +"""Upload verification and manual resolution (US05-04). + +Revision ID: 0010_upload_verification +Revises: 0009_upload_report_outcomes +Create Date: 2026-08-16 + +Per-item server evidence plus the append-only history that produced it, so an +uncertain upload can be resolved without ever guessing that it succeeded. +""" + +import sqlalchemy as sa +from alembic import op + +revision = "0010_upload_verification" +down_revision = "0009_upload_report_outcomes" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.add_column( + "upload_batches", sa.Column("verified_at", sa.DateTime(timezone=True), nullable=True) + ) + # At least one uploaded file was edited afterwards: a visible warning that also + # blocks re-running the batch. + op.add_column( + "upload_batches", + sa.Column("stale_bytes", sa.Boolean(), nullable=False, server_default=sa.false()), + ) + + # present | absent | inconclusive | manual + op.add_column("upload_items", sa.Column("verification", sa.String(), nullable=True)) + op.add_column( + "upload_items", sa.Column("verified_at", sa.DateTime(timezone=True), nullable=True) + ) + op.add_column("upload_items", sa.Column("observed_sha256", sa.String(), nullable=True)) + op.add_column( + "upload_items", + sa.Column("changed_after_upload", sa.Boolean(), nullable=False, server_default=sa.false()), + ) + + op.create_table( + "upload_verifications", + sa.Column("id", sa.String(), primary_key=True), + sa.Column( + "batch_id", + sa.String(), + sa.ForeignKey("upload_batches.id", ondelete="CASCADE"), + nullable=False, + index=True, + ), + sa.Column("asset_id", sa.String(), nullable=False), + sa.Column("action", sa.String(), nullable=False), + sa.Column("source", sa.String(), nullable=False), + sa.Column("result", sa.String(), nullable=False), + sa.Column("outcome", sa.String(), nullable=True), + sa.Column("evidence", sa.String(), nullable=False), + sa.Column("actor", sa.String(), nullable=True), + sa.Column( + "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now() + ), + ) + + +def downgrade() -> None: + op.drop_table("upload_verifications") + for column in ("changed_after_upload", "observed_sha256", "verified_at", "verification"): + op.drop_column("upload_items", column) + for column in ("stale_bytes", "verified_at"): + op.drop_column("upload_batches", column) diff --git a/photo_pipeline/api/routes/uploads.py b/photo_pipeline/api/routes/uploads.py index 8803ce9..29d60be 100644 --- a/photo_pipeline/api/routes/uploads.py +++ b/photo_pipeline/api/routes/uploads.py @@ -1,9 +1,12 @@ -"""Upload preflight and batch API (US05-01, US05-02). +"""Upload preflight, batch, and verification API (US05-01, US05-02, US05-04). Preflight is a command, not a resource read: it contacts the Immich server, hashes the current bytes, and issues a token. Creating a batch requires that token, and starting one enqueues a durable job on the single ``upload`` lane — the API never runs the uploader in the request thread. + +``verify`` and ``resolve`` are the way out of an uncertain outcome; ``start`` +refuses one with ``409`` rather than letting the browser retry it. """ from __future__ import annotations @@ -19,6 +22,11 @@ from photo_pipeline.services.upload_batches import ( BatchError, UploadBatchService, ) +from photo_pipeline.services.upload_verification import ( + UploadVerificationService, + VerificationError, + retry_blockers, +) from photo_pipeline.services.uploads import UploadError, UploadService router = APIRouter(tags=["uploads"]) @@ -36,6 +44,16 @@ class CreateBatchRequest(PreflightRequest): token: str +class ResolveRequest(BaseModel): + asset_id: str + # uploaded | upgraded | duplicate | skipped | failed | unknown + outcome: str + # What the operator actually checked, and who they are — both mandatory so a + # manual resolution can never look like server evidence. + evidence: str + actor: str + + def _service(request: Request) -> UploadService: return UploadService(request.app.state.session_factory, config=request.app.state.config) @@ -44,6 +62,12 @@ def _batches(request: Request) -> UploadBatchService: return UploadBatchService(request.app.state.session_factory, config=request.app.state.config) +def _verification(request: Request) -> UploadVerificationService: + return UploadVerificationService( + request.app.state.session_factory, config=request.app.state.config + ) + + def _error(status: int, code: str, message: str) -> JSONResponse: return JSONResponse(status_code=status, content={"error": {"code": code, "message": message}}) @@ -61,9 +85,11 @@ def preflight(request: Request, body: PreflightRequest | None = None): def create_batches(body: CreateBatchRequest, request: Request): """Turn an approved preflight into one durable batch per album.""" try: - return {"batches": _batches(request).create( - body.albums, token=body.token, allow_partial=body.allow_partial - )} + return { + "batches": _batches(request).create( + body.albums, token=body.token, allow_partial=body.allow_partial + ) + } except UploadError as error: return _error(422, "unknown_album", str(error)) except BatchConflict as error: @@ -92,6 +118,11 @@ def start_batch(batch_id: str, request: Request): batch = service.get(batch_id) if batch is None: return _error(404, "not_found", f"unknown upload batch {batch_id}") + # An uncertain outcome or bytes changed after upload must be resolved first + # (US05-04); the worker refuses them too, but the user is told here. + blocked = retry_blockers(batch) + if blocked: + return _error(409, blocked[0]["code"], blocked[0]["message"]) try: job = JobService(request.app.state.session_factory).enqueue( UPLOAD_BATCH, @@ -105,6 +136,38 @@ def start_batch(batch_id: str, request: Request): return {"batch_id": batch_id, "job": job} +@router.post("/upload-batches/{batch_id}/verify") +def verify_batch(batch_id: str, request: Request): + """Check the batch against Immich and the bytes on disk (US05-04).""" + try: + return _verification(request).verify(batch_id) + except VerificationError as error: + return _error(404 if error.code == "not_found" else 422, error.code, str(error)) + + +@router.post("/upload-batches/{batch_id}/resolve") +def resolve_item(batch_id: str, body: ResolveRequest, request: Request): + """Record an operator's own verification of one item. Evidence is mandatory.""" + try: + return _verification(request).resolve( + batch_id, + body.asset_id, + outcome=body.outcome, + evidence=body.evidence, + actor=body.actor, + ) + except VerificationError as error: + return _error(404 if error.code == "not_found" else 422, error.code, str(error)) + + +@router.get("/upload-batches/{batch_id}/verifications") +def list_verifications(batch_id: str, request: Request): + batch = _batches(request).get(batch_id) + if batch is None: + return _error(404, "not_found", f"unknown upload batch {batch_id}") + return {"verifications": _verification(request).history(batch_id)} + + @router.post("/upload-batches/{batch_id}/cancel") def cancel_batch(batch_id: str, request: Request): try: diff --git a/photo_pipeline/integrations/immich_go.py b/photo_pipeline/integrations/immich_go.py index 589dc44..d974327 100644 --- a/photo_pipeline/integrations/immich_go.py +++ b/photo_pipeline/integrations/immich_go.py @@ -8,7 +8,9 @@ activity log. Both come from the same builder so the preview can never drift fro the command that would actually run. Server reachability uses ``/api/server/ping`` through stdlib ``urllib`` — the app -has no HTTP client dependency and this is one request. +has no HTTP client dependency and this is one request. :func:`bulk_upload_check` +uses the same client to ask Immich which uploaded bytes it already holds, which is +the authoritative evidence behind upload verification (US05-04). :func:`run_upload` is the only place the uploader is actually executed. It never uses a shell (the argument list goes straight to ``execve``, so no path or album @@ -35,6 +37,11 @@ from pathlib import Path REDACTED = "***" PING_PATH = "/api/server/ping" PING_TIMEOUT_SECONDS = 5.0 +# Immich's own deduplication endpoint: the authoritative answer to "do you already +# have these exact bytes?" used to verify uncertain uploads (US05-04). +BULK_CHECK_PATH = "/api/assets/bulk-upload-check" +CHECK_TIMEOUT_SECONDS = 30.0 +CHECK_BATCH_SIZE = 500 # Reports are kept in full up to this size; beyond it the tail is dropped and the # result is flagged truncated rather than growing without bound (concept §17). MAX_REPORT_BYTES = 4_000_000 @@ -89,6 +96,68 @@ def ping(server_url: str, *, timeout: float = PING_TIMEOUT_SECONDS) -> tuple[boo return False, "server did not answer with pong" +def bulk_upload_check( + server_url: str, + api_key: str | None, + checksums: dict[str, str], + *, + timeout: float = CHECK_TIMEOUT_SECONDS, +) -> dict: + """Ask Immich which of these exact bytes it already holds (US05-04). + + ``checksums`` maps an application key (the asset id) to the SHA-1 of the bytes + that were uploaded — the digest Immich itself deduplicates on. The answer is + ``{"reachable", "detail", "present"}`` where ``present`` maps each key to + ``True`` (the server rejected it as a duplicate, so it holds those bytes), + ``False`` (the server would accept it, so it does not), or ``None`` (the server + answered something this adapter will not interpret). + + An unreachable or unparsable server is reported, never guessed at: the caller + must treat it as uncertainty rather than absence. + """ + if not server_url or not api_key: + return {"reachable": False, "detail": "no Immich credentials configured", "present": {}} + keys = list(checksums) + present: dict[str, bool | None] = {} + for start in range(0, len(keys), CHECK_BATCH_SIZE): + # ponytail: fixed chunk size; make it configurable if a server ever rejects it. + chunk = keys[start : start + CHECK_BATCH_SIZE] + payload = {"assets": [{"id": key, "checksum": checksums[key]} for key in chunk]} + request = urllib.request.Request( # noqa: S310 — http(s) URL from configuration + server_url.rstrip("/") + BULK_CHECK_PATH, + data=json.dumps(payload).encode("utf-8"), + headers={"Content-Type": "application/json", "x-api-key": api_key}, + method="POST", + ) + try: + with urllib.request.urlopen(request, timeout=timeout) as response: # noqa: S310 + body = json.loads(response.read().decode("utf-8") or "{}") + except (urllib.error.URLError, OSError, ValueError, TimeoutError) as error: + return {"reachable": False, "detail": f"{type(error).__name__}: {error}", "present": {}} + results = body.get("results") + if not isinstance(results, list): + return { + "reachable": False, + "detail": "unrecognised bulk-upload-check response", + "present": {}, + } + for result in results: + if not isinstance(result, dict) or result.get("id") not in checksums: + continue + present[result["id"]] = _holds_bytes(result) + return {"reachable": True, "detail": None, "present": present} + + +def _holds_bytes(result: dict) -> bool | None: + """Whether one bulk-upload-check result means the server already has the file.""" + action, reason = result.get("action"), result.get("reason") + if action == "reject": + # Only a duplicate proves possession; "unsupported-format" and friends say + # nothing about whether the bytes are there. + return True if reason == "duplicate" else None + return False if action == "accept" else None + + def build_command( *, binary: str, diff --git a/photo_pipeline/models/__init__.py b/photo_pipeline/models/__init__.py index 09196e3..8dc1023 100644 --- a/photo_pipeline/models/__init__.py +++ b/photo_pipeline/models/__init__.py @@ -14,7 +14,7 @@ from photo_pipeline.models.duplicates import ( from photo_pipeline.models.jobs import Job, JobEvent, JobItem from photo_pipeline.models.renames import RenameOperation, RenamePlan from photo_pipeline.models.thumbnails import Thumbnail -from photo_pipeline.models.uploads import UploadBatch, UploadItem +from photo_pipeline.models.uploads import UploadBatch, UploadItem, UploadVerification from photo_pipeline.models.workflow import AnalysisResult, SafetyReview __all__ = [ @@ -32,6 +32,7 @@ __all__ = [ "Thumbnail", "UploadBatch", "UploadItem", + "UploadVerification", "SafetyReview", "AnalysisResult", ] diff --git a/photo_pipeline/models/uploads.py b/photo_pipeline/models/uploads.py index 7bafa7a..381e1b1 100644 --- a/photo_pipeline/models/uploads.py +++ b/photo_pipeline/models/uploads.py @@ -69,6 +69,11 @@ class UploadBatch(Base): outcome_counts: Mapped[str | None] = mapped_column(String) # JSON, from the items report_counts: Mapped[str | None] = mapped_column(String) # JSON, uploader's own + # Verification (US05-04). ``stale_bytes`` means at least one uploaded file has + # been edited since: the batch carries a visible warning and cannot be re-run. + verified_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + stale_bytes: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + error_code: Mapped[str | None] = mapped_column(String) error_message: Mapped[str | None] = mapped_column(String) @@ -102,6 +107,45 @@ class UploadItem(Base): outcome: Mapped[str | None] = mapped_column(String) evidence: Mapped[str | None] = mapped_column(String) # the bounded report line outcome_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + + # Verification against the server (US05-04): present | absent | inconclusive | + # manual. NULL until the item has been verified; ``inconclusive`` whenever the + # server could not answer — which is never treated as success. + verification: Mapped[str | None] = mapped_column(String) + verified_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + # The bytes on disk at verification time, and whether they still are the bytes + # this batch uploaded. + observed_sha256: Mapped[str | None] = mapped_column(String) + changed_after_upload: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now() ) + + +class UploadVerification(Base): + """Append-only evidence for every verification and manual resolution (US05-04). + + The item row is a projection of the latest answer; this table is the history + that answers "who decided this, on what evidence, and when?". Rows are never + updated or deleted, so a manual resolution can always be told apart from + server evidence. + """ + + __tablename__ = "upload_verifications" + + id: Mapped[str] = mapped_column(String, primary_key=True) + batch_id: Mapped[str] = mapped_column( + ForeignKey("upload_batches.id", ondelete="CASCADE"), nullable=False, index=True + ) + asset_id: Mapped[str] = mapped_column(String, nullable=False) + action: Mapped[str] = mapped_column(String, nullable=False) # verify | resolve + source: Mapped[str] = mapped_column(String, nullable=False) # immich_api | operator + # present | absent | inconclusive for a verify; the recorded outcome for a resolve. + result: Mapped[str] = mapped_column(String, nullable=False) + outcome: Mapped[str | None] = mapped_column(String) + evidence: Mapped[str] = mapped_column(String, nullable=False) + actor: Mapped[str | None] = mapped_column(String) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) diff --git a/photo_pipeline/services/upload_batches.py b/photo_pipeline/services/upload_batches.py index 6eaac46..b048d40 100644 --- a/photo_pipeline/services/upload_batches.py +++ b/photo_pipeline/services/upload_batches.py @@ -16,7 +16,8 @@ cannot undo, so the discipline is: - **an interrupted attempt is uncertain, not failed.** Immich may have accepted files the app never saw a report for, so ``recover`` marks a batch whose worker vanished ``unknown_requires_verification`` (concept §15) instead of retrying it - blindly. Resolving that is US05-04. + blindly. :func:`~photo_pipeline.services.upload_verification.retry_blockers` + (US05-04) is what decides whether an attempt may start at all. Per-asset upload *results* are not interpreted here: after the attempt ends the report is handed to :class:`~photo_pipeline.services.upload_reports. @@ -40,6 +41,7 @@ from photo_pipeline.integrations import immich_go from photo_pipeline.models import UploadBatch, UploadItem from photo_pipeline.services.hashing import sha1_file from photo_pipeline.services.upload_reports import UploadReportService +from photo_pipeline.services.upload_verification import retry_blockers from photo_pipeline.services.uploads import UploadService @@ -62,6 +64,7 @@ RUNNABLE_STATES = frozenset({BatchState.PLANNED, BatchState.FAILED, BatchState.C # States whose batch is still the live one for its album. OPEN_STATES = frozenset({BatchState.PLANNED, BatchState.RUNNING, BatchState.CANCELLING}) + class ItemState: PENDING = "pending" # ``sent`` means the batch process exited cleanly, not that Immich confirmed the @@ -162,13 +165,11 @@ class UploadBatchService: def run(self, batch_id: str, *, worker_id: str = "uploader", cancelled=None) -> dict: """Run one batch to completion. Blocks for the duration of the upload.""" batch = self._require(batch_id) - if batch["state"] not in RUNNABLE_STATES: - code = ( - "requires_verification" - if batch["state"] == BatchState.UNKNOWN - else "not_runnable" - ) - raise BatchError(code, f"batch {batch_id} is {batch['state']}") + # Retry policy (US05-04): a safe failure may run again, an uncertain outcome + # or bytes edited after upload may not. + blocked = retry_blockers(batch) + if blocked: + raise BatchError(blocked[0]["code"], blocked[0]["message"]) busy = [row for row in self.list() if row["id"] != batch_id and row["state"] in LANE_STATES] if busy: raise BatchConflict("lane_busy", f"upload batch {busy[0]['id']} is still running") @@ -189,9 +190,7 @@ class UploadBatchService: token = self._claim(batch_id, worker_id=worker_id) attempt = self.get(batch_id)["attempt_count"] - report_path = ( - Path(self._config.data_dir) / "uploads" / f"{batch_id}-attempt-{attempt}.log" - ) + report_path = Path(self._config.data_dir) / "uploads" / f"{batch_id}-attempt-{attempt}.log" key = self._config.immich_api_key.get_secret_value() if self._config.immich_api_key else "" command = immich_go.build_command( binary=self._config.immich_go_binary, @@ -426,6 +425,10 @@ def _batch_dict(row: UploadBatch, items: list[UploadItem]) -> dict: "outcome_state": row.outcome_state, "outcome_counts": json.loads(row.outcome_counts) if row.outcome_counts else None, "report_counts": json.loads(row.report_counts) if row.report_counts else None, + # Verification evidence (US05-04). ``stale_bytes`` is the visible warning + # that an uploaded file has since been edited. + "verified_at": row.verified_at.isoformat() if row.verified_at else None, + "stale_bytes": row.stale_bytes, "started_at": row.started_at.isoformat() if row.started_at else None, "finished_at": row.finished_at.isoformat() if row.finished_at else None, "items": [ @@ -438,6 +441,10 @@ def _batch_dict(row: UploadBatch, items: list[UploadItem]) -> dict: "outcome": item.outcome, "evidence": item.evidence, "outcome_at": item.outcome_at.isoformat() if item.outcome_at else None, + "verification": item.verification, + "verified_at": item.verified_at.isoformat() if item.verified_at else None, + "observed_sha256": item.observed_sha256, + "changed_after_upload": item.changed_after_upload, } for item in items ], diff --git a/photo_pipeline/services/upload_verification.py b/photo_pipeline/services/upload_verification.py new file mode 100644 index 0000000..f6760f7 --- /dev/null +++ b/photo_pipeline/services/upload_verification.py @@ -0,0 +1,349 @@ +"""UploadVerificationService — resolving uncertain uploads (US05-04). + +US05-03 leaves a batch honest but sometimes uncertain: a killed uploader, an +unpinned report grammar, or a file the report never mentioned all end as +``unknown``. Retrying such a batch is the dangerous move — Immich may already hold +the files, and a blind retry is how a lost response turns into a second server +asset. So this service resolves uncertainty *before* anything is re-run: + +- **evidence beats the report.** Verification asks Immich itself whether it holds + the exact SHA-1 the batch recorded before uploading. That answer is authoritative + over the uploader's text, in both directions: present makes an unknown item + ``uploaded``, absent makes it ``failed`` and therefore safe to retry. +- **no answer is never "no".** An unreachable server, missing credentials, or a + response this adapter will not interpret leave the item ``inconclusive``. The + batch stays ``unknown_requires_verification`` and stays un-runnable. +- **changed bytes are a stale warning, not a silent re-upload.** Verification + re-hashes what is on disk. A file edited after its upload is flagged, the batch + is marked ``stale_bytes``, and re-running it is refused: uploading again would + create or upgrade a server asset the user never approved (concept §8). +- **manual resolution is evidence, not permission.** An operator may record what + they checked in Immich, but only with a non-empty note and their identity, and + every decision is appended to an immutable history alongside the server answers. + +Retry policy lives in :func:`retry_blockers`, which :class:`UploadBatchService` +enforces before every attempt and the API surfaces as ``409``. +""" + +from __future__ import annotations + +import uuid +from datetime import datetime, timezone +from pathlib import Path + +from sqlalchemy import select +from sqlalchemy.orm import sessionmaker + +from photo_pipeline.config import Config +from photo_pipeline.integrations import immich_go +from photo_pipeline.integrations import immich_go_report as report_parser +from photo_pipeline.models import UploadBatch, UploadItem, UploadVerification +from photo_pipeline.services.hashing import sha256_file +from photo_pipeline.services.upload_reports import REQUIRES_VERIFICATION, VERIFIED + +PRESENT = "present" +ABSENT = "absent" +INCONCLUSIVE = "inconclusive" +MANUAL = "manual" + +MAX_EVIDENCE_CHARS = 500 + + +def _now() -> datetime: + return datetime.now(timezone.utc) + + +class VerificationError(RuntimeError): + """The request cannot be carried out (unknown batch/item, missing evidence).""" + + def __init__(self, code: str, message: str) -> None: + super().__init__(message) + self.code = code + + +class UploadVerificationService: + def __init__(self, session_factory: sessionmaker, *, config: Config) -> None: + self._session_factory = session_factory + self._config = config + + # ── verification ────────────────────────────────────────────────────────── + + def verify(self, batch_id: str) -> dict: + """Check every item of a batch against the server and the bytes on disk. + + Returns ``{"batch_id", "state", "outcome_state", "stale_bytes", + "server_reachable", "detail", "counts", "items"}``. Safe to call + repeatedly: the same evidence produces the same rows, and each run appends + its own history entries. + """ + with self._session_factory() as session: + batch = session.get(UploadBatch, batch_id) + if batch is None: + raise VerificationError("not_found", f"unknown upload batch {batch_id!r}") + items = list( + session.scalars( + select(UploadItem) + .where(UploadItem.batch_id == batch_id) + .order_by(UploadItem.path) + ) + ) + answer = immich_go.bulk_upload_check( + self._config.immich_server_url, + self._config.immich_api_key.get_secret_value() + if self._config.immich_api_key + else None, + {item.asset_id: item.sha1 for item in items if item.sha1}, + ) + detail = answer["detail"] + + for item in items: + observed, changed = _current_bytes(item) + held = answer["present"].get(item.asset_id) + if item.sha1 is None: + result, evidence = INCONCLUSIVE, "no upload hash was recorded for this file" + elif held is True: + result, evidence = PRESENT, f"Immich holds sha1 {item.sha1}" + elif held is False: + result, evidence = ABSENT, f"Immich does not hold sha1 {item.sha1}" + else: + result = INCONCLUSIVE + evidence = detail or "the server did not classify these bytes" + + item.verification = result + item.verified_at = _now() + item.observed_sha256 = observed + item.changed_after_upload = changed + # The server is authoritative over the report text — but only when + # it actually answered. + if result == PRESENT: + item.outcome = report_parser.UPLOADED + item.evidence = evidence + item.outcome_at = _now() + elif result == ABSENT: + item.outcome = report_parser.FAILED + item.evidence = evidence + item.outcome_at = _now() + session.add( + _event( + batch_id, + item.asset_id, + action="verify", + source="immich_api", + result=result, + outcome=item.outcome, + evidence=evidence, + ) + ) + + _resolve_batch(batch, items) + session.commit() + return _report(batch, items, server_reachable=answer["reachable"], detail=detail) + + # ── manual resolution ───────────────────────────────────────────────────── + + def resolve( + self, batch_id: str, asset_id: str, *, outcome: str, evidence: str, actor: str + ) -> dict: + """Record an operator's own verification of one item. + + ``evidence`` and ``actor`` are mandatory: a manual resolution is only worth + keeping if it says what was checked and who checked it. + """ + if outcome not in report_parser.OUTCOMES: + raise VerificationError("invalid_outcome", f"unknown upload outcome {outcome!r}") + evidence = (evidence or "").strip() + actor = (actor or "").strip() + if not evidence: + raise VerificationError("evidence_required", "a manual resolution must record evidence") + if not actor: + raise VerificationError("actor_required", "a manual resolution must record its author") + + with self._session_factory() as session: + batch = session.get(UploadBatch, batch_id) + if batch is None: + raise VerificationError("not_found", f"unknown upload batch {batch_id!r}") + item = session.get(UploadItem, {"batch_id": batch_id, "asset_id": asset_id}) + if item is None: + raise VerificationError("not_found", f"{asset_id!r} is not part of this batch") + + item.outcome = outcome + item.evidence = evidence[:MAX_EVIDENCE_CHARS] + item.outcome_at = _now() + item.verification = MANUAL + item.verified_at = _now() + session.add( + _event( + batch_id, + asset_id, + action="resolve", + source="operator", + result=MANUAL, + outcome=outcome, + evidence=evidence, + actor=actor, + ) + ) + + items = list( + session.scalars( + select(UploadItem) + .where(UploadItem.batch_id == batch_id) + .order_by(UploadItem.path) + ) + ) + _resolve_batch(batch, items) + session.commit() + return _report(batch, items, server_reachable=None, detail=None) + + # ── history ─────────────────────────────────────────────────────────────── + + def history(self, batch_id: str) -> list[dict]: + """Every verification and resolution recorded for this batch, oldest first.""" + with self._session_factory() as session: + rows = session.scalars( + select(UploadVerification) + .where(UploadVerification.batch_id == batch_id) + .order_by(UploadVerification.created_at, UploadVerification.id) + ) + return [ + { + "id": row.id, + "asset_id": row.asset_id, + "action": row.action, + "source": row.source, + "result": row.result, + "outcome": row.outcome, + "evidence": row.evidence, + "actor": row.actor, + "created_at": row.created_at.isoformat() if row.created_at else None, + } + for row in rows + ] + + +# ── retry policy ───────────────────────────────────────────────────────────── + + +def retry_blockers(batch: dict) -> list[dict]: + """Why this batch may not be (re)run, in the order the user should fix them. + + A plain uploader failure is a *safe* failure: nothing uncertain happened, so it + is retryable. An uncertain outcome and edited bytes are not. + """ + from photo_pipeline.services.upload_batches import BatchState, RUNNABLE_STATES + + blockers: list[dict] = [] + if batch["state"] == BatchState.UNKNOWN: + blockers.append( + { + "code": "requires_verification", + "message": "this upload's outcome is uncertain; verify it before retrying", + } + ) + if batch.get("stale_bytes"): + blockers.append( + { + "code": "changed_after_upload", + "message": "files in this batch changed after they were uploaded; " + "re-approve them through a fresh preflight", + } + ) + if not blockers and batch["state"] not in RUNNABLE_STATES: + blockers.append({"code": "not_runnable", "message": f"batch is {batch['state']}"}) + return blockers + + +# ── internals ──────────────────────────────────────────────────────────────── + + +def _current_bytes(item: UploadItem) -> tuple[str | None, bool]: + """``(hash on disk now, changed since upload)``. A missing file counts as changed.""" + path = Path(item.path) + if not path.exists(): + return None, item.sha256 is not None + observed = sha256_file(path) + return observed, bool(item.sha256 and observed != item.sha256) + + +def _event( + batch_id: str, + asset_id: str, + *, + action: str, + source: str, + result: str, + outcome: str | None, + evidence: str, + actor: str | None = None, +) -> UploadVerification: + return UploadVerification( + id=str(uuid.uuid4()), + batch_id=batch_id, + asset_id=asset_id, + action=action, + source=source, + result=result, + outcome=outcome, + evidence=evidence[:MAX_EVIDENCE_CHARS], + actor=actor, + created_at=_now(), + ) + + +def _resolve_batch(batch: UploadBatch, items: list[UploadItem]) -> None: + """Fold the item evidence back into the batch's own state. + + An uncertain batch only leaves that state once every item is accounted for: + all present makes it succeeded, any absent makes it a safe failure to retry, + and a single inconclusive item keeps it uncertain. + """ + from photo_pipeline.services.upload_batches import BatchState + + batch.stale_bytes = any(item.changed_after_upload for item in items) + batch.verified_at = _now() + unresolved = [item for item in items if item.outcome in (None, report_parser.UNKNOWN)] + batch.outcome_state = VERIFIED if items and not unresolved else REQUIRES_VERIFICATION + + if batch.state != BatchState.UNKNOWN or unresolved: + return + if any(item.outcome == report_parser.FAILED for item in items): + batch.state = BatchState.FAILED + batch.error_code = "verified_incomplete" + batch.error_message = "verification proved some files never reached Immich" + else: + batch.state = BatchState.SUCCEEDED + batch.error_code = batch.error_message = None + + +def _report( + batch: UploadBatch, + items: list[UploadItem], + *, + server_reachable: bool | None, + detail: str | None, +) -> dict: + counts: dict[str, int] = {} + for item in items: + key = item.verification or "unverified" + counts[key] = counts.get(key, 0) + 1 + return { + "batch_id": batch.id, + "state": batch.state, + "outcome_state": batch.outcome_state, + "stale_bytes": batch.stale_bytes, + "server_reachable": server_reachable, + "detail": detail, + "counts": counts, + "items": [ + { + "asset_id": item.asset_id, + "path": item.path, + "verification": item.verification, + "outcome": item.outcome, + "evidence": item.evidence, + "changed_after_upload": item.changed_after_upload, + "verified_at": item.verified_at.isoformat() if item.verified_at else None, + } + for item in items + ], + } diff --git a/tests/integration/test_upload_verification.py b/tests/integration/test_upload_verification.py new file mode 100644 index 0000000..acc02af --- /dev/null +++ b/tests/integration/test_upload_verification.py @@ -0,0 +1,535 @@ +"""Verifying, retrying, and resolving uncertain uploads (US05-04). + +The Immich boundary is a real HTTP server here: it answers ``/api/server/ping`` +and ``/api/assets/bulk-upload-check`` exactly as the app's own adapter parses +them, and the set of checksums it "holds" is what each fault scenario controls. +The uploader stays a real executable driven through the real batch service, so +every state under test is reached the way production reaches it. + +The invariant these tests defend is one-directional: uncertainty may only become +success when something authoritative said so — the server, or an operator who +recorded what they checked. +""" + +import json +import stat +import threading +import uuid +from datetime import datetime, timezone +from http.server import BaseHTTPRequestHandler, HTTPServer +from pathlib import Path + +import pytest +from fastapi.testclient import TestClient + +from photo_pipeline.api.app import create_app +from photo_pipeline.config import Config +from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations +from photo_pipeline.models import AnalysisResult, Asset, SafetyReview, UploadBatch +from photo_pipeline.services.hashing import sha256_file +from photo_pipeline.services.upload_batches import BatchError, BatchState, UploadBatchService +from photo_pipeline.services.upload_reports import REQUIRES_VERIFICATION, VERIFIED +from photo_pipeline.services.upload_verification import ( + ABSENT, + INCONCLUSIVE, + MANUAL, + PRESENT, + UploadVerificationService, + VerificationError, + retry_blockers, +) +from photo_pipeline.services.uploads import UploadService + +NOW = datetime(2026, 1, 1, tzinfo=timezone.utc) +SENTINEL_KEY = "immich-sentinel-9f3a2b" +UPLOADER_VERSION = "immich-go 0.21.0" + + +# ── fake Immich ────────────────────────────────────────────────────────────── + + +class _FakeImmich: + """The server's view of the world: which checksums it holds, and whether it + is willing to answer at all.""" + + def __init__(self) -> None: + self.held: set[str] = set() + self.available = True + self.checked: list[str] = [] + + +def _handler(state: _FakeImmich): + class Handler(BaseHTTPRequestHandler): + def do_GET(self): # noqa: N802 (BaseHTTPRequestHandler API) + self._json(200, {"res": "pong"}) + + def do_POST(self): # noqa: N802 + body = json.loads(self.rfile.read(int(self.headers["Content-Length"] or 0)) or "{}") + if not state.available: + self._json(503, {"error": "unavailable"}) + return + if self.headers.get("x-api-key") != SENTINEL_KEY: + self._json(401, {"error": "unauthorized"}) + return + results = [] + for asset in body.get("assets", []): + state.checked.append(asset["checksum"]) + results.append( + {"id": asset["id"], "action": "reject", "reason": "duplicate"} + if asset["checksum"] in state.held + else {"id": asset["id"], "action": "accept"} + ) + self._json(200, {"results": results}) + + def _json(self, status: int, payload: dict) -> None: + body = json.dumps(payload).encode() + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, *args): + pass + + return Handler + + +@pytest.fixture +def immich(): + state = _FakeImmich() + server = HTTPServer(("127.0.0.1", 0), _handler(state)) + threading.Thread(target=server.serve_forever, daemon=True).start() + state.url = f"http://127.0.0.1:{server.server_port}" + yield state + server.shutdown() + server.server_close() + + +# ── environment ────────────────────────────────────────────────────────────── + + +def _uploader(tmp_path, report_body: str = "", *, exit_code: int = 0, version=UPLOADER_VERSION): + path = tmp_path / "immich-go" + path.write_text( + "#!/bin/sh\n" + f'if [ "$1" = "--version" ]; then echo "{version}"; exit 0; fi\n' + f"cat <<'REPORT'\n{report_body}\nREPORT\n" + f"exit {exit_code}\n" + ) + path.chmod(path.stat().st_mode | stat.S_IEXEC | stat.S_IXGRP | stat.S_IXOTH) + return path + + +def _env(tmp_path, immich, uploader=None): + (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_IMMICH_SERVER_URL": immich.url, + "PHOTO_PIPELINE_IMMICH_API_KEY": SENTINEL_KEY, + "PHOTO_PIPELINE_IMMICH_GO_BINARY": str( + uploader if uploader is not None else _uploader(tmp_path) + ), + } + ) + run_migrations(config.database_url) + return config, create_session_factory(create_db_engine(config.database_url)), lib + + +def _album(sf, lib, album="rome", names=("a.jpg", "b.jpg")): + folder = lib / album + folder.mkdir(parents=True, exist_ok=True) + with sf() as session: + for name in names: + path = folder / name + path.write_bytes(f"{album}/{name}".encode() * 16) + 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=sha256_file(path), + ) + ) + session.add( + SafetyReview( + id=str(uuid.uuid4()), asset_id=asset_id, decision="sfw", exif_verified_at=NOW + ) + ) + session.add(AnalysisResult(asset_id=asset_id, status="analyzed", exif_written_at=NOW)) + session.commit() + return folder + + +def _batch(sf, config, albums=None): + report = UploadService(sf, config=config).preflight(albums) + assert report["state"] == "ready", report["blockers"] + (batch,) = UploadBatchService(sf, config=config).create(albums, token=report["token"]) + return batch + + +def _interrupt(sf, config, batch_id): + """Leave behind exactly what a killed worker leaves: a mid-flight attempt.""" + with sf() as session: + session.get(UploadBatch, batch_id).state = BatchState.RUNNING + session.commit() + UploadBatchService(sf, config=config).recover() + + +def _server_holds(immich, batch): + immich.held.update(item["sha1"] for item in batch["items"]) + + +def _by_name(batch) -> dict: + return {Path(item["path"]).name: item for item in batch["items"]} + + +def _uncertain_batch(tmp_path, immich, *, hold: bool): + """A batch whose attempt died mid-upload, with the server holding the bytes or not.""" + config, sf, lib = _env(tmp_path, immich) + _album(sf, lib) + batch = _batch(sf, config) + if hold: + _server_holds(immich, batch) + _interrupt(sf, config, batch["id"]) + return config, sf, lib, batch + + +# ── accepted-but-unknown ───────────────────────────────────────────────────── + + +def test_acceptance_then_lost_response_is_confirmed_by_the_server(tmp_path, immich): + """The classic lost response: Immich took the files, the app never saw a report.""" + config, sf, lib, batch = _uncertain_batch(tmp_path, immich, hold=True) + + result = UploadVerificationService(sf, config=config).verify(batch["id"]) + + assert result["state"] == BatchState.SUCCEEDED, "server evidence resolves the uncertainty" + assert result["outcome_state"] == VERIFIED + assert result["counts"] == {PRESENT: 2} + verified = UploadBatchService(sf, config=config).get(batch["id"]) + for item in verified["items"]: + assert item["outcome"] == "uploaded" + assert item["verification"] == PRESENT + assert item["sha1"] in item["evidence"], "the exact bytes are named in the evidence" + assert immich.checked, "verification actually asked the server" + + +def test_timeout_before_acceptance_leaves_a_safe_retry(tmp_path, immich): + """Nothing arrived, so the batch becomes a plain failure and may run again.""" + config, sf, lib, batch = _uncertain_batch(tmp_path, immich, hold=False) + service = UploadBatchService(sf, config=config) + + result = UploadVerificationService(sf, config=config).verify(batch["id"]) + + assert result["state"] == BatchState.FAILED + assert result["counts"] == {ABSENT: 2} + assert {item["outcome"] for item in service.get(batch["id"])["items"]} == {"failed"} + assert retry_blockers(service.get(batch["id"])) == [] + assert service.run(batch["id"])["state"] == BatchState.SUCCEEDED + + +def test_an_uncertain_batch_cannot_be_retried_before_verification(tmp_path, immich): + config, sf, lib, batch = _uncertain_batch(tmp_path, immich, hold=True) + service = UploadBatchService(sf, config=config) + + with pytest.raises(BatchError) as error: + service.run(batch["id"]) + + assert error.value.code == "requires_verification" + with TestClient(create_app(config)) as client: + response = client.post(f"/api/v1/upload-batches/{batch['id']}/start") + assert response.status_code == 409 + assert response.json()["error"]["code"] == "requires_verification" + + +def test_a_partly_arrived_batch_is_a_failure_not_a_success(tmp_path, immich): + config, sf, lib = _env(tmp_path, immich) + _album(sf, lib) + batch = _batch(sf, config) + immich.held.add(batch["items"][0]["sha1"]) # only one file made it + _interrupt(sf, config, batch["id"]) + + result = UploadVerificationService(sf, config=config).verify(batch["id"]) + + assert result["state"] == BatchState.FAILED + assert result["counts"] == {PRESENT: 1, ABSENT: 1} + assert result["outcome_state"] == VERIFIED, "every file is accounted for" + + +# ── parser uncertainty ─────────────────────────────────────────────────────── + + +def test_parser_uncertainty_is_settled_by_server_evidence(tmp_path, immich): + """An unpinned uploader version leaves every item unknown; the server decides.""" + config, sf, lib = _env(tmp_path, immich, _uploader(tmp_path, "done", version="immich-go 9.9.9")) + _album(sf, lib) + batch = _batch(sf, config) + service = UploadBatchService(sf, config=config) + ran = service.run(batch["id"]) + assert ran["state"] == BatchState.SUCCEEDED and ran["outcome_state"] == REQUIRES_VERIFICATION + _server_holds(immich, ran) + + result = UploadVerificationService(sf, config=config).verify(batch["id"]) + + assert result["outcome_state"] == VERIFIED + assert {item["outcome"] for item in service.get(batch["id"])["items"]} == {"uploaded"} + + +def test_an_unreachable_server_never_turns_uncertainty_into_success(tmp_path, immich): + config, sf, lib, batch = _uncertain_batch(tmp_path, immich, hold=True) + immich.available = False + service = UploadBatchService(sf, config=config) + + result = UploadVerificationService(sf, config=config).verify(batch["id"]) + + assert result["server_reachable"] is False + assert result["detail"], "the reason the server could not answer is reported" + assert result["counts"] == {INCONCLUSIVE: 2} + assert result["state"] == BatchState.UNKNOWN, "still uncertain, not succeeded" + assert result["outcome_state"] == REQUIRES_VERIFICATION + assert [b["code"] for b in retry_blockers(service.get(batch["id"]))] == [ + "requires_verification" + ] + + +# ── safe failures ──────────────────────────────────────────────────────────── + + +def test_a_plain_uploader_failure_is_retryable_without_verification(tmp_path, immich): + """Nothing uncertain happened: the process failed before/while reporting an error.""" + config, sf, lib = _env(tmp_path, immich, _uploader(tmp_path, "boom", exit_code=1)) + _album(sf, lib) + batch = _batch(sf, config) + service = UploadBatchService(sf, config=config) + assert service.run(batch["id"])["state"] == BatchState.FAILED + + assert retry_blockers(service.get(batch["id"])) == [] + with TestClient(create_app(config)) as client: + assert client.post(f"/api/v1/upload-batches/{batch['id']}/start").status_code == 200 + + +# ── changed bytes ──────────────────────────────────────────────────────────── + + +def test_bytes_changed_after_upload_warn_and_block_a_rerun(tmp_path, immich): + config, sf, lib = _env(tmp_path, immich) + folder = _album(sf, lib) + batch = _batch(sf, config) + service = UploadBatchService(sf, config=config) + ran = service.run(batch["id"]) + _server_holds(immich, ran) + (folder / "a.jpg").write_bytes(b"edited after the upload") + + result = UploadVerificationService(sf, config=config).verify(batch["id"]) + + assert result["stale_bytes"] is True + changed = _by_name(result)["a.jpg"] + assert changed["changed_after_upload"] is True + assert _by_name(result)["b.jpg"]["changed_after_upload"] is False + stored = service.get(batch["id"]) + assert stored["stale_bytes"] is True, "the warning is durable, not only in the response" + assert _by_name(stored)["a.jpg"]["observed_sha256"] != _by_name(stored)["a.jpg"]["sha256"] + with pytest.raises(BatchError) as error: + service.run(batch["id"]) + assert error.value.code == "changed_after_upload" + with TestClient(create_app(config)) as client: + response = client.post(f"/api/v1/upload-batches/{batch['id']}/start") + assert response.status_code == 409 + assert response.json()["error"]["code"] == "changed_after_upload" + + +def test_a_deleted_file_counts_as_changed_after_upload(tmp_path, immich): + config, sf, lib = _env(tmp_path, immich) + folder = _album(sf, lib) + batch = _batch(sf, config) + ran = UploadBatchService(sf, config=config).run(batch["id"]) + _server_holds(immich, ran) + (folder / "a.jpg").unlink() + + result = UploadVerificationService(sf, config=config).verify(batch["id"]) + + assert _by_name(result)["a.jpg"]["changed_after_upload"] is True + assert result["stale_bytes"] is True + # The bytes are still on the server: verification is about the upload, not the + # local file's continued existence. + assert _by_name(result)["a.jpg"]["verification"] == PRESENT + + +# ── repeated verification and history ──────────────────────────────────────── + + +def test_repeated_verification_converges_and_keeps_every_answer(tmp_path, immich): + config, sf, lib, batch = _uncertain_batch(tmp_path, immich, hold=True) + service = UploadVerificationService(sf, config=config) + + first = service.verify(batch["id"]) + second = service.verify(batch["id"]) + + assert first["counts"] == second["counts"] == {PRESENT: 2} + assert first["state"] == second["state"] == BatchState.SUCCEEDED + history = service.history(batch["id"]) + assert len(history) == 4, "the audit trail appends, it never overwrites" + assert {entry["source"] for entry in history} == {"immich_api"} + assert all(entry["action"] == "verify" for entry in history) + + +def test_verification_that_changes_its_mind_keeps_both_answers(tmp_path, immich): + """A file that was absent and later present must show both, in order.""" + config, sf, lib, batch = _uncertain_batch(tmp_path, immich, hold=False) + service = UploadVerificationService(sf, config=config) + service.verify(batch["id"]) + _server_holds(immich, batch) + + service.verify(batch["id"]) + + asset_id = batch["items"][0]["asset_id"] + results = [e["result"] for e in service.history(batch["id"]) if e["asset_id"] == asset_id] + assert results == [ABSENT, PRESENT] + + +# ── manual resolution ──────────────────────────────────────────────────────── + + +def test_manual_resolution_requires_evidence_and_an_author(tmp_path, immich): + config, sf, lib, batch = _uncertain_batch(tmp_path, immich, hold=True) + service = UploadVerificationService(sf, config=config) + asset_id = batch["items"][0]["asset_id"] + + for kwargs, code in ( + ({"evidence": " ", "actor": "dom"}, "evidence_required"), + ({"evidence": "checked in Immich", "actor": ""}, "actor_required"), + ({"evidence": "checked", "actor": "dom", "outcome": "definitely-fine"}, "invalid_outcome"), + ): + with pytest.raises(VerificationError) as error: + service.resolve(batch["id"], asset_id, **{"outcome": "uploaded", **kwargs}) + assert error.value.code == code + + assert service.history(batch["id"]) == [], "a refused resolution records nothing" + assert UploadBatchService(sf, config=config).get(batch["id"])["state"] == BatchState.UNKNOWN + + +def test_manual_resolution_is_recorded_as_operator_evidence(tmp_path, immich): + """An operator may settle what the server cannot — but never anonymously.""" + config, sf, lib, batch = _uncertain_batch(tmp_path, immich, hold=True) + immich.available = False + service = UploadVerificationService(sf, config=config) + service.verify(batch["id"]) # inconclusive: the server is down + + for item in batch["items"]: + result = service.resolve( + batch["id"], + item["asset_id"], + outcome="uploaded", + evidence="found in Immich by checksum in the web UI", + actor="dom", + ) + + assert result["state"] == BatchState.SUCCEEDED + assert result["outcome_state"] == VERIFIED + resolutions = [e for e in service.history(batch["id"]) if e["action"] == "resolve"] + assert len(resolutions) == 2 + assert {e["actor"] for e in resolutions} == {"dom"} + assert {e["source"] for e in resolutions} == {"operator"} + assert all("web UI" in e["evidence"] for e in resolutions) + stored = UploadBatchService(sf, config=config).get(batch["id"]) + assert {item["verification"] for item in stored["items"]} == {MANUAL}, ( + "a manual answer stays distinguishable from server evidence" + ) + + +def test_resolving_one_item_leaves_the_batch_uncertain(tmp_path, immich): + config, sf, lib, batch = _uncertain_batch(tmp_path, immich, hold=True) + immich.available = False + service = UploadVerificationService(sf, config=config) + service.verify(batch["id"]) + + result = service.resolve( + batch["id"], + batch["items"][0]["asset_id"], + outcome="uploaded", + evidence="visible in Immich", + actor="dom", + ) + + assert result["state"] == BatchState.UNKNOWN + assert result["outcome_state"] == REQUIRES_VERIFICATION + + +def test_resolving_an_unknown_item_or_batch_is_refused(tmp_path, immich): + config, sf, lib, batch = _uncertain_batch(tmp_path, immich, hold=True) + service = UploadVerificationService(sf, config=config) + + for batch_id, asset_id in ((batch["id"], "not-in-this-batch"), ("no-such-batch", "x")): + with pytest.raises(VerificationError) as error: + service.resolve(batch_id, asset_id, outcome="uploaded", evidence="checked", actor="dom") + assert error.value.code == "not_found" + + +def test_verifying_an_unknown_batch_is_refused(tmp_path, immich): + config, sf, lib = _env(tmp_path, immich) + + with pytest.raises(VerificationError) as error: + UploadVerificationService(sf, config=config).verify("does-not-exist") + + assert error.value.code == "not_found" + + +# ── API surface and durability ─────────────────────────────────────────────── + + +def test_the_api_verifies_resolves_and_lists_history_without_secrets(tmp_path, immich): + config, sf, lib, batch = _uncertain_batch(tmp_path, immich, hold=False) + + with TestClient(create_app(config)) as client: + verified = client.post(f"/api/v1/upload-batches/{batch['id']}/verify") + resolved = client.post( + f"/api/v1/upload-batches/{batch['id']}/resolve", + json={ + "asset_id": batch["items"][0]["asset_id"], + "outcome": "skipped", + "evidence": "the file was withdrawn from the album", + "actor": "dom", + }, + ) + refused = client.post( + f"/api/v1/upload-batches/{batch['id']}/resolve", + json={ + "asset_id": batch["items"][0]["asset_id"], + "outcome": "uploaded", + "evidence": "", + "actor": "dom", + }, + ) + history = client.get(f"/api/v1/upload-batches/{batch['id']}/verifications") + missing = client.post("/api/v1/upload-batches/does-not-exist/verify") + + assert verified.status_code == 200 and verified.json()["counts"] == {ABSENT: 2} + assert resolved.status_code == 200 + assert refused.status_code == 422 and refused.json()["error"]["code"] == "evidence_required" + assert len(history.json()["verifications"]) == 3 + assert missing.status_code == 404 + assert SENTINEL_KEY not in verified.text + resolved.text + history.text + + +def test_verification_survives_a_restart(tmp_path, immich): + config, sf, lib, batch = _uncertain_batch(tmp_path, immich, hold=True) + UploadVerificationService(sf, config=config).verify(batch["id"]) + + with TestClient(create_app(config)) as client: # a fresh application process + fetched = client.get(f"/api/v1/upload-batches/{batch['id']}").json() + history = client.get(f"/api/v1/upload-batches/{batch['id']}/verifications").json() + + assert fetched["state"] == BatchState.SUCCEEDED + assert fetched["verified_at"] + assert {item["verification"] for item in fetched["items"]} == {PRESENT} + assert len(history["verifications"]) == 2 + assert json.dumps(fetched) # the record stays JSON-serialisable for the UI diff --git a/tests/story_traceability.json b/tests/story_traceability.json index 0998374..24b4c27 100644 --- a/tests/story_traceability.json +++ b/tests/story_traceability.json @@ -113,6 +113,10 @@ "US05-03": [ "tests/unit/test_immich_go_report.py", "tests/integration/test_upload_reports.py" + ], + "US05-04": [ + "tests/unit/test_immich_bulk_check.py", + "tests/integration/test_upload_verification.py" ] } } diff --git a/tests/unit/test_immich_bulk_check.py b/tests/unit/test_immich_bulk_check.py new file mode 100644 index 0000000..9008976 --- /dev/null +++ b/tests/unit/test_immich_bulk_check.py @@ -0,0 +1,27 @@ +"""How one bulk-upload-check result is read (US05-04). + +Only a duplicate rejection proves Immich holds the bytes. Every other answer — a +rejection for another reason, an action this adapter does not know — must stay +uncertain, because "the server did not say yes" is not "the file is not there". +""" + +from photo_pipeline.integrations.immich_go import _holds_bytes, bulk_upload_check + + +def test_a_duplicate_rejection_is_the_only_proof_of_possession(): + assert _holds_bytes({"action": "reject", "reason": "duplicate"}) is True + assert _holds_bytes({"action": "accept"}) is False + + +def test_any_other_answer_is_uncertain_never_absent(): + assert _holds_bytes({"action": "reject", "reason": "unsupported-format"}) is None + assert _holds_bytes({"action": "quarantine"}) is None + assert _holds_bytes({}) is None + + +def test_missing_credentials_are_reported_not_silently_treated_as_absence(): + result = bulk_upload_check("", None, {"asset": "abc"}) + + assert result["reachable"] is False + assert result["present"] == {} + assert "credentials" in result["detail"]