US05-03: Parse and Persist Uploader Outcomes #73

Merged
domverse merged 1 commits from us/US05-03-parse-and-persist-uploader-outcomes into main 2026-08-16 15:02:34 +02:00
8 changed files with 916 additions and 7 deletions

View File

@@ -0,0 +1,52 @@
"""Parsed uploader outcomes (US05-03).
Revision ID: 0009_upload_report_outcomes
Revises: 0008_upload_batches
Create Date: 2026-08-16
Per-item outcomes parsed from the immich-go report plus the parser evidence that
produced them, so "what did Immich do with this file?" survives a restart.
"""
import sqlalchemy as sa
from alembic import op
revision = "0009_upload_report_outcomes"
down_revision = "0008_upload_batches"
branch_labels = None
depends_on = None
def upgrade() -> None:
# NULL parser = the uploader version has no pinned grammar; the batch is then
# requires_verification however the process exited.
op.add_column("upload_batches", sa.Column("parser", sa.String(), nullable=True))
op.add_column("upload_batches", sa.Column("parser_version", sa.Integer(), nullable=True))
op.add_column(
"upload_batches", sa.Column("parsed_at", sa.DateTime(timezone=True), nullable=True)
)
# verified | requires_verification
op.add_column("upload_batches", sa.Column("outcome_state", sa.String(), nullable=True))
op.add_column("upload_batches", sa.Column("outcome_counts", sa.String(), nullable=True))
op.add_column("upload_batches", sa.Column("report_counts", sa.String(), nullable=True))
# uploaded | upgraded | duplicate | skipped | failed | unknown
op.add_column("upload_items", sa.Column("outcome", sa.String(), nullable=True))
op.add_column("upload_items", sa.Column("evidence", sa.String(), nullable=True))
op.add_column(
"upload_items", sa.Column("outcome_at", sa.DateTime(timezone=True), nullable=True)
)
def downgrade() -> None:
for column in ("outcome_at", "evidence", "outcome"):
op.drop_column("upload_items", column)
for column in (
"report_counts",
"outcome_counts",
"outcome_state",
"parsed_at",
"parser_version",
"parser",
):
op.drop_column("upload_batches", column)

View File

