334 lines
13 KiB
Python
334 lines
13 KiB
Python
"""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")
|