US05-06: Automate Phase E End-to-End Acceptance (#76)

tests/e2e/_pipeline_harness.py

@@ -9,14 +9,18 @@ and worker, never mocked inside a test.
 
 from __future__ import annotations
 
+import json
 import os
 import socket
+import stat
 import subprocess
 import sys
+import threading
 import time
 import uuid
 from dataclasses import dataclass
 from datetime import datetime, timezone
+from http.server import BaseHTTPRequestHandler, HTTPServer
 from pathlib import Path
 
 import httpx
@@ -258,3 +262,182 @@ def approve_album(base: str, *, album: str = "rome", name: str) -> None:
             json={**payload, "expected_version": current["version"]},
             timeout=10,
         ).raise_for_status()
+
+
+# ── Phase E: an upload-ready album, a fake Immich, and a real fake uploader ───
+
+SENTINEL_KEY = "immich-sentinel-9f3a2b"
+UPLOADER_VERSION = "immich-go 0.21.0"  # a pinned family, so reports are parsable
+
+# Uploader bodies for the pinned ``text-v1`` grammar. ``$6`` is the folder argument
+# of ``upload from-folder``.
+REPORTING_UPLOADER = (
+    'echo "INFO  uploaded $6/a.jpg"\n'
+    'echo "INFO  server has the same file $6/b.jpg"\n'
+    'echo "Uploaded 1, duplicates 1"\n'
+    "exit 0\n"
+)
+# Exits cleanly but says nothing about any file: the process succeeded, the
+# per-file outcome is unknown.
+SILENT_UPLOADER = "exit 0\n"
+
+
+def _immich_handler(state: dict):
+    class Handler(BaseHTTPRequestHandler):
+        def do_GET(self):  # noqa: N802 (BaseHTTPRequestHandler API)
+            self._json(200, {"res": "pong"})
+
+        def do_POST(self):  # noqa: N802
+            length = int(self.headers.get("Content-Length", 0))
+            payload = json.loads(self.rfile.read(length) or b"{}")
+            if state["mode"] == "broken":
+                self.send_error(500, "bulk-upload-check is unavailable")
+                return
+            reject = state["mode"] == "present"
+            self._json(
+                200,
+                {
+                    "results": [
+                        {
+                            "id": asset["id"],
+                            "action": "reject" if reject else "accept",
+                            "reason": "duplicate" if reject else None,
+                        }
+                        for asset in payload.get("assets", [])
+                    ]
+                },
+            )
+
+        def _json(self, code: int, body: dict) -> None:
+            raw = json.dumps(body).encode()
+            self.send_response(code)
+            self.send_header("Content-Type", "application/json")
+            self.send_header("Content-Length", str(len(raw)))
+            self.end_headers()
+            self.wfile.write(raw)
+
+        def log_message(self, *args):
+            pass
+
+    return Handler
+
+
+class FakeImmich:
+    """An Immich that answers ping, and says whether it holds the exact bytes.
+
+    ``mode`` is what the next verification will find: ``present`` (the server
+    deduplicates them, so it has them), ``absent`` (it would accept them, so it does
+    not), or ``broken`` (no usable answer at all).
+    """
+
+    def __init__(self) -> None:
+        self.state = {"mode": "present"}
+        self._server = HTTPServer(("127.0.0.1", 0), _immich_handler(self.state))
+        threading.Thread(target=self._server.serve_forever, daemon=True).start()
+        self.url = f"http://127.0.0.1:{self._server.server_port}"
+        self._running = True
+
+    def mode(self, mode: str) -> None:
+        self.state["mode"] = mode
+
+    def stop(self) -> None:
+        """Idempotent, so a test may take Immich away mid-journey."""
+        if not self._running:
+            return
+        self._running = False
+        self._server.shutdown()
+        self._server.server_close()
+
+
+def fake_uploader(tmp_path: Path, body: str) -> Path:
+    """A real executable standing in for immich-go.
+
+    ``--version`` answers like the real tool; any other invocation appends its
+    complete argv to ``immich-go.argv`` — which is how a test proves the uploader
+    ran, what folder it was handed, or that it never ran at all.
+    """
+    path = tmp_path / "immich-go"
+    path.write_text(
+        "#!/bin/sh\n"
+        f'if [ "$1" = "--version" ]; then echo "{UPLOADER_VERSION}"; exit 0; fi\n'
+        f'printf "%s\\n" "$*" >> "{tmp_path / "immich-go.argv"}"\n'
+        f"{body}"
+    )
+    path.chmod(path.stat().st_mode | stat.S_IEXEC | stat.S_IXGRP | stat.S_IXOTH)
+    return path
+
+
+def uploader_argv(tmp_path: Path) -> list[str]:
+    """Every upload invocation the fake uploader saw, oldest first."""
+    log = tmp_path / "immich-go.argv"
+    return log.read_text().splitlines() if log.exists() else []
+
+
+def mark_upload_ready(seeded: Seeded, *, unverified: tuple[str, ...] = ()) -> None:
+    """Give every seeded photo the verified EXIF checkpoints upload requires.
+
+    ``unverified`` names stems whose analysis checkpoint stays incomplete, which is
+    what makes an album partially blocked.
+    """
+    from sqlalchemy import select
+
+    from photo_pipeline.models import AnalysisResult, SafetyReview
+
+    blocked = {seeded.asset_ids[stem] for stem in unverified}
+    with session_factory(seeded) as sf:
+        with sf() as session:
+            for review in session.scalars(select(SafetyReview)):
+                review.exif_verified_at = NOW
+            for analysis in session.scalars(select(AnalysisResult)):
+                analysis.exif_written_at = None if analysis.asset_id in blocked else NOW
+            session.commit()
+
+
+class UploadStack:
+    """A seeded, upload-ready library plus the server, worker, and fake Immich."""
+
+    def __init__(self, tmp_path: Path, seeded: Seeded) -> None:
+        self.tmp_path = tmp_path
+        self.seeded = seeded
+        self.immich = FakeImmich()
+        self.server: Server | None = None
+        self.worker: subprocess.Popen | None = None
+        self.base = ""
+
+    def start(
+        self,
+        *,
+        uploader: str = REPORTING_UPLOADER,
+        worker: bool = True,
+        credentials: bool = True,
+    ) -> "UploadStack":
+        env = {
+            "PHOTO_PIPELINE_IMMICH_SERVER_URL": self.immich.url if credentials else "",
+            "PHOTO_PIPELINE_IMMICH_GO_BINARY": str(fake_uploader(self.tmp_path, uploader)),
+        }
+        if credentials:
+            env["PHOTO_PIPELINE_IMMICH_API_KEY"] = SENTINEL_KEY
+        self.server = Server(self.seeded, extra_env=env).start()
+        self.base = self.server.base
+        if worker:
+            self.worker = start_worker(self.seeded, extra_env=env)
+        return self
+
+    def restart_server(self) -> None:
+        """A genuinely fresh process against the same database and library."""
+        self.server.stop()
+        self.server.start()
+
+    def batches(self) -> list[dict]:
+        return httpx.get(f"{self.base}/api/v1/upload-batches", timeout=20).json()["batches"]
+
+    def argv(self) -> list[str]:
+        return uploader_argv(self.tmp_path)
+
+    def stop(self) -> None:
+        if self.worker is not None:
+            self.worker.kill()
+            self.worker.wait(timeout=10)
+        if self.server is not None:
+            self.server.stop()
+        self.immich.stop()