@@ -0,0 +1,204 @@
"""immich-go report parsing (US05-03).
The uploader's output is the only local evidence of what Immich did with each file,
and its wording changes between releases (concept §15 "external integration
risks"). So parsing is deliberately conservative:
- **the parser is chosen by uploader version, not by guessing the format.** Each
supported version family maps to one adapter with a pinned line grammar. An
unrecognised version yields no adapter at all, which makes every item uncertain
rather than optimistically successful.
- **an unmatched line is never success.** Lines the adapter does not recognise are
counted (``unparsed``) and kept as evidence; they never classify an item.
- **evidence is bounded.** Each entry keeps a trimmed copy of the line that
classified it, and the number of entries is capped, so a looping uploader cannot
turn the report into unbounded database rows.
Two grammars are supported. ``text-v1`` is the default human-readable log of the
0.21/0.22 line, ``json-v1`` the structured log (``--log-type=json``) of 0.23/0.24.
Both classify through one shared phrase table, so the vocabulary cannot drift
between them:
```text
text-v1 INFO uploaded /lib/rome/a.jpg
INFO server has the same file /lib/rome/b.jpg
ERROR error uploading /lib/rome/e.jpg: connection reset
Uploaded 1, duplicates 1, errors 1
json-v1 {"level":"INFO","msg":"uploaded","file":"/lib/rome/a.jpg"}
{"level":"INFO","msg":"report","counts":{"uploaded":1}}
```
Nothing here touches the database; :mod:`photo_pipeline.services.upload_reports`
turns a parse result into durable per-item outcomes.
"""
from __future__ import annotations
import json
import re
PARSER_VERSION = 1
# Per-item outcomes. ``UNKNOWN`` is the safe default everywhere: it means the app
# does not know what happened to that file and US05-04 must verify it.
UPLOADED = "uploaded"
UPGRADED = "upgraded"
DUPLICATE = "duplicate"
SKIPPED = "skipped"
FAILED = "failed"
UNKNOWN = "unknown"
OUTCOMES = (UPLOADED, UPGRADED, DUPLICATE, SKIPPED, FAILED, UNKNOWN)
# Longest/most specific phrases first: "server has an older file" also contains
# "server has", and an error line about uploading also contains "upload".
_PHRASES: tuple[tuple[str, str], ...] = (
("server has the same file", DUPLICATE),
("server has an older file", UPGRADED),
("upgraded", UPGRADED),
("duplicate", DUPLICATE),
("discarded", SKIPPED),
("skipped", SKIPPED),
("error", FAILED),
("failed", FAILED),
("uploaded", UPLOADED),
("upload", UPLOADED),
)
# Summary line of the text grammar: "Uploaded 3, duplicates 1, errors 2".
_SUMMARY_WORDS = {
"uploaded": UPLOADED,
"upgraded": UPGRADED,
"duplicates": DUPLICATE,
"duplicate": DUPLICATE,
"skipped": SKIPPED,
"discarded": SKIPPED,
"errors": FAILED,
"error": FAILED,
}
_SUMMARY_PAIR = re.compile(r"([A-Za-z]+)\s+(\d+)")
# A path is anything that looks absolute, up to an explanatory ": reason" tail.
_PATH = re.compile(r"(/[^\s:][^:]*?)(?=:|$)")
MAX_EVIDENCE_CHARS = 300
# The report file is already byte-capped; this caps the rows it can produce.
MAX_ENTRIES = 20_000
def parser_for(uploader_version: str | None) -> str | None:
"""Adapter name for a recorded uploader version, or ``None`` when unsupported.
Unsupported is not an error — it is the honest answer that this build's output
format was never pinned, and it makes the whole report uncertain.
"""
match = re.search(r"(\d+)\.(\d+)", uploader_version or "")
if match is None:
return None
return _SUPPORTED.get(f"{match.group(1)}.{match.group(2)}")
def parse(report: str, uploader_version: str | None) -> dict:
"""Classify a raw report.
Returns ``{"parser", "parser_version", "supported", "entries", "counts",
"unparsed", "entries_truncated"}`` where ``entries`` is a list of
``{"path", "outcome", "evidence"}`` and ``counts`` is the uploader's own
summary when it printed one (``None`` otherwise, never invented).
"""
parser = parser_for(uploader_version)
result = {
"parser": parser,
"parser_version": PARSER_VERSION,
"supported": parser is not None,
"entries": [],
"counts": None,
"unparsed": 0,
"entries_truncated": False,
}
if parser is None:
return result
read_line = _text_line if parser == "text-v1" else _json_line
for raw in report.splitlines():
line = raw.strip()
if not line:
continue
path, outcome, counts = read_line(line)
if counts is not None:
# Later summaries win: the uploader prints its totals once, at the end.
result["counts"] = counts
continue
if path is None or outcome is None:
result["unparsed"] += 1
continue
if len(result["entries"]) >= MAX_ENTRIES:
result["entries_truncated"] = True
continue
result["entries"].append(
{"path": path, "outcome": outcome, "evidence": line[:MAX_EVIDENCE_CHARS]}
)
return result
def _classify(text: str) -> str | None:
lowered = text.lower()
for phrase, outcome in _PHRASES:
if phrase in lowered:
return outcome
return None
def _summary(text: str) -> dict[str, int] | None:
"""Totals from a summary line, or ``None`` when the line is not one."""
counts: dict[str, int] = {}
for word, number in _SUMMARY_PAIR.findall(text):
outcome = _SUMMARY_WORDS.get(word.lower())
if outcome is None:
return None # an unknown noun means this is not the summary grammar
counts[outcome] = counts.get(outcome, 0) + int(number)
return counts or None
def _text_line(line: str) -> tuple[str | None, str | None, dict | None]:
path_match = _PATH.search(line)
if path_match is None:
return None, None, _summary(line)
path = path_match.group(1).strip()
# Classify from the words around the path, never from the path itself: an
# album called "errors" must not turn an upload into a failure.
context = line.replace(path, " ")
return path, _classify(context), None
def _json_line(line: str) -> tuple[str | None, str | None, dict | None]:
try:
record = json.loads(line)
except ValueError:
return None, None, None
if not isinstance(record, dict):
return None, None, None
counts = record.get("counts")
if isinstance(counts, dict):
totals = {
_SUMMARY_WORDS[key.lower()]: int(value)
for key, value in counts.items()
if key.lower() in _SUMMARY_WORDS and isinstance(value, int)
}
return None, None, totals or None
path = record.get("file")
message = record.get("msg")
if not isinstance(path, str) or not isinstance(message, str):
return None, None, None
return path, _classify(message), None
# Version family → adapter. Pinning this is the point: a build outside the list is
# uncertain by construction (concept §15 "pin and record supported immich-go
# versions").
_SUPPORTED = {
"0.21": "text-v1",
"0.22": "text-v1",
"0.23": "json-v1",
"0.24": "json-v1",
}
SUPPORTED_VERSIONS = tuple(sorted(_SUPPORTED))

