US02-07: Automate Phase B End-to-End Acceptance (#60)

tests/e2e/_pipeline_harness.py

@@ -0,0 +1,161 @@
+"""Reusable Phase B end-to-end harness: seed an isolated library, then launch the
+real API server and durable worker as child processes so tests talk to them only
+over HTTP/SSE — the same boundaries used in production (concept §18).
+
+External vision is replaced at its integration edge by the in-repo recording fake
+(``PHOTO_PIPELINE_FAKE_VISION_LOG``); it is invoked through the real AnalysisService
+and worker, never mocked inside a test.
+"""
+
+from __future__ import annotations
+
+import os
+import socket
+import subprocess
+import sys
+import time
+from dataclasses import dataclass
+from pathlib import Path
+
+import httpx
+import numpy as np
+from PIL import Image
+
+REPO = Path(__file__).resolve().parents[2]
+
+
+def image(path: Path, seed: int) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    rng = np.random.default_rng(seed)
+    Image.fromarray(rng.integers(0, 256, (96, 128, 3), dtype=np.uint8)).save(path, quality=90)
+
+
+def free_port() -> int:
+    with socket.socket() as sock:
+        sock.bind(("127.0.0.1", 0))
+        return sock.getsockname()[1]
+
+
+@dataclass
+class Seeded:
+    data: Path
+    lib: Path
+    asset_ids: dict[str, str]  # filename stem -> asset id
+
+
+def seed_library(tmp_path: Path, files: dict[str, int], decisions: dict[str, str]) -> Seeded:
+    """Scan ``files`` (stem -> seed) into a fresh DB and apply safety ``decisions``
+    (stem -> ``sfw``/``nsfw``) in-process before any server starts, so the seed is
+    deterministic and independent of the API under test."""
+    data = tmp_path / "data"
+    data.mkdir()
+    lib = tmp_path / "lib"
+    lib.mkdir()
+    for stem, seed in files.items():
+        image(lib / f"{stem}.jpg", seed)
+
+    from photo_pipeline.config import Config
+    from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
+    from photo_pipeline.models import Asset
+    from photo_pipeline.services.inventory import InventoryService
+    from photo_pipeline.services.safety import SafetyService
+    from sqlalchemy import select
+
+    config = Config.from_env(
+        {"PHOTO_PIPELINE_DATA_DIR": str(data), "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib)}
+    )
+    run_migrations(config.database_url)
+    engine = create_db_engine(config.database_url)
+    sf = create_session_factory(engine)
+    InventoryService(sf).scan(lib)
+    with sf() as session:
+        rows = list(session.execute(select(Asset.id, Asset.current_path)).all())
+    ids = {Path(path).stem: aid for aid, path in rows}
+    safety = SafetyService(sf)
+    for stem, decision in decisions.items():
+        safety.decide(ids[stem], decision, write_exif=False)
+    engine.dispose()
+    return Seeded(data=data, lib=lib, asset_ids=ids)
+
+
+def _env(seeded: Seeded, port: int, *, extra: dict[str, str] | None = None) -> dict[str, str]:
+    env = {
+        "PATH": os.environ.get("PATH", ""),
+        "PHOTO_PIPELINE_DATA_DIR": str(seeded.data),
+        "PHOTO_PIPELINE_LIBRARY_ROOTS": str(seeded.lib),
+        "PHOTO_PIPELINE_HOST": "127.0.0.1",
+        "PHOTO_PIPELINE_PORT": str(port),
+    }
+    if extra:
+        env.update(extra)
+    return env
+
+
+class Server:
+    """A real ``photo_pipeline serve`` child process, ready when /health/ready is 200."""
+
+    def __init__(self, seeded: Seeded, *, extra_env: dict[str, str] | None = None) -> None:
+        self._seeded = seeded
+        self._extra = extra_env
+        self.port = free_port()
+        self.base = f"http://127.0.0.1:{self.port}"
+        self.proc: subprocess.Popen | None = None
+
+    def start(self) -> "Server":
+        self.proc = subprocess.Popen(
+            [sys.executable, "-m", "photo_pipeline", "serve"],
+            cwd=str(REPO),
+            env=_env(self._seeded, self.port, extra=self._extra),
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+        )
+        deadline = time.monotonic() + 30
+        while time.monotonic() < deadline:
+            if self.proc.poll() is not None:
+                _, err = self.proc.communicate()
+                raise RuntimeError(f"server exited: {err.decode(errors='replace')}")
+            try:
+                if httpx.get(f"{self.base}/api/v1/health/ready", timeout=1).status_code == 200:
+                    return self
+            except httpx.HTTPError:
+                time.sleep(0.2)
+        self.stop()
+        raise RuntimeError("server never became ready")
+
+    def stop(self) -> None:
+        if self.proc is None:
+            return
+        self.proc.terminate()
+        try:
+            self.proc.wait(timeout=10)
+        except subprocess.TimeoutExpired:
+            self.proc.kill()
+        self.proc = None
+
+
+def start_worker(seeded: Seeded, *, fake_vision_log: Path) -> subprocess.Popen:
+    """Launch a real durable worker wired to the recording vision fake."""
+    return subprocess.Popen(
+        [sys.executable, "-m", "photo_pipeline", "worker", "--id", "e2e-worker"],
+        cwd=str(REPO),
+        env=_env(
+            seeded,
+            free_port(),  # unused by the worker, but keeps the env shape uniform
+            extra={"PHOTO_PIPELINE_FAKE_VISION_LOG": str(fake_vision_log)},
+        ),
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+    )
+
+
+def wait_until(predicate, *, timeout: float = 20, interval: float = 0.1):
+    """Poll ``predicate`` until it returns a truthy value or the timeout elapses.
+    Returns the truthy value; raises on timeout. Synchronizes on observable state
+    instead of sleeping for a fixed duration (concept §18)."""
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        value = predicate()
+        if value:
+            return value
+        time.sleep(interval)
+    raise AssertionError("condition not met before timeout")

tests/e2e/test_analysis_browser.py

@@ -0,0 +1,58 @@
+"""Phase B browser journey (US02-07): start a content-analysis job from the Analyze
+view and watch live progress arrive over the real SSE adapter, against a real server
+and durable worker. The frontend's SSE→polling event adapter and the worker are
+exercised end to end; the vision edge is the recording fake.
+"""
+
+from __future__ import annotations
+
+import pytest
+from playwright.sync_api import expect
+
+from tests.e2e._pipeline_harness import Server, seed_library, start_worker
+
+pytestmark = pytest.mark.phase_b
+
+
+@pytest.fixture
+def running_stack(tmp_path):
+    seeded = seed_library(tmp_path, {"beach": 1}, {"beach": "sfw"})
+    log = tmp_path / "vision.log"
+    server = Server(seeded).start()
+    worker = start_worker(seeded, fake_vision_log=log)
+    try:
+        yield server
+    finally:
+        worker.terminate()
+        server.stop()
+
+
+def test_analyze_start_shows_live_progress_and_completes(page, running_stack):
+    errors = []
+    page.on("console", lambda m: errors.append(m.text) if m.type == "error" else None)
+
+    page.goto(f"{running_stack.base}/app/#/analyze")
+    run = page.get_by_test_id("run-analysis")
+    run.wait_for()
+    assert run.is_enabled()  # one confirmed-SFW asset is eligible
+
+    run.click()
+
+    # Live progress streams into the activity log via the browser's SSE adapter.
+    log = page.get_by_test_id("analyze-log")
+    log.get_by_text("Started analysis job").wait_for()
+
+    # On completion the view re-renders with the updated counts: the SFW asset is
+    # now analysed. Waiting on this observable state (not a sleep) proves the whole
+    # start → worker → SSE → done round trip worked.
+    analyzed_value = (
+        page.get_by_test_id("analyze-counts")
+        .locator(".stat")
+        .filter(has_text="Analyzed")
+        .locator(".stat-value")
+    )
+    # ``expect`` re-queries the locator each poll, so it survives the re-render that
+    # replaces the counts DOM when the job completes.
+    expect(analyzed_value).to_have_text("1", timeout=15000)
+
+    assert errors == [], f"console errors: {errors}"

tests/e2e/test_phase_b_pipeline.py

@@ -0,0 +1,254 @@
+"""Phase B black-box end-to-end acceptance (US02-07).
+
+Launches the real FastAPI server and durable worker as child processes and drives
+them only over HTTP/SSE — no service/repository imports for the behaviour under
+test. Covers the Phase B journeys the browser suite does not exhaustively assert:
+analysis start/progress, resumable SSE reconnect, the polling fallback,
+cancellation, per-asset error inspection, one-mutating-job rejection during
+read-only browsing, the NSFW→vision privacy gate, and durability across restart.
+
+External vision is the in-repo recording fake, invoked through the real worker.
+"""
+
+from __future__ import annotations
+
+import json
+
+import httpx
+import pytest
+
+from tests.e2e._pipeline_harness import Server, seed_library, start_worker, wait_until
+
+pytestmark = pytest.mark.phase_b
+
+TIMEOUT = 10
+
+
+def _job_state(base: str, job_id: str) -> str:
+    return httpx.get(f"{base}/api/v1/jobs/{job_id}", timeout=TIMEOUT).json()["state"]
+
+
+def _sse_events(
+    base: str,
+    job_id: str,
+    *,
+    after: int = 0,
+    last_event_id: int | None = None,
+    stop_after: int | None = None,
+) -> list[dict]:
+    """Read the resumable SSE stream, returning ``[{seq, type, message}, ...]``. When
+    ``stop_after`` is set, close the stream after that many events to model a client
+    that drops mid-stream. ``last_event_id`` sends the reconnect header."""
+    headers = {}
+    if last_event_id is not None:
+        headers["Last-Event-ID"] = str(last_event_id)
+    url = f"{base}/api/v1/jobs/{job_id}/events/stream?after={after}"
+    events: list[dict] = []
+    seq: int | None = None
+    with httpx.stream("GET", url, headers=headers, timeout=TIMEOUT) as response:
+        response.raise_for_status()
+        for line in response.iter_lines():
+            if line.startswith("id:"):
+                seq = int(line[3:].strip())
+            elif line.startswith("data:"):
+                payload = json.loads(line[5:].strip())
+                if seq is not None:  # skip the terminal `event: done` (empty {} data)
+                    events.append({"seq": seq, **payload})
+                    seq = None
+                    if stop_after is not None and len(events) >= stop_after:
+                        return events
+            elif line.startswith("event: done"):
+                break
+    return events
+
+
+# ── analysis job: start, progress, and completion ────────────────────────────
+
+
+def test_analysis_job_runs_to_completion_over_real_http(tmp_path):
+    seeded = seed_library(tmp_path, {"beach": 1}, {"beach": "sfw"})
+    log = tmp_path / "vision.log"
+    server = Server(seeded).start()
+    worker = start_worker(seeded, fake_vision_log=log)
+    try:
+        job = httpx.post(f"{server.base}/api/v1/analysis/jobs", timeout=TIMEOUT).json()
+        wait_until(lambda: _job_state(server.base, job["id"]) == "succeeded", timeout=TIMEOUT)
+
+        counts = httpx.get(f"{server.base}/api/v1/analysis/counts", timeout=TIMEOUT).json()
+        assert counts["analyzed"] == 1
+        # The fake actually ran through the real worker for the one SFW asset.
+        assert log.read_text().strip().endswith("beach.jpg")
+    finally:
+        worker.terminate()
+        server.stop()
+
+
+# ── resumable SSE reconnect and the polling fallback ─────────────────────────
+
+
+def test_sse_reconnect_resumes_without_gap_or_duplicate(tmp_path):
+    seeded = seed_library(tmp_path, {"beach": 1, "meadow": 2}, {"beach": "sfw", "meadow": "sfw"})
+    log = tmp_path / "vision.log"
+    server = Server(seeded).start()
+    worker = start_worker(seeded, fake_vision_log=log)
+    try:
+        job = httpx.post(f"{server.base}/api/v1/analysis/jobs", timeout=TIMEOUT).json()
+        wait_until(lambda: _job_state(server.base, job["id"]) == "succeeded", timeout=TIMEOUT)
+
+        full = _sse_events(server.base, job["id"])
+        assert [e["type"] for e in full][:2] == ["queued", "claimed"]
+        assert full[-1]["type"] == "state:succeeded"
+        assert len(full) >= 3
+
+        # A client that reads only the first event, drops, then reconnects with the
+        # last id it saw must receive exactly the remaining events — no gap, no dup.
+        first = _sse_events(server.base, job["id"], stop_after=1)
+        assert len(first) == 1
+        rest = _sse_events(server.base, job["id"], last_event_id=first[0]["seq"])
+        assert first + rest == full
+        seqs = [e["seq"] for e in full]
+        assert seqs == sorted(set(seqs))  # strictly increasing, unique
+    finally:
+        worker.terminate()
+        server.stop()
+
+
+def test_polling_fallback_returns_the_same_durable_events(tmp_path):
+    seeded = seed_library(tmp_path, {"beach": 1}, {"beach": "sfw"})
+    log = tmp_path / "vision.log"
+    server = Server(seeded).start()
+    worker = start_worker(seeded, fake_vision_log=log)
+    try:
+        job = httpx.post(f"{server.base}/api/v1/analysis/jobs", timeout=TIMEOUT).json()
+        wait_until(lambda: _job_state(server.base, job["id"]) == "succeeded", timeout=TIMEOUT)
+
+        sse = _sse_events(server.base, job["id"])
+        polled = httpx.get(
+            f"{server.base}/api/v1/jobs/{job['id']}/events?after=0", timeout=TIMEOUT
+        ).json()
+        assert polled["state"] == "succeeded"
+        assert [(e["seq"], e["type"]) for e in polled["events"]] == [
+            (e["seq"], e["type"]) for e in sse
+        ]
+    finally:
+        worker.terminate()
+        server.stop()
+
+
+# ── cancellation ─────────────────────────────────────────────────────────────
+
+
+def test_queued_job_cancels_and_runs_no_items(tmp_path):
+    # No worker: the safety job stays queued so cancellation is deterministic.
+    seeded = seed_library(tmp_path, {"beach": 1, "city": 2}, {})
+    server = Server(seeded).start()
+    try:
+        job = httpx.post(f"{server.base}/api/v1/safety/jobs", timeout=TIMEOUT).json()
+        cancelled = httpx.post(
+            f"{server.base}/api/v1/jobs/{job['id']}/cancel", timeout=TIMEOUT
+        ).json()
+        assert cancelled["state"] == "cancelled"
+
+        snapshot = httpx.get(f"{server.base}/api/v1/jobs/{job['id']}", timeout=TIMEOUT).json()
+        assert snapshot["progress"]["by_state"].get("succeeded", 0) == 0
+    finally:
+        server.stop()
+
+
+# ── per-asset error inspection without losing completed work ──────────────────
+
+
+def test_one_analysis_error_is_inspectable_and_others_still_complete(tmp_path):
+    seeded = seed_library(tmp_path, {"beach": 1, "boom": 2}, {"beach": "sfw", "boom": "sfw"})
+    log = tmp_path / "vision.log"
+    server = Server(seeded).start()
+    worker = start_worker(seeded, fake_vision_log=log)
+    try:
+        job = httpx.post(f"{server.base}/api/v1/analysis/jobs", timeout=TIMEOUT).json()
+        wait_until(lambda: _job_state(server.base, job["id"]) == "succeeded", timeout=TIMEOUT)
+
+        counts = httpx.get(f"{server.base}/api/v1/analysis/counts", timeout=TIMEOUT).json()
+        assert counts["analyzed"] == 1 and counts["error"] == 1  # boom failed, beach kept
+
+        boom_id = seeded.asset_ids["boom"]
+        result = httpx.get(
+            f"{server.base}/api/v1/analysis/results/{boom_id}", timeout=TIMEOUT
+        ).json()
+        assert result["status"] == "error"
+    finally:
+        worker.terminate()
+        server.stop()
+
+
+# ── one mutating job at a time, read-only browsing stays available ───────────
+
+
+def test_second_mutation_rejected_while_reads_continue(tmp_path):
+    # No worker: the first safety job holds the library_write lock for the test.
+    seeded = seed_library(tmp_path, {"beach": 1, "city": 2}, {})
+    server = Server(seeded).start()
+    try:
+        first = httpx.post(f"{server.base}/api/v1/safety/jobs", timeout=TIMEOUT)
+        assert first.status_code == 200
+
+        second = httpx.post(f"{server.base}/api/v1/safety/jobs", timeout=TIMEOUT)
+        assert second.status_code == 409
+        assert second.json()["error"]["code"] == "lock_held"
+
+        # Read-only browsing is unaffected by the held mutation lock.
+        for path in ("/api/v1/workflow", "/api/v1/library/assets", "/api/v1/safety/queue"):
+            assert httpx.get(f"{server.base}{path}", timeout=TIMEOUT).status_code == 200
+    finally:
+        server.stop()
+
+
+# ── privacy gate: NSFW assets never reach the vision provider ────────────────
+
+
+def test_nsfw_asset_never_reaches_the_vision_provider(tmp_path):
+    seeded = seed_library(tmp_path, {"beach": 1, "city": 2}, {"beach": "sfw", "city": "nsfw"})
+    log = tmp_path / "vision.log"
+    server = Server(seeded).start()
+    worker = start_worker(seeded, fake_vision_log=log)
+    try:
+        job = httpx.post(f"{server.base}/api/v1/analysis/jobs", timeout=TIMEOUT).json()
+        wait_until(lambda: _job_state(server.base, job["id"]) == "succeeded", timeout=TIMEOUT)
+
+        logged = log.read_text()
+        assert "beach.jpg" in logged  # the SFW asset was analysed
+        assert "city.jpg" not in logged  # the NSFW asset never reached the provider
+
+        # And the NSFW asset was never even enqueued: no analysis row exists for it.
+        city_id = seeded.asset_ids["city"]
+        assert (
+            httpx.get(
+                f"{server.base}/api/v1/analysis/results/{city_id}", timeout=TIMEOUT
+            ).status_code
+            == 404
+        )
+    finally:
+        worker.terminate()
+        server.stop()
+
+
+# ── durability: decisions survive a full server restart ──────────────────────
+
+
+def test_safety_decisions_survive_a_restart(tmp_path):
+    seeded = seed_library(tmp_path, {"beach": 1, "city": 2}, {"beach": "sfw", "city": "nsfw"})
+    server = Server(seeded).start()
+    try:
+        before = httpx.get(f"{server.base}/api/v1/safety/queue?state=nsfw", timeout=TIMEOUT).json()
+        assert any(row["asset_id"] == seeded.asset_ids["city"] for row in before["items"])
+    finally:
+        server.stop()
+
+    # Fresh process against the same data dir: the durable decision is still there.
+    restarted = Server(seeded).start()
+    try:
+        after = httpx.get(
+            f"{restarted.base}/api/v1/safety/queue?state=nsfw", timeout=TIMEOUT
+        ).json()
+        assert any(row["asset_id"] == seeded.asset_ids["city"] for row in after["items"])
+    finally:
+        restarted.stop()

tests/e2e/test_worker_kill.py

@@ -14,6 +14,10 @@ from photo_pipeline.services.jobs import ItemState, JobService, JobState
 
 REPO = Path(__file__).resolve().parents[2]
 
+# Worker-recovery is also part of the Phase B acceptance command (US02-07: the
+# "resume" journey); it stays mapped to US02-03 for traceability.
+pytestmark = pytest.mark.phase_b
+
 # A worker that claims the job, marks its item running, signals it started, then
 # blocks — so the test can kill it mid-item.
 WORKER_SCRIPT = """