photoanalyzer/tests/e2e/_pipeline_harness.py

"""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
import uuid
from dataclasses import dataclass
from datetime import datetime, timezone
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")


# ── Phase D: an analysed album, ready to be named and renamed ────────────────

NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)


class session_factory:
    """Session factory against a seeded database, for the few things a test has to
    set up or inspect below the API — journal states, mainly."""

    def __init__(self, seeded: Seeded) -> None:
        self._seeded = seeded

    def __enter__(self):
        from photo_pipeline.config import Config
        from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations

        config = Config.from_env(
            {
                "PHOTO_PIPELINE_DATA_DIR": str(self._seeded.data),
                "PHOTO_PIPELINE_LIBRARY_ROOTS": str(self._seeded.lib),
            }
        )
        run_migrations(config.database_url)
        self._engine = create_db_engine(config.database_url)
        return create_session_factory(self._engine)

    def __exit__(self, *_):
        self._engine.dispose()
        return False


def seed_album(tmp_path: Path, album: str = "rome", names: tuple[str, ...] = ("a.jpg", "b.jpg")):
    """A library holding one album folder whose photos are confirmed SFW and analysed
    — the state a naming proposal, and therefore a rename plan, is built from."""
    from sqlalchemy import select

    from photo_pipeline.models import AnalysisResult, Asset, SafetyReview
    from photo_pipeline.services.inventory import InventoryService

    seeded = seed_library(tmp_path, {}, {})
    folder = seeded.lib / album
    folder.mkdir(parents=True)
    for index, name in enumerate(names):
        image(folder / name, index + 1)

    with session_factory(seeded) as sf:
        InventoryService(sf).scan(seeded.lib)
        with sf() as session:
            rows = list(session.execute(select(Asset.id, Asset.current_path)).all())
            for asset_id, path in rows:
                session.add(
                    SafetyReview(
                        id=str(uuid.uuid4()), asset_id=asset_id, decision="sfw", created_at=NOW
                    )
                )
                session.add(
                    AnalysisResult(
                        asset_id=asset_id,
                        status="analyzed",
                        description=f"a view of {path}",
                        tags='["ruins", "city"]',
                        approx_year=2019,
                        location_hint="Rome",
                    )
                )
            session.commit()
        seeded.asset_ids.update({Path(path).stem: aid for aid, path in rows})
    return seeded


def approve_album(base: str, *, album: str = "rome", name: str) -> None:
    """Generate a proposal, set its final name, and approve it over HTTP."""
    httpx.post(f"{base}/api/v1/albums/proposals", json={}, timeout=10).raise_for_status()
    for payload, route in (
        ({"name": name}, "edit"),
        ({}, "approve"),
    ):
        current = httpx.get(f"{base}/api/v1/albums/proposals/{album}", timeout=10).json()
        httpx.post(
            f"{base}/api/v1/albums/proposals/{album}/{route}",
            json={**payload, "expected_version": current["version"]},
            timeout=10,
        ).raise_for_status()