diff --git a/migrations/versions/0009_upload_report_outcomes.py b/migrations/versions/0009_upload_report_outcomes.py new file mode 100644 index 0000000..4a35088 --- /dev/null +++ b/migrations/versions/0009_upload_report_outcomes.py @@ -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) diff --git a/photo_pipeline/integrations/immich_go_report.py b/photo_pipeline/integrations/immich_go_report.py new file mode 100644 index 0000000..3147f69 --- /dev/null +++ b/photo_pipeline/integrations/immich_go_report.py @@ -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)) diff --git a/photo_pipeline/models/uploads.py b/photo_pipeline/models/uploads.py index 821397c..7bafa7a 100644 --- a/photo_pipeline/models/uploads.py +++ b/photo_pipeline/models/uploads.py @@ -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() ) diff --git a/photo_pipeline/services/upload_batches.py b/photo_pipeline/services/upload_batches.py index 3d97ad4..6eaac46 100644 --- a/photo_pipeline/services/upload_batches.py +++ b/photo_pipeline/services/upload_batches.py @@ -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 ], diff --git a/photo_pipeline/services/upload_reports.py b/photo_pipeline/services/upload_reports.py new file mode 100644 index 0000000..6a69acf --- /dev/null +++ b/photo_pipeline/services/upload_reports.py @@ -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()) diff --git a/tests/integration/test_upload_reports.py b/tests/integration/test_upload_reports.py new file mode 100644 index 0000000..d8fec70 --- /dev/null +++ b/tests/integration/test_upload_reports.py @@ -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") diff --git a/tests/story_traceability.json b/tests/story_traceability.json index b0f8301..0998374 100644 --- a/tests/story_traceability.json +++ b/tests/story_traceability.json @@ -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" ] } } diff --git a/tests/unit/test_immich_go_report.py b/tests/unit/test_immich_go_report.py new file mode 100644 index 0000000..38a9ecc --- /dev/null +++ b/tests/unit/test_immich_go_report.py @@ -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"