photoanalyzer/tests/e2e/test_frontend_shell.py

"""Static application shell: the JS unit suite plus browser journeys asserting
asset loading, deep links, reload-restores-filters, JSON-only APIs, and a clean
console/network (US02-05).

The unit suite (frontend/js/tests/) runs in the browser over the server's own
static mount — ES modules can't import over file://, so the running server serves
them; the tests themselves stub fetch/EventSource and never touch the API.
"""

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]


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

    from photo_pipeline.config import Config
    from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
    from photo_pipeline.services.inventory import InventoryService

    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)
    InventoryService(create_session_factory(engine)).scan(lib)
    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_js_unit_suite_passes(page, server):
    page.goto(f"{server.base}/app/js/tests/harness.html")
    page.locator("#results").wait_for()
    results = page.evaluate("window.__RESULTS__")
    failing = [c["name"] for c in results["cases"] if not c["ok"]]
    assert results["failed"] == 0, f"JS unit failures: {failing}"
    assert results["passed"] >= 12


def test_shell_loads_assets_without_console_or_network_errors(page, server):
    errors, failed = [], []
    page.on("console", lambda m: errors.append(m.text) if m.type == "error" else None)
    page.on("requestfailed", lambda r: failed.append(r.url))

    page.goto(f"{server.base}/app/#/inventory")
    page.get_by_test_id("asset-row").first.wait_for()
    assert page.locator("nav a[data-nav]").count() >= 2
    assert errors == [], f"console errors: {errors}"
    assert failed == [], f"failed requests: {failed}"


def test_deep_link_and_reload_restore_view_and_filters(page, server):
    # Deep-link straight to a filtered inventory view, then reload: the hash router
    # must restore both the view and the filter with no server-rendered state.
    page.goto(f"{server.base}/app/#/inventory?q=beach")
    search = page.get_by_label("Filter by path")
    search.wait_for()
    assert search.input_value() == "beach"

    page.reload()
    page.get_by_label("Filter by path").wait_for()
    assert page.get_by_label("Filter by path").input_value() == "beach"
    assert "q=beach" in page.evaluate("location.hash")


def test_api_is_json_only_and_html_never_intercepts(server):
    # /api/v1 always answers JSON — the static SPA mount must never shadow it.
    ok = httpx.get(f"{server.base}/api/v1/inventory/assets", timeout=5)
    assert ok.headers["content-type"].startswith("application/json")

    missing = httpx.get(f"{server.base}/api/v1/does-not-exist", timeout=5)
    assert missing.status_code == 404
    assert missing.headers["content-type"].startswith("application/json")

    shell = httpx.get(f"{server.base}/app/", timeout=5)
    assert shell.headers["content-type"].startswith("text/html")