Files
bggpipe/tests/test_webreview.py
T
Eric Wagoner 824719edc7 Mascot: Juniper's bggpipe piper joins the header, favicon, and README
Original art (a bagpiper whose bag is a board game box) lives in
assets/; web-sized derivatives ship in the package: a face-crop avatar
in the header and favicon, and the framed full-length piper on the
review done screen. Served via an allowlisted /static route. README
leads with the full portrait, credited to Juniper.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-08-01 17:35:53 -04:00

337 lines
10 KiB
Python

"""Web review UI tests via FastAPI's TestClient — no server, no network,
no live BGG (the injected client 401s on any cache miss)."""
from __future__ import annotations
import json
import httpx
from fastapi.testclient import TestClient
from bggpipe.bgg_client import BGGClient, cache_key
from bggpipe.config import Config
from bggpipe.resolve import read_matches, write_matches
from bggpipe.webreview import create_app, load_thumbnails
CITADELS_CANDIDATES = json.dumps(
[
{
"bgg_id": 478,
"name": "Citadels",
"year": 2000,
"type": "boardgame",
"owned": 85000,
"rank": 250,
},
{
"bgg_id": 205398,
"name": "Citadels",
"year": 2016,
"type": "boardgame",
"owned": 24000,
"rank": 400,
},
]
)
VERSION_CANDIDATES = json.dumps(
[
{
"version_id": 111,
"name": "First edition",
"year": 1975,
"publishers": ["TSR"],
"languages": ["English"],
"score": 3,
},
{
"version_id": 222,
"name": "Second edition",
"year": 1980,
"publishers": ["TSR"],
"languages": ["English"],
"score": 3,
},
]
)
def _row(**overrides) -> dict:
row = {
"title_raw": "",
"bgg_id": "",
"bgg_name": "",
"year": "",
"type": "",
"match_status": "auto",
"version_id": "",
"version_name": "",
"version_status": "version_unknown",
"candidates_json": "[]",
"version_candidates_json": "[]",
"source_photos": "shelf.jpg",
}
row.update(overrides)
return row
def make_cfg(tmp_path) -> Config:
cfg = Config(data_dir=tmp_path / "data", photos_dir=tmp_path / "photos")
cfg.photos_dir.mkdir(parents=True)
(cfg.photos_dir / "shelf.jpg").write_bytes(b"\xff\xd8\xff\xdbfakejpeg")
write_matches(
cfg.matches_path,
[
_row(
title_raw="Citadels",
match_status="ambiguous",
candidates_json=CITADELS_CANDIDATES,
),
_row(title_raw="Mystery", match_status="unmatched"),
_row(
title_raw="Dungeon!",
match_status="auto",
bgg_id="1339",
bgg_name="Dungeon!",
version_status="version_ambiguous",
version_candidates_json=VERSION_CANDIDATES,
),
],
)
cfg.titles_path.write_text(
json.dumps(
[
{
"title_raw": "Citadels",
"publisher_hint": "Fantasy Flight",
"edition_hint": "",
"source_photos": ["shelf.jpg"],
},
{
# extracted but never resolved (no BGG token yet)
"title_raw": "Fresh Off The Shelf",
"confidence": "high",
"source_photos": ["shelf.jpg"],
},
]
)
)
cfg.unidentified_path.write_text(
json.dumps(
{
"shelf.jpg": [
{
"location": "top shelf, far left",
"partial_text": "EMP",
"art_notes": "black box, gold letters",
}
]
}
)
)
return cfg
def unauthorized_client(tmp_path) -> BGGClient:
return BGGClient(
cache_dir=tmp_path / "no_cache",
transport=httpx.MockTransport(
lambda req: httpx.Response(401, text="Unauthorized")
),
)
def make_client(tmp_path) -> tuple[TestClient, Config]:
cfg = make_cfg(tmp_path)
app = create_app(cfg, client=unauthorized_client(tmp_path))
return TestClient(app), cfg
def test_index_serves_page(tmp_path):
web, _ = make_client(tmp_path)
response = web.get("/")
assert response.status_code == 200
assert "bggpipe" in response.text
def test_state_lists_pending_versions_and_tickets(tmp_path):
web, _ = make_client(tmp_path)
state = web.get("/api/state").json()
assert [r["title_raw"] for r in state["pending"]] == ["Citadels", "Mystery"]
citadels = state["pending"][0]
assert citadels["cues"]["publisher"] == "Fantasy Flight"
assert citadels["photos"] == ["shelf.jpg"]
assert citadels["candidates"][0]["thumbnail"] is None # stub cache: placeholder
assert [v["title_raw"] for v in state["versions"]] == ["Dungeon!"]
assert state["unidentified"][0]["location"] == "top shelf, far left"
assert state["summary"]["total"] == 3
def test_pick_candidate_persists(tmp_path):
web, cfg = make_client(tmp_path)
state = web.post(
"/api/decision",
json={
"title_raw": "Citadels",
"source_photos": "shelf.jpg",
"action": "pick",
"bgg_id": 205398,
},
).json()
assert [r["title_raw"] for r in state["pending"]] == ["Mystery"]
saved = {r["title_raw"]: r for r in read_matches(cfg.matches_path)}
assert saved["Citadels"]["match_status"] == "approved"
assert saved["Citadels"]["bgg_id"] == "205398"
def test_manual_id_degrades_without_token(tmp_path):
web, cfg = make_client(tmp_path)
response = web.post(
"/api/decision",
json={
"title_raw": "Mystery",
"source_photos": "shelf.jpg",
"action": "manual",
"bgg_id": 99999,
},
)
assert response.status_code == 200
saved = {r["title_raw"]: r for r in read_matches(cfg.matches_path)}
assert saved["Mystery"]["match_status"] == "approved"
assert saved["Mystery"]["bgg_id"] == "99999"
assert saved["Mystery"]["bgg_name"] == "" # lookup blocked, id recorded
def test_reject_and_bad_pick(tmp_path):
web, cfg = make_client(tmp_path)
web.post(
"/api/decision",
json={"title_raw": "Mystery", "source_photos": "shelf.jpg", "action": "reject"},
)
saved = {r["title_raw"]: r for r in read_matches(cfg.matches_path)}
assert saved["Mystery"]["match_status"] == "rejected"
bad = web.post(
"/api/decision",
json={
"title_raw": "Citadels",
"source_photos": "shelf.jpg",
"action": "pick",
"bgg_id": 42,
},
)
assert bad.status_code == 400
def test_version_pick_and_unknown(tmp_path):
web, cfg = make_client(tmp_path)
state = web.post(
"/api/version",
json={
"title_raw": "Dungeon!",
"source_photos": "shelf.jpg",
"action": "pick",
"version_id": 222,
},
).json()
assert state["versions"] == []
saved = {r["title_raw"]: r for r in read_matches(cfg.matches_path)}
assert saved["Dungeon!"]["version_status"] == "version_approved"
assert saved["Dungeon!"]["version_id"] == "222"
assert state["summary"]["version_updates"] == 1
def test_dismiss_persists_across_restarts(tmp_path):
web, cfg = make_client(tmp_path)
state = web.post(
"/api/dismiss",
json={
"photo": "shelf.jpg",
"location": "top shelf, far left",
"partial_text": "EMP",
"art_notes": "black box, gold letters",
},
).json()
assert state["unidentified"] == []
# a brand-new app instance (fresh server) still honors the dismissal
web2 = TestClient(create_app(cfg, client=unauthorized_client(tmp_path)))
assert web2.get("/api/state").json()["unidentified"] == []
def test_photo_serving_is_locked_down(tmp_path):
web, _ = make_client(tmp_path)
assert web.get("/photos/shelf.jpg").status_code == 200
assert web.get("/photos/nope.jpg").status_code == 404
assert web.get("/photos/..%2Fdata%2Fmatches.csv").status_code == 404
def test_thumbnails_come_from_cached_thing_xml(tmp_path):
cache = tmp_path / "cache"
cache.mkdir()
key = cache_key("thing", {"id": "478", "stats": "1"})
(cache / key).write_text(
'<items><item type="boardgame" id="478">'
"<thumbnail>https://cf.example/citadels.jpg</thumbnail>"
'<name type="primary" value="Citadels"/></item></items>'
)
thumbs = load_thumbnails(cache)
assert thumbs == {478: "https://cf.example/citadels.jpg"}
def test_catalog_and_unresolved_backlog_are_visible(tmp_path):
"""Extracted-but-unresolved titles must not vanish from the UI, and the
done state must not claim diff-readiness while they exist."""
web, _ = make_client(tmp_path)
state = web.get("/api/state").json()
assert state["summary"]["extracted"] == 2
assert state["summary"]["unresolved"] == 1
by_title = {c["title_raw"]: c for c in state["catalog"]}
assert by_title["Fresh Off The Shelf"]["status"] == "awaiting_resolve"
assert by_title["Citadels"]["status"] == "ambiguous"
def test_merge_veto_roundtrip(tmp_path):
cfg = make_cfg(tmp_path)
rows = read_matches(cfg.matches_path)
rows.append(
_row(
title_raw="Jokin Ha...",
match_status="merged",
bgg_id="193621",
bgg_name="Joking Hazard",
source_photos="shelf.jpg",
)
)
rows[-1]["merged_into"] = "Joking Hazard"
write_matches(cfg.matches_path, rows)
web = TestClient(create_app(cfg, client=unauthorized_client(tmp_path)))
state = web.get("/api/state").json()
(merge,) = state["merges"]
assert merge["title_raw"] == "Jokin Ha..."
assert merge["merged_into"] == "Joking Hazard"
state = web.post(
"/api/veto-merge",
json={"title_raw": "Jokin Ha...", "source_photos": "shelf.jpg"},
).json()
assert state["merges"] == []
saved = {r["title_raw"]: r for r in read_matches(cfg.matches_path)}
assert saved["Jokin Ha..."]["match_status"] == "approved"
assert saved["Jokin Ha..."]["merged_into"] == ""
# veto on a non-merged row is refused
bad = web.post(
"/api/veto-merge",
json={"title_raw": "Jokin Ha...", "source_photos": "shelf.jpg"},
)
assert bad.status_code == 400
def test_static_assets_served_with_allowlist(tmp_path):
web, _ = make_client(tmp_path)
logo = web.get("/static/logo.jpg")
assert logo.status_code == 200
assert logo.headers["content-type"] == "image/jpeg"
assert web.get("/static/favicon.png").status_code == 200
assert web.get("/static/nope.js").status_code == 404
assert web.get("/static/..%2Ftemplates%2Freview.html").status_code == 404