diff --git a/frontend/js/api.js b/frontend/js/api.js index fa401f4..4648764 100644 --- a/frontend/js/api.js +++ b/frontend/js/api.js @@ -1,11 +1,22 @@ -// Single API client: one place for fetch, JSON, and the error envelope. -const BASE = "/api/v1"; +// Single API client: one place for fetch, JSON, the error envelope, and request +// cancellation. Every method accepts an optional { signal } from cancellable(). +export const BASE = "/api/v1"; -async function request(path, options = {}) { - const response = await fetch(BASE + path, { - headers: { "Content-Type": "application/json" }, - ...options, - }); +async function request(path, { signal, ...options } = {}) { + let response; + try { + response = await fetch(BASE + path, { + headers: { "Content-Type": "application/json" }, + signal, + ...options, + }); + } catch (error) { + // A caller-cancelled fetch is not a failure; tag it so views can ignore it. + if (error.name === "AbortError") { + error.code = "cancelled"; + } + throw error; + } let body = null; try { body = await response.json(); @@ -22,17 +33,37 @@ async function request(path, options = {}) { return body; } +// Cancellation: hand callers a signal to pass into any request and a cancel() to +// abort the in-flight fetch. Aborted requests reject with code "cancelled". +export function cancellable() { + const controller = new AbortController(); + return { signal: controller.signal, cancel: () => controller.abort() }; +} + export const api = { - listAssets: (params = {}) => - request("/inventory/assets?" + new URLSearchParams(params).toString()), - listClusters: (params = {}) => - request("/duplicates/clusters?" + new URLSearchParams(params).toString()), - getCluster: (id) => request(`/duplicates/clusters/${encodeURIComponent(id)}`), - decide: (id, payload) => + listAssets: (params = {}, opts = {}) => + request("/inventory/assets?" + new URLSearchParams(params).toString(), opts), + listClusters: (params = {}, opts = {}) => + request("/duplicates/clusters?" + new URLSearchParams(params).toString(), opts), + getCluster: (id, opts = {}) => + request(`/duplicates/clusters/${encodeURIComponent(id)}`, opts), + decide: (id, payload, opts = {}) => request(`/duplicates/clusters/${encodeURIComponent(id)}/decision`, { method: "POST", body: JSON.stringify(payload), + ...opts, }), thumbnailUrl: (assetId, size = 512) => `${BASE}/assets/${encodeURIComponent(assetId)}/thumbnail?size=${size}`, + + // ── Jobs ──────────────────────────────────────────────────────────────── + startJob: (payload, opts = {}) => + request("/jobs", { method: "POST", body: JSON.stringify(payload), ...opts }), + getJob: (id, opts = {}) => request(`/jobs/${encodeURIComponent(id)}`, opts), + cancelJob: (id, opts = {}) => + request(`/jobs/${encodeURIComponent(id)}/cancel`, { method: "POST", ...opts }), + jobEvents: (id, after = 0, opts = {}) => + request(`/jobs/${encodeURIComponent(id)}/events?after=${after}`, opts), + jobEventsStreamUrl: (id, after = 0) => + `${BASE}/jobs/${encodeURIComponent(id)}/events/stream?after=${after}`, }; diff --git a/frontend/js/events.js b/frontend/js/events.js new file mode 100644 index 0000000..0bf905e --- /dev/null +++ b/frontend/js/events.js @@ -0,0 +1,62 @@ +// Job activity adapter: prefer SSE (resumable, pushed), fall back to polling the +// same durable events when EventSource is missing or the stream errors. The cursor +// is the event `seq`, shared by both transports, so switching mid-stream never +// drops or repeats an event. subscribeJob returns a stop() handle. +import { api } from "./api.js"; + +// Mirrors ACTIVE_STATES in services/jobs.py; a job outside this set is terminal +// and the poller stops. +const ACTIVE = new Set(["queued", "running", "cancelling", "retry_queued"]); + +export function subscribeJob(jobId, { onEvent, onDone, onError, after = 0, pollMs = 1000 } = {}) { + let cursor = after; + let stopped = false; + let source = null; + let timer = null; + + function stop() { + if (stopped) return; + stopped = true; + if (source) source.close(); + if (timer) clearTimeout(timer); + if (onDone) onDone(cursor); + } + + async function poll() { + if (stopped) return; + try { + const data = await api.jobEvents(jobId, cursor); + for (const event of data.events) { + cursor = event.seq; + if (onEvent) onEvent(event); + } + if (!ACTIVE.has(data.state)) return stop(); + } catch (error) { + if (!stopped && onError) onError(error); + } + if (!stopped) timer = setTimeout(poll, pollMs); + } + + if (typeof EventSource === "function") { + source = new EventSource(api.jobEventsStreamUrl(jobId, cursor)); + source.onmessage = (message) => { + const event = JSON.parse(message.data); + cursor = Number(message.lastEventId) || cursor; + if (onEvent) onEvent({ seq: cursor, ...event }); + }; + // Server sends `event: done` when the job reaches a terminal state. + source.addEventListener("done", stop); + // Network/stream error: drop SSE and continue from the same cursor via polling + // so no consumer notices the transport switch. + source.onerror = () => { + if (stopped || !source) return; + source.close(); + source = null; + poll(); + }; + } else { + poll(); + } + + return { stop }; +} diff --git a/frontend/js/store.js b/frontend/js/store.js new file mode 100644 index 0000000..611e327 --- /dev/null +++ b/frontend/js/store.js @@ -0,0 +1,23 @@ +// Shared observable store: one state object, subscribe for changes, set() to +// patch. Views read state and re-render on notify. Nothing here touches the URL +// (the router owns navigable state) or the network (the api client owns that) — +// this is only the in-memory glue between them. +export function createStore(initial = {}) { + let state = { ...initial }; + const listeners = new Set(); + + function set(patch) { + const next = typeof patch === "function" ? patch(state) : patch; + state = { ...state, ...next }; + for (const listener of listeners) listener(state); + } + + return { + get: () => state, + set, + subscribe(listener) { + listeners.add(listener); + return () => listeners.delete(listener); + }, + }; +} diff --git a/frontend/js/tests/harness.html b/frontend/js/tests/harness.html new file mode 100644 index 0000000..e797219 --- /dev/null +++ b/frontend/js/tests/harness.html @@ -0,0 +1,10 @@ + + + + + Shell unit tests + + + + + diff --git a/frontend/js/tests/unit.js b/frontend/js/tests/unit.js new file mode 100644 index 0000000..d24fdf7 --- /dev/null +++ b/frontend/js/tests/unit.js @@ -0,0 +1,150 @@ +// In-browser unit tests for the shell primitives. No server, no JS toolchain: +// globals (fetch, EventSource) are stubbed and the modules are exercised directly. +// Driven by tests/e2e/test_frontend_shell.py, which loads harness.html and reads +// window.__RESULTS__. Covers API errors + cancellation, store transitions, +// routing, and the SSE→polling event fallback (US02-05 acceptance). +import { createStore } from "../store.js"; +import { parseHash, navigate } from "../router.js"; +import { api, cancellable } from "../api.js"; +import { subscribeJob } from "../events.js"; + +const cases = []; +function ok(name, cond) { + cases.push({ name, ok: !!cond }); +} +async function throws(name, fn, predicate) { + try { + await fn(); + cases.push({ name, ok: false }); + } catch (error) { + cases.push({ name, ok: !!predicate(error) }); + } +} + +const realFetch = window.fetch; +const realES = window.EventSource; +function jsonResponse(status, body) { + return { ok: status < 400, status, statusText: "", json: async () => body }; +} +const tick = (ms = 0) => new Promise((r) => setTimeout(r, ms)); + +async function run() { + // ── store ──────────────────────────────────────────────────────────────── + { + const store = createStore({ n: 0 }); + const seen = []; + const off = store.subscribe((s) => seen.push(s.n)); + store.set({ n: 1 }); + store.set((s) => ({ n: s.n + 1 })); + ok("store applies object and updater patches", store.get().n === 2); + ok("store notifies each change", seen.length === 2 && seen[1] === 2); + off(); + store.set({ n: 99 }); + ok("store unsubscribe stops notifications", seen.length === 2); + } + + // ── router ─────────────────────────────────────────────────────────────── + { + navigate("/inventory", { q: "beach", offset: 0, empty: "" }); + const { path, params } = parseHash(); + ok("router restores path", path === "/inventory"); + ok("router restores non-empty filters", params.q === "beach" && params.offset === "0"); + ok("router drops empty filters", !("empty" in params)); + location.hash = ""; + ok("router defaults to /inventory", parseHash().path === "/inventory"); + } + + // ── api errors + cancellation ──────────────────────────────────────────── + { + window.fetch = async () => + jsonResponse(404, { error: { code: "not_found", message: "unknown job x" } }); + await throws( + "api surfaces the error envelope", + () => api.getJob("x"), + (e) => e.status === 404 && e.code === "not_found" && e.message === "unknown job x" + ); + + window.fetch = async () => { + const err = new Error("aborted"); + err.name = "AbortError"; + throw err; + }; + const { signal, cancel } = cancellable(); + cancel(); + await throws( + "api tags aborted requests as cancelled", + () => api.getJob("x", { signal }), + (e) => e.code === "cancelled" + ); + window.fetch = realFetch; + } + + // ── event fallback: polling when EventSource is unavailable ─────────────── + { + window.EventSource = undefined; + const pages = [ + jsonResponse(200, { state: "running", events: [{ seq: 1, type: "queued", message: null }] }), + jsonResponse(200, { + state: "succeeded", + events: [{ seq: 2, type: "state:succeeded", message: null }], + }), + ]; + let call = 0; + window.fetch = async () => pages[Math.min(call++, pages.length - 1)]; + const got = []; + await new Promise((resolve) => { + subscribeJob("job1", { + pollMs: 1, + onEvent: (e) => got.push(e.seq), + onDone: resolve, + }); + }); + ok("polling delivers events in order", got.join(",") === "1,2"); + ok("polling stops on terminal state", call >= 2); + window.fetch = realFetch; + } + + // ── event stream: SSE path, then done ───────────────────────────────────── + { + class FakeES { + constructor() { + this.listeners = {}; + setTimeout(() => { + this.onmessage?.({ lastEventId: "5", data: JSON.stringify({ type: "state:running", message: "go" }) }); + this.listeners.done?.(); + }, 0); + } + addEventListener(type, fn) { + this.listeners[type] = fn; + } + close() { + this.closed = true; + } + } + window.EventSource = FakeES; + let polled = false; + window.fetch = async () => { + polled = true; + return jsonResponse(200, { state: "succeeded", events: [] }); + }; + const got = []; + await new Promise((resolve) => { + subscribeJob("job2", { onEvent: (e) => got.push(e), onDone: resolve }); + }); + ok("SSE delivers event with seq from lastEventId", got.length === 1 && got[0].seq === 5); + ok("SSE carries type from payload", got[0].type === "state:running"); + ok("SSE path does not poll", polled === false); + window.fetch = realFetch; + window.EventSource = realES; + } + + await tick(); + const failed = cases.filter((c) => !c.ok); + window.__RESULTS__ = { passed: cases.length - failed.length, failed: failed.length, cases }; + const pre = document.createElement("pre"); + pre.id = "results"; + pre.textContent = cases.map((c) => `${c.ok ? "PASS" : "FAIL"} ${c.name}`).join("\n"); + document.body.appendChild(pre); +} + +run(); diff --git a/photo_pipeline/api/routes/jobs.py b/photo_pipeline/api/routes/jobs.py index 1ba42b4..be9eb6f 100644 --- a/photo_pipeline/api/routes/jobs.py +++ b/photo_pipeline/api/routes/jobs.py @@ -102,10 +102,13 @@ async def stream_events(job_id: str, request: Request, after: int = 0): for _ in range(600): for event in service.events_after(job_id, cursor): cursor = event["seq"] + # Default (unnamed) event so a browser EventSource receives every + # event via onmessage without enumerating the open-ended type set + # (state:*, claimed, …); the type travels in the JSON payload, + # matching the polling fallback's {seq, type, message} shape. yield ( f"id: {event['seq']}\n" - f"event: {event['type']}\n" - f"data: {json.dumps({'message': event['message']})}\n\n" + f"data: {json.dumps({'type': event['type'], 'message': event['message']})}\n\n" ) snapshot = service.get(job_id) if snapshot is None or snapshot["state"] not in ACTIVE_STATES: diff --git a/tests/e2e/test_frontend_shell.py b/tests/e2e/test_frontend_shell.py new file mode 100644 index 0000000..69d1a2f --- /dev/null +++ b/tests/e2e/test_frontend_shell.py @@ -0,0 +1,143 @@ +"""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")