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
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")