US05-03: Parse and Persist Uploader Outcomes #73
52
migrations/versions/0009_upload_report_outcomes.py
Normal file
52
migrations/versions/0009_upload_report_outcomes.py
Normal 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)
|
||||
204
photo_pipeline/integrations/immich_go_report.py
Normal file
204
photo_pipeline/integrations/immich_go_report.py
Normal 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))
|
||||
@@ -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()
|
||||
)
|
||||
|
||||
@@ -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
|
||||
],
|
||||
|
||||
140
photo_pipeline/services/upload_reports.py
Normal file
140
photo_pipeline/services/upload_reports.py
Normal 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())
|
||||
333
tests/integration/test_upload_reports.py
Normal file
333
tests/integration/test_upload_reports.py
Normal 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")
|
||||
@@ -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"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
139
tests/unit/test_immich_go_report.py
Normal file
139
tests/unit/test_immich_go_report.py
Normal 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"
|
||||
Reference in New Issue
Block a user