Eric Wagoner
d3cf0c6151
One shared stylesheet (static/app.css) replaces the two ad-hoc style blocks, with tokens drawn from the mascot drawing: sky background, cream game-board cards inside confident outlines with flat offset shadows, hair-purple for brand and actions, shirt-green for go, bow-tie orange for danger, pipe-fitting gold for trim, jeans navy for chrome — and the rainbow game path as a stripe under the header, the one loud element. Accent colors split into object and *-ink variants so text on light surfaces holds AA contrast. Navigation: both pages share a header with brand-home link, a Primary nav with aria-current, and tally counts that link to the lists they count (dashboard tallies deep-link into the review catalog; review tallies jump to their sections). Accessibility: skip link, landmark nav, polite live region for stage activity, status role on banners, the dropzone is a real button, candidate rows are focusable and activate with Enter/Space, focus-visible ring throughout, reduced motion respected. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
217 lines
6.6 KiB
Python
217 lines
6.6 KiB
Python
"""Dashboard/API tests: job lifecycle, pipeline status, photo upload —
|
|
stage functions are injected so nothing slow or networked ever runs."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
|
|
import typer
|
|
from fastapi.testclient import TestClient
|
|
|
|
from bggpipe.config import Config
|
|
from bggpipe.jobs import JobRunner
|
|
from bggpipe.webreview import create_app
|
|
|
|
|
|
def _cfg(tmp_path) -> Config:
|
|
cfg = Config(data_dir=tmp_path / "data", photos_dir=tmp_path / "photos")
|
|
cfg.photos_dir.mkdir(parents=True)
|
|
cfg.data_dir.mkdir(parents=True)
|
|
return cfg
|
|
|
|
|
|
def _app(cfg, stages=None, jobs=None) -> TestClient:
|
|
return TestClient(create_app(cfg, stages=stages or {}, jobs=jobs))
|
|
|
|
|
|
# -- JobRunner ----------------------------------------------------------
|
|
|
|
|
|
def test_job_captures_output_and_finishes():
|
|
runner = JobRunner()
|
|
|
|
def stage():
|
|
typer.echo("line one")
|
|
typer.echo("line two")
|
|
|
|
assert runner.start("extract", stage)
|
|
runner.wait()
|
|
snap = runner.snapshot()
|
|
assert snap["status"] == "done"
|
|
assert snap["log"] == ["line one", "line two"]
|
|
|
|
|
|
def test_job_failure_is_reported_not_swallowed():
|
|
runner = JobRunner()
|
|
runner.start("resolve", lambda: (_ for _ in ()).throw(RuntimeError("boom")))
|
|
runner.wait()
|
|
snap = runner.snapshot()
|
|
assert snap["status"] == "failed"
|
|
assert "RuntimeError: boom" in snap["error"]
|
|
|
|
|
|
def test_typer_exit_code_counts_as_failure():
|
|
runner = JobRunner()
|
|
|
|
def stage():
|
|
raise typer.Exit(code=1)
|
|
|
|
runner.start("diff", stage)
|
|
runner.wait()
|
|
assert runner.snapshot()["status"] == "failed"
|
|
|
|
|
|
def test_single_slot_rejects_second_job():
|
|
runner = JobRunner()
|
|
release = threading.Event()
|
|
assert runner.start("extract", release.wait)
|
|
assert not runner.start("resolve", lambda: None) # slot busy
|
|
release.set()
|
|
runner.wait()
|
|
assert runner.start("resolve", lambda: None) # slot free again
|
|
runner.wait()
|
|
|
|
|
|
# -- /api/run + /api/job ------------------------------------------------
|
|
|
|
|
|
def test_run_stage_lifecycle_via_api(tmp_path):
|
|
cfg = _cfg(tmp_path)
|
|
ran = []
|
|
jobs = JobRunner()
|
|
web = _app(cfg, stages={"extract": lambda: ran.append(1)}, jobs=jobs)
|
|
|
|
res = web.post("/api/run/extract")
|
|
assert res.status_code == 200
|
|
jobs.wait()
|
|
assert ran == [1]
|
|
assert web.get("/api/job").json()["status"] == "done"
|
|
|
|
|
|
def test_unknown_stage_404s(tmp_path):
|
|
web = _app(_cfg(tmp_path))
|
|
assert web.post("/api/run/frobnicate").status_code == 404
|
|
|
|
|
|
def test_busy_runner_409s(tmp_path):
|
|
cfg = _cfg(tmp_path)
|
|
release = threading.Event()
|
|
jobs = JobRunner()
|
|
web = _app(
|
|
cfg, stages={"extract": release.wait, "resolve": lambda: None}, jobs=jobs
|
|
)
|
|
assert web.post("/api/run/extract").status_code == 200
|
|
assert web.post("/api/run/resolve").status_code == 409
|
|
release.set()
|
|
jobs.wait()
|
|
|
|
|
|
def test_upload_defaults_to_dry_run(tmp_path):
|
|
cfg = _cfg(tmp_path)
|
|
calls = []
|
|
jobs = JobRunner()
|
|
|
|
def upload(dry_run=True, limit=None):
|
|
calls.append({"dry_run": dry_run, "limit": limit})
|
|
|
|
web = _app(cfg, stages={"upload": upload}, jobs=jobs)
|
|
web.post("/api/run/upload") # no body: the safe direction
|
|
jobs.wait()
|
|
web.post("/api/run/upload", json={"dry_run": False, "limit": 2})
|
|
jobs.wait()
|
|
assert calls == [
|
|
{"dry_run": True, "limit": None},
|
|
{"dry_run": False, "limit": 2},
|
|
]
|
|
|
|
|
|
# -- /api/pipeline ------------------------------------------------------
|
|
|
|
|
|
def test_pipeline_reports_counts_and_never_values(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("BGG_USERNAME", "supersecretname")
|
|
monkeypatch.delenv("BGG_API_TOKEN", raising=False)
|
|
cfg = _cfg(tmp_path)
|
|
(cfg.photos_dir / "a.jpg").write_bytes(b"x")
|
|
web = _app(cfg)
|
|
payload = web.get("/api/pipeline").json()
|
|
assert payload["photos"] == 1
|
|
assert payload["env"]["BGG_USERNAME"] is True
|
|
assert payload["env"]["BGG_API_TOKEN"] is False
|
|
assert "supersecretname" not in web.get("/api/pipeline").text
|
|
|
|
|
|
def test_pipeline_flags_stub_data(tmp_path):
|
|
cfg = _cfg(tmp_path)
|
|
(cfg.data_dir / "STUB_DATA.marker").write_text("stub")
|
|
assert _app(cfg).get("/api/pipeline").json()["stub_data"] is True
|
|
|
|
|
|
# -- /api/photos --------------------------------------------------------
|
|
|
|
|
|
def test_photo_upload_saves_and_invalidates_raw_cache(tmp_path):
|
|
cfg = _cfg(tmp_path)
|
|
cfg.extract_raw_dir.mkdir(parents=True)
|
|
stale = cfg.extract_raw_dir / "shelf.jpg.json"
|
|
stale.write_text("{}")
|
|
web = _app(cfg)
|
|
res = web.post(
|
|
"/api/photos", files={"files": ("shelf.jpg", b"\xff\xd8jpegdata", "image/jpeg")}
|
|
)
|
|
assert res.status_code == 200
|
|
assert (cfg.photos_dir / "shelf.jpg").read_bytes() == b"\xff\xd8jpegdata"
|
|
assert not stale.exists() # re-upload means re-extract
|
|
|
|
|
|
def test_photo_upload_rejects_non_photos_and_path_tricks(tmp_path):
|
|
cfg = _cfg(tmp_path)
|
|
web = _app(cfg)
|
|
res = web.post("/api/photos", files={"files": ("notes.txt", b"hi", "text/plain")})
|
|
assert res.status_code == 400
|
|
res = web.post(
|
|
"/api/photos",
|
|
files={"files": ("../../escape.jpg", b"x", "image/jpeg")},
|
|
)
|
|
if res.status_code == 200: # client may strip the path; the name must be bare
|
|
assert (cfg.photos_dir / "escape.jpg").exists()
|
|
assert not (tmp_path / "escape.jpg").exists()
|
|
|
|
|
|
# -- pages --------------------------------------------------------------
|
|
|
|
|
|
def test_dashboard_and_review_pages_serve(tmp_path):
|
|
web = _app(_cfg(tmp_path))
|
|
assert "Pipeline" in web.get("/").text
|
|
assert "bggpipe" in web.get("/review").text
|
|
|
|
|
|
# -- design system + navigation -----------------------------------------
|
|
|
|
|
|
def test_stylesheet_is_served_and_linked_by_both_pages(tmp_path):
|
|
web = _app(_cfg(tmp_path))
|
|
css = web.get("/static/app.css")
|
|
assert css.status_code == 200
|
|
assert css.headers["content-type"].startswith("text/css")
|
|
assert "--accent" in css.text # the token layer, not an empty file
|
|
for path in ("/", "/review"):
|
|
assert 'href="/static/app.css"' in web.get(path).text
|
|
|
|
|
|
def test_both_pages_carry_navigation_and_skip_link(tmp_path):
|
|
web = _app(_cfg(tmp_path))
|
|
for path, current in (("/", 'href="/"'), ("/review", 'href="/review"')):
|
|
html = web.get(path).text
|
|
assert 'nav aria-label="Primary"' in html
|
|
assert f'<a {current} aria-current="page"' in html.replace("\n", " ") or (
|
|
current in html and 'aria-current="page"' in html
|
|
)
|
|
assert 'class="skip"' in html
|
|
|
|
|
|
def test_activity_region_announces_politely(tmp_path):
|
|
html = _app(_cfg(tmp_path)).get("/").text
|
|
assert 'aria-live="polite"' in html
|