photoanalyzer/tests/e2e/test_workflow_views.py

"""Browser journeys for the Phase B views (US02-06): Workflow home, Safety review,
Library, Analyze, Stats — plus the read-only-during-jobs rule.

The library is seeded in-process (scan → one SFW/one NSFW decision → one fake-analyzed
result) before the real server starts, so the views have data to render exactly as
they would in production.
"""

import socket
import subprocess
import sys
import time
from pathlib import Path
from types import SimpleNamespace

import httpx
import numpy as np
import pytest
from PIL import Image

REPO = Path(__file__).resolve().parents[2]


def _image(path, seed):
    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():
    with socket.socket() as sock:
        sock.bind(("127.0.0.1", 0))
        return sock.getsockname()[1]


class _Fake:
    def analyze(self, path, *, album_hint):
        return {"description": "a sunny beach at golden hour", "tags": ["beach", "sunset"], "people_count": 2, "setting": "outdoor"}


@pytest.fixture
def server(tmp_path):
    data = tmp_path / "data"
    data.mkdir()
    lib = tmp_path / "lib"
    lib.mkdir()
    _image(lib / "beach.jpg", 1)
    _image(lib / "city.jpg", 2)
    _image(lib / "night.jpg", 3)

    from photo_pipeline.config import Config
    from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
    from photo_pipeline.services.analysis import AnalysisService
    from photo_pipeline.services.inventory import InventoryService
    from photo_pipeline.services.safety import SafetyService

    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)
    from sqlalchemy import select

    from photo_pipeline.models import Asset

    with sf() as session:
        ids = list(session.scalars(select(Asset.id).order_by(Asset.current_path)))
    safety = SafetyService(sf)
    safety.decide(ids[0], "sfw", write_exif=False)
    safety.decide(ids[1], "nsfw", write_exif=False)
    AnalysisService(sf, provider=_Fake()).run([ids[0]])
    engine.dispose()

    port = _free_port()
    env = {
        "PATH": __import__("os").environ.get("PATH", ""),
        "PHOTO_PIPELINE_DATA_DIR": str(data),
        "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib),
        "PHOTO_PIPELINE_HOST": "127.0.0.1",
        "PHOTO_PIPELINE_PORT": str(port),
    }
    proc = subprocess.Popen(
        [sys.executable, "-m", "photo_pipeline", "serve"],
        cwd=str(REPO),
        env=env,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )
    base = f"http://127.0.0.1:{port}"
    deadline = time.monotonic() + 30
    ready = False
    while time.monotonic() < deadline:
        if proc.poll() is not None:
            out, err = proc.communicate()
            pytest.fail(f"server exited: {err.decode(errors='replace')}")
        try:
            if httpx.get(f"{base}/api/v1/health/ready", timeout=1).status_code == 200:
                ready = True
                break
        except httpx.HTTPError:
            time.sleep(0.2)
    if not ready:
        proc.terminate()
        pytest.fail("server never became ready")
    try:
        yield SimpleNamespace(base=base)
    finally:
        proc.terminate()
        try:
            proc.wait(timeout=10)
        except subprocess.TimeoutExpired:
            proc.kill()


def test_workflow_home_shows_cards_with_counts_and_labels(page, server):
    page.goto(f"{server.base}/app/#/workflow")
    page.get_by_test_id("stage-safety").wait_for()
    # Status is text, not color alone.
    assert page.get_by_test_id("status-safety").inner_text().strip() != ""
    # 1 sfw, 1 nsfw, 1 undecided from the seed.
    assert "1 undecided" in page.get_by_test_id("counts-safety").inner_text()
    assert page.get_by_test_id("stage-analysis").count() == 1
    assert page.get_by_test_id("action-safety").is_visible()


def test_actions_disabled_and_explained_while_a_job_runs(page, server):
    # Start a durable job (no worker consumes it), then the home page must disable
    # job-starting actions and explain why.
    httpx.post(f"{server.base}/api/v1/safety/jobs", timeout=5).raise_for_status()
    page.goto(f"{server.base}/app/#/workflow")
    page.get_by_test_id("job-running").wait_for()
    assert page.get_by_test_id("action-safety").is_disabled()


def test_safety_review_decide_persists_across_reload(page, server):
    page.goto(f"{server.base}/app/#/safety?state=undecided")
    row = page.get_by_test_id("safety-row").first
    row.wait_for()
    asset_id = row.get_attribute("data-asset-id")
    row.get_by_test_id("decide-sfw").click()

    # The decided asset leaves the undecided queue and appears under SFW after reload.
    page.goto(f"{server.base}/app/#/safety?state=sfw")
    page.get_by_test_id("safety-body").wait_for()
    page.wait_for_selector(f'[data-asset-id="{asset_id}"]')


def test_library_search_and_stats(page, server):
    page.goto(f"{server.base}/app/#/library?q=beach")
    page.get_by_test_id("library-card").first.wait_for()
    assert "beach" in page.get_by_test_id("library-card").first.inner_text().lower()

    page.goto(f"{server.base}/app/#/stats")
    page.get_by_test_id("stats-albums").wait_for()
    assert page.get_by_test_id("stats-status").inner_text().strip() != ""


def test_analyze_view_shows_counts(page, server):
    page.goto(f"{server.base}/app/#/analyze")
    page.get_by_test_id("analyze-counts").wait_for()
    assert page.get_by_test_id("run-analysis").is_visible()