View File

@@ -57,6 +57,18 @@ class UploadBatch(Base):
report_bytes: Mapped[int | None] = mapped_column(Integer)
report_truncated: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
exit_code: Mapped[int | None] = mapped_column(Integer)
# Parsed report evidence (US05-03). ``parser`` is NULL when the uploader's
# version has no pinned grammar; ``outcome_state`` is then
# ``requires_verification`` regardless of how the process exited.
parser: Mapped[str | None] = mapped_column(String)
parser_version: Mapped[int | None] = mapped_column(Integer)
parsed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
# verified | requires_verification; NULL until a report has been parsed.
outcome_state: Mapped[str | None] = mapped_column(String)
outcome_counts: Mapped[str | None] = mapped_column(String) # JSON, from the items
report_counts: Mapped[str | None] = mapped_column(String) # JSON, uploader's own
error_code: Mapped[str | None] = mapped_column(String)
error_message: Mapped[str | None] = mapped_column(String)
@@ -81,10 +93,15 @@ class UploadItem(Base):
# Hashes of the bytes as they were when the batch was created.
sha256: Mapped[str | None] = mapped_column(String)
sha1: Mapped[str | None] = mapped_column(String)
# pending | sent | failed — the per-asset upload *result* is parsed from the
# report in US05-03 and verified in US05-04; ``sent`` only means the batch
# process exited successfully.
# pending | sent | failed — what the batch *process* did with this item;
# ``sent`` only means the uploader exited successfully.
state: Mapped[str] = mapped_column(String, nullable=False, default="pending")
# What the uploader's report says happened (US05-03): uploaded | upgraded |
# duplicate | skipped | failed | unknown. NULL before the report is parsed;
# ``unknown`` whenever the report does not classify this file — never success.
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))
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
)

View File

@@ -18,10 +18,11 @@ cannot undo, so the discipline is:
vanished ``unknown_requires_verification`` (concept §15) instead of retrying it
blindly. Resolving that is US05-04.
Per-asset upload *results* are deliberately not interpreted here — parsing the
report is US05-03. What this story guarantees is that the report exists, is bounded,
carries no credential, and that the batch's state is an honest description of what
the process did.
Per-asset upload *results* are not interpreted here: after the attempt ends the
report is handed to :class:`~photo_pipeline.services.upload_reports.
UploadReportService` (US05-03), which classifies every file. The batch ``state``
stays an honest description of what the *process* did; ``outcome_state`` says
whether the report's evidence is complete enough to trust.
"""
from __future__ import annotations
@@ -38,6 +39,7 @@ from photo_pipeline.config import Config
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.uploads import UploadService
@@ -238,6 +240,13 @@ class UploadBatchService:
self._finish(batch_id, token=token, state=state, error=error, result=result)
if item_state:
self._set_items(batch_id, item_state)
# The report is the only evidence of what happened per file, so it is read
# while it is fresh (US05-03). A parse failure must not lose the batch
# outcome that was just recorded; the items simply stay unknown.
try:
UploadReportService(self._session_factory).ingest(batch_id)
except OSError:
pass
return self.get(batch_id)
def cancel(self, batch_id: str) -> dict:
@@ -409,6 +418,14 @@ def _batch_dict(row: UploadBatch, items: list[UploadItem]) -> dict:
"exit_code": row.exit_code,
"error_code": row.error_code,
"error_message": row.error_message,
# Parsed report evidence (US05-03): what the uploader said per file, and
# whether that evidence is complete enough to be trusted.
"parser": row.parser,
"parser_version": row.parser_version,
"parsed_at": row.parsed_at.isoformat() if row.parsed_at else None,
"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,
"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": [
@@ -418,6 +435,9 @@ def _batch_dict(row: UploadBatch, items: list[UploadItem]) -> dict:
"sha256": item.sha256,
"sha1": item.sha1,
"state": item.state,
"outcome": item.outcome,
"evidence": item.evidence,
"outcome_at": item.outcome_at.isoformat() if item.outcome_at else None,
}
for item in items
],

View File

@@ -0,0 +1,140 @@
"""UploadReportService — durable per-item upload outcomes (US05-03).
US05-02 proves the uploader ran and kept a bounded, credential-free report. This
service is the step that reads that report and answers the operator's actual
question — *what happened to each photo?* — without ever guessing in the app's
favour:
- **an item is ``unknown`` until the report says otherwise.** A file the report
never mentions, a report from an unpinned uploader version, and a line the
grammar does not recognise all leave the item ``unknown``. Exit code 0 is
evidence the *process* ended well, never evidence that a particular file reached
Immich (concept §15 "wrong upload state").
- **counts are reconciled, not trusted.** The uploader's own summary is stored
next to the totals derived from the items. A disagreement makes the batch
``requires_verification`` even when every line parsed.
- **reprocessing is idempotent.** Outcomes are keyed by ``(batch_id, asset_id)``
and rewritten in place, so parsing the same report again — after a restart, or
because the operator asked — converges on the same rows and the same counts.
The batch's own ``state`` still describes the process (US05-02's contract);
``outcome_state`` describes the evidence. Turning ``requires_verification`` into
verification, retry, and manual resolution is US05-04.
"""
from __future__ import annotations
import json
import os
from datetime import datetime, timezone
from pathlib import Path
from sqlalchemy import select
from sqlalchemy.orm import sessionmaker
from photo_pipeline.integrations import immich_go_report as report_parser
from photo_pipeline.models import UploadBatch, UploadItem
VERIFIED = "verified"
REQUIRES_VERIFICATION = "requires_verification"
def _now() -> datetime:
return datetime.now(timezone.utc)
def _key(path: str) -> str:
return os.path.normpath(path)
class UploadReportService:
def __init__(self, session_factory: sessionmaker) -> None:
self._session_factory = session_factory
def ingest(self, batch_id: str) -> dict:
"""Parse this batch's report and persist an outcome for every item.
Returns ``{"parser", "parser_version", "outcome_state", "counts",
"report_counts", "unmatched", "unparsed"}``. Safe to call repeatedly.
"""
with self._session_factory() as session:
batch = session.get(UploadBatch, batch_id)
if batch is None:
raise KeyError(f"unknown upload batch {batch_id!r}")
items = list(session.scalars(select(UploadItem).where(UploadItem.batch_id == batch_id)))
parsed = report_parser.parse(_read_report(batch.report_path), batch.uploader_version)
by_path = {_key(entry["path"]): entry for entry in parsed["entries"]}
# Basenames are the fallback: an uploader may log a path relative to the
# folder it was given. Ambiguous basenames are dropped rather than
# guessed at.
by_name: dict[str, dict] = {}
for key, entry in by_path.items():
name = os.path.basename(key)
by_name[name] = None if name in by_name else entry
matched: set[str] = set()
counts = {outcome: 0 for outcome in report_parser.OUTCOMES}
for item in items:
key = _key(item.path)
entry = by_path.get(key) or by_name.get(os.path.basename(key))
if entry is not None:
matched.add(_key(entry["path"]))
item.outcome = entry["outcome"] if entry else report_parser.UNKNOWN
item.evidence = entry["evidence"] if entry else None
item.outcome_at = _now()
counts[item.outcome] += 1
unmatched = sorted(set(by_path) - matched)
reconciled = _reconciles(parsed["counts"], counts)
state = (
VERIFIED
if (
parsed["supported"]
and items
and not counts[report_parser.UNKNOWN]
and not unmatched
and not parsed["entries_truncated"]
and reconciled
)
else REQUIRES_VERIFICATION
)
batch.parser = parsed["parser"]
batch.parser_version = parsed["parser_version"]
batch.parsed_at = _now()
batch.outcome_state = state
batch.outcome_counts = json.dumps(counts, sort_keys=True)
batch.report_counts = (
json.dumps(parsed["counts"], sort_keys=True) if parsed["counts"] else None
)
session.commit()
return {
"parser": parsed["parser"],
"parser_version": parsed["parser_version"],
"outcome_state": state,
"counts": counts,
"report_counts": parsed["counts"],
"unmatched": unmatched,
"unparsed": parsed["unparsed"],
}
def _read_report(report_path: str | None) -> str:
"""The raw report, or ``""`` when the attempt never produced one."""
if not report_path:
return ""
path = Path(report_path)
return path.read_text(errors="replace") if path.exists() else ""
def _reconciles(report_counts: dict | None, derived: dict) -> bool:
"""Whether the uploader's own totals agree with the per-item outcomes.
No summary is not a disagreement — most of the uncertainty it would catch is
already caught by unmatched entries and unknown items.
"""
if not report_counts:
return True
return all(derived.get(outcome) == total for outcome, total in report_counts.items())

View File

@@ -0,0 +1,333 @@
"""Durable per-item upload outcomes (US05-03).
The uploader is a real executable that prints a real report, driven through the
real batch service, so the classification path exercised here is the one
production uses. What matters is that the stored outcome is never more optimistic
than the evidence: an unmentioned file, an unpinned uploader version, or counts
that disagree all leave the batch requiring verification.
"""
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.integrations import immich_go_report as report_parser
from photo_pipeline.models import AnalysisResult, Asset, SafetyReview
from photo_pipeline.services.hashing import sha256_file
from photo_pipeline.services.upload_batches import BatchState, UploadBatchService
from photo_pipeline.services.upload_reports import (
REQUIRES_VERIFICATION,
VERIFIED,
UploadReportService,
)
from photo_pipeline.services.uploads import UploadService
NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
SUPPORTED_VERSION = "immich-go 0.21.0"
# ── fake external boundary ───────────────────────────────────────────────────
class _PingHandler(BaseHTTPRequestHandler):
def do_GET(self): # noqa: N802 (BaseHTTPRequestHandler API)
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(b'{"res":"pong"}')
def log_message(self, *args):
pass
@pytest.fixture
def immich_server():
server = HTTPServer(("127.0.0.1", 0), _PingHandler)
threading.Thread(target=server.serve_forever, daemon=True).start()
yield f"http://127.0.0.1:{server.server_port}"
server.shutdown()
server.server_close()
def _uploader(tmp_path, report_body: str, *, version=SUPPORTED_VERSION):
"""An uploader that answers ``--version`` and prints ``report_body``."""
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"
"exit 0\n"
)
path.chmod(path.stat().st_mode | stat.S_IEXEC | stat.S_IXGRP | stat.S_IXOTH)
return path
# ── environment ──────────────────────────────────────────────────────────────
def _env(tmp_path, server_url, uploader):
(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": server_url,
"PHOTO_PIPELINE_IMMICH_API_KEY": "immich-sentinel-9f3a2b",
"PHOTO_PIPELINE_IMMICH_GO_BINARY": str(uploader),
}
)
run_migrations(config.database_url)
return config, create_session_factory(create_db_engine(config.database_url)), lib
def _album(sf, lib, album="rome", names=("new.jpg", "dup.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 _run(sf, config, albums=None):
"""Create a batch from a fresh preflight and run it to completion."""
report = UploadService(sf, config=config).preflight(albums)
assert report["state"] == "ready", report["blockers"]
service = UploadBatchService(sf, config=config)
(batch,) = service.create(albums, token=report["token"])
return service.run(batch["id"])
def _outcomes(batch) -> dict:
return {Path(item["path"]).name: item["outcome"] for item in batch["items"]}
def _report_for(folder, **outcomes) -> str:
"""A text-v1 report plus a matching summary line."""
lines = {
"uploaded": "INFO uploaded {path}",
"duplicate": "INFO server has the same file {path}",
"upgraded": "INFO server has an older file, upgrading {path}",
"failed": "ERROR error uploading {path}: connection reset",
}
body = [lines[outcome].format(path=folder / name) for name, outcome in outcomes.items()]
totals: dict[str, int] = {}
for outcome in outcomes.values():
totals[outcome] = totals.get(outcome, 0) + 1
words = {
"uploaded": "Uploaded",
"duplicate": "duplicates",
"upgraded": "upgraded",
"failed": "errors",
}
body.append(", ".join(f"{words[k]} {v}" for k, v in totals.items()))
return "\n".join(body)
# ── classification ───────────────────────────────────────────────────────────
def test_every_file_gets_its_reported_outcome_with_evidence(tmp_path, immich_server):
lib = tmp_path / "lib"
report = _report_for(lib / "rome", **{"new.jpg": "uploaded", "dup.jpg": "duplicate"})
config, sf, lib = _env(tmp_path, immich_server, _uploader(tmp_path, report))
_album(sf, lib)
batch = _run(sf, config)
assert batch["state"] == BatchState.SUCCEEDED
assert batch["outcome_state"] == VERIFIED
assert _outcomes(batch) == {"new.jpg": "uploaded", "dup.jpg": "duplicate"}
assert batch["outcome_counts"]["uploaded"] == 1
assert batch["outcome_counts"]["duplicate"] == 1
assert batch["parser"] == "text-v1"
assert batch["parser_version"] == report_parser.PARSER_VERSION
assert batch["parsed_at"]
for item in batch["items"]:
assert item["sha1"] and item["sha256"], "the uploaded bytes stay identifiable"
assert Path(item["path"]).name in item["evidence"]
assert item["outcome_at"]
def test_a_file_the_report_never_mentions_is_unknown_not_sent(tmp_path, immich_server):
"""Exit code 0 says the process ended well, not that this photo reached Immich."""
lib = tmp_path / "lib"
report = _report_for(lib / "rome", **{"new.jpg": "uploaded"})
config, sf, lib = _env(tmp_path, immich_server, _uploader(tmp_path, report))
_album(sf, lib)
batch = _run(sf, config)
assert batch["state"] == BatchState.SUCCEEDED, "the process itself succeeded"
assert _outcomes(batch)["dup.jpg"] == report_parser.UNKNOWN
assert batch["outcome_state"] == REQUIRES_VERIFICATION
def test_an_unsupported_uploader_version_makes_every_item_unknown(tmp_path, immich_server):
lib = tmp_path / "lib"
report = _report_for(lib / "rome", **{"new.jpg": "uploaded", "dup.jpg": "uploaded"})
uploader = _uploader(tmp_path, report, version="immich-go 9.99.0")
config, sf, lib = _env(tmp_path, immich_server, uploader)
_album(sf, lib)
batch = _run(sf, config)
assert batch["parser"] is None
assert set(_outcomes(batch).values()) == {report_parser.UNKNOWN}
assert batch["outcome_state"] == REQUIRES_VERIFICATION
def test_an_empty_report_leaves_everything_unknown(tmp_path, immich_server):
config, sf, lib = _env(tmp_path, immich_server, _uploader(tmp_path, ""))
_album(sf, lib)
batch = _run(sf, config)
assert set(_outcomes(batch).values()) == {report_parser.UNKNOWN}
assert batch["outcome_state"] == REQUIRES_VERIFICATION
def test_a_failed_upload_is_classified_from_its_own_line(tmp_path, immich_server):
lib = tmp_path / "lib"
report = _report_for(lib / "rome", **{"new.jpg": "uploaded", "dup.jpg": "failed"})
config, sf, lib = _env(tmp_path, immich_server, _uploader(tmp_path, report))
_album(sf, lib)
batch = _run(sf, config)
assert _outcomes(batch) == {"new.jpg": "uploaded", "dup.jpg": "failed"}
# Every file is accounted for, so the evidence is complete even though one
# upload failed — resolving the failure is US05-04's job, not a re-parse.
assert batch["outcome_state"] == VERIFIED
# ── reconciliation ───────────────────────────────────────────────────────────
def test_a_summary_that_disagrees_with_the_lines_requires_verification(tmp_path, immich_server):
lib = tmp_path / "lib"
report = (
_report_for(lib / "rome", **{"new.jpg": "uploaded", "dup.jpg": "uploaded"}).rsplit("\n", 1)[
0
]
+ "\nUploaded 5"
)
config, sf, lib = _env(tmp_path, immich_server, _uploader(tmp_path, report))
_album(sf, lib)
batch = _run(sf, config)
assert batch["report_counts"] == {"uploaded": 5}
assert batch["outcome_counts"]["uploaded"] == 2
assert batch["outcome_state"] == REQUIRES_VERIFICATION
def test_a_reported_file_outside_the_batch_requires_verification(tmp_path, immich_server):
"""The uploader touched something the batch never approved."""
lib = tmp_path / "lib"
report = _report_for(
lib / "rome", **{"new.jpg": "uploaded", "dup.jpg": "uploaded", "stranger.jpg": "uploaded"}
)
config, sf, lib = _env(tmp_path, immich_server, _uploader(tmp_path, report))
_album(sf, lib)
batch = _run(sf, config)
assert set(_outcomes(batch).values()) == {"uploaded"}
assert batch["outcome_state"] == REQUIRES_VERIFICATION
# ── idempotency and restart ──────────────────────────────────────────────────
def test_reprocessing_the_same_report_is_idempotent(tmp_path, immich_server):
lib = tmp_path / "lib"
report = _report_for(lib / "rome", **{"new.jpg": "uploaded", "dup.jpg": "duplicate"})
config, sf, lib = _env(tmp_path, immich_server, _uploader(tmp_path, report))
_album(sf, lib)
batch = _run(sf, config)
first = UploadReportService(sf).ingest(batch["id"])
second = UploadReportService(sf).ingest(batch["id"])
assert first == second
reread = UploadBatchService(sf, config=config).get(batch["id"])
assert _outcomes(reread) == _outcomes(batch)
assert len(reread["items"]) == 2, "re-import must not duplicate item rows"
assert reread["outcome_counts"] == batch["outcome_counts"]
def test_outcomes_survive_a_restart_and_reach_the_api(tmp_path, immich_server):
lib = tmp_path / "lib"
report = _report_for(lib / "rome", **{"new.jpg": "uploaded", "dup.jpg": "duplicate"})
config, sf, lib = _env(tmp_path, immich_server, _uploader(tmp_path, report))
_album(sf, lib)
batch = _run(sf, config)
with TestClient(create_app(config)) as client: # a fresh application process
fetched = client.get(f"/api/v1/upload-batches/{batch['id']}").json()
assert fetched["outcome_state"] == VERIFIED
assert {Path(i["path"]).name: i["outcome"] for i in fetched["items"]} == {
"new.jpg": "uploaded",
"dup.jpg": "duplicate",
}
assert json.dumps(fetched) # the record stays JSON-serialisable for the UI
def test_a_missing_report_file_does_not_lose_the_batch(tmp_path, immich_server):
lib = tmp_path / "lib"
report = _report_for(lib / "rome", **{"new.jpg": "uploaded", "dup.jpg": "duplicate"})
config, sf, lib = _env(tmp_path, immich_server, _uploader(tmp_path, report))
_album(sf, lib)
batch = _run(sf, config)
Path(batch["report_path"]).unlink()
result = UploadReportService(sf).ingest(batch["id"])
assert result["outcome_state"] == REQUIRES_VERIFICATION
assert set(_outcomes(UploadBatchService(sf, config=config).get(batch["id"])).values()) == {
report_parser.UNKNOWN
}
def test_ingesting_an_unknown_batch_is_an_error(tmp_path, immich_server):
config, sf, lib = _env(tmp_path, immich_server, _uploader(tmp_path, ""))
with pytest.raises(KeyError):
UploadReportService(sf).ingest("does-not-exist")

View File

@@ -109,6 +109,10 @@
],
"US05-02": [
"tests/integration/test_upload_batches.py"
],
"US05-03": [
"tests/unit/test_immich_go_report.py",
"tests/integration/test_upload_reports.py"
]
}
}

View File

@@ -0,0 +1,139 @@
"""Golden parser fixtures for immich-go reports (US05-03).
The reports below are the pinned grammar for each supported uploader version. If a
future immich-go changes its wording, the fix is a new adapter and a new golden
report — never a looser pattern here, because a loose pattern is how an unread line
becomes a false success.
"""
import pytest
from photo_pipeline.integrations import immich_go_report as parser
TEXT_REPORT = """
Scanning /lib/rome
INFO uploaded /lib/rome/new.jpg
INFO server has the same file /lib/rome/dup.jpg
INFO server has an older file, upgrading /lib/rome/old.jpg
WARN discarded /lib/rome/notes.txt: unsupported file type
ERROR error uploading /lib/rome/broken.jpg: connection reset by peer
Uploaded 1, upgraded 1, duplicates 1, skipped 1, errors 1
"""
JSON_REPORT = """
{"time":"2026-01-01T10:00:00Z","level":"INFO","msg":"uploaded","file":"/lib/rome/new.jpg"}
{"level":"INFO","msg":"server has the same file","file":"/lib/rome/dup.jpg"}
{"level":"INFO","msg":"server has an older file, upgrading","file":"/lib/rome/old.jpg"}
{"level":"WARN","msg":"discarded: unsupported file type","file":"/lib/rome/notes.txt"}
{"level":"ERROR","msg":"error uploading","file":"/lib/rome/broken.jpg"}
{"level":"INFO","msg":"report","counts":{"uploaded":1,"upgraded":1,"duplicates":1,\
"skipped":1,"errors":1}}
"""
EXPECTED = {
"/lib/rome/new.jpg": parser.UPLOADED,
"/lib/rome/dup.jpg": parser.DUPLICATE,
"/lib/rome/old.jpg": parser.UPGRADED,
"/lib/rome/notes.txt": parser.SKIPPED,
"/lib/rome/broken.jpg": parser.FAILED,
}
EXPECTED_COUNTS = {
parser.UPLOADED: 1,
parser.UPGRADED: 1,
parser.DUPLICATE: 1,
parser.SKIPPED: 1,
parser.FAILED: 1,
}
def _outcomes(parsed) -> dict:
return {entry["path"]: entry["outcome"] for entry in parsed["entries"]}
@pytest.mark.parametrize(
("version", "report", "expected_parser"),
[
("immich-go 0.21.0", TEXT_REPORT, "text-v1"),
("immich-go 0.22.3", TEXT_REPORT, "text-v1"),
("immich-go 0.23.1", JSON_REPORT, "json-v1"),
("immich-go 0.24.0", JSON_REPORT, "json-v1"),
],
)
def test_every_supported_version_classifies_every_outcome(version, report, expected_parser):
parsed = parser.parse(report, version)
assert parsed["parser"] == expected_parser and parsed["supported"]
assert parsed["parser_version"] == parser.PARSER_VERSION
assert _outcomes(parsed) == EXPECTED
assert parsed["counts"] == EXPECTED_COUNTS
def test_an_unsupported_version_parses_nothing_at_all():
"""A build whose output was never pinned must not be read optimistically."""
parsed = parser.parse(TEXT_REPORT, "immich-go 9.99.0")
assert parsed["parser"] is None and parsed["supported"] is False
assert parsed["entries"] == [] and parsed["counts"] is None
@pytest.mark.parametrize("version", [None, "", "immich-go dev", "unknown"])
def test_a_missing_or_unreadable_version_is_unsupported(version):
assert parser.parser_for(version) is None
def test_malformed_lines_are_counted_never_classified():
report = (
"{not json at all\n"
'{"level":"INFO","msg":"uploaded"}\n' # no file
'["not","an","object"]\n'
'{"level":"INFO","msg":"uploaded","file":"/lib/rome/new.jpg"}\n'
)
parsed = parser.parse(report, "immich-go 0.23.1")
assert _outcomes(parsed) == {"/lib/rome/new.jpg": parser.UPLOADED}
assert parsed["unparsed"] == 3
def test_a_line_with_a_path_but_no_known_verb_is_not_an_outcome():
parsed = parser.parse("INFO considering /lib/rome/a.jpg\n", "immich-go 0.21.0")
assert parsed["entries"] == [] and parsed["unparsed"] == 1
def test_the_album_name_never_classifies_the_line():
"""A folder called "errors" must not turn a successful upload into a failure."""
parsed = parser.parse("INFO uploaded /lib/errors/a.jpg\n", "immich-go 0.21.0")
assert _outcomes(parsed) == {"/lib/errors/a.jpg": parser.UPLOADED}
def test_paths_with_spaces_and_reasons_are_read_whole():
report = "ERROR error uploading /lib/summer holiday/a b.jpg: connection reset\n"
parsed = parser.parse(report, "immich-go 0.21.0")
assert _outcomes(parsed) == {"/lib/summer holiday/a b.jpg": parser.FAILED}
def test_evidence_is_bounded_per_entry():
line = "INFO uploaded /lib/rome/a.jpg " + "x" * 5_000
(entry,) = parser.parse(line, "immich-go 0.21.0")["entries"]
assert len(entry["evidence"]) == parser.MAX_EVIDENCE_CHARS
def test_a_looping_uploader_cannot_produce_unbounded_entries(monkeypatch):
monkeypatch.setattr(parser, "MAX_ENTRIES", 10)
report = "".join(f"INFO uploaded /lib/rome/{n}.jpg\n" for n in range(50))
parsed = parser.parse(report, "immich-go 0.21.0")
assert len(parsed["entries"]) == 10 and parsed["entries_truncated"] is True
def test_a_report_without_a_summary_reports_no_counts():
parsed = parser.parse("INFO uploaded /lib/rome/a.jpg\n", "immich-go 0.21.0")
assert parsed["counts"] is None, "counts are read, never invented"