eric/bggpipe
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
Files
main
bggpipe/tests/test_webreview.py
T
Eric Wagoner 25c1432a03 Docs catch-up + in-app Help page 
README: the Catalog page description and a new "Fixing what the model
gets wrong" section cover the curation actions (edit/split/remove) and
their durability contract. Spec: amended-as-built notes — duplicate
same-edition copies are now supported via explicit splits, review is
both a TUI and the web app, and the data model lists the three curation
stores plus the dismissed-tickets file.

New Help page in the app (nav entry on every page): the six-stage flow,
what each page is for, the curation semantics with undo instructions, a
status-chip legend rendered with the real chips, keyboard shortcuts,
and where the data lives on disk.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016jXZFSTZQKzAC8fqpWSz9g
2026-08-02 20:20:03 -04:00

778 lines
25 KiB
Python
Raw Permalink Blame History

"""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
# -- live data reload ---------------------------------------------------
def test_state_picks_up_external_matches_rewrite(tmp_path):
web, cfg = make_client(tmp_path)
assert web.get("/api/state").json()["summary"]["total"] == 3
rows = read_matches(cfg.matches_path)
rows.append(_row(title_raw="Newcomer", match_status="unmatched"))
write_matches(cfg.matches_path, rows)
fresh = web.get("/api/state").json()
assert fresh["summary"]["total"] == 4
assert any(p["title_raw"] == "Newcomer" for p in fresh["pending"])
def test_decision_lands_on_externally_added_row(tmp_path):
# A row that did not exist at server start is still reviewable.
web, cfg = make_client(tmp_path)
rows = read_matches(cfg.matches_path)
rows.append(_row(title_raw="Newcomer", match_status="unmatched"))
write_matches(cfg.matches_path, rows)
res = web.post(
"/api/decision",
json={
"title_raw": "Newcomer",
"source_photos": "shelf.jpg",
"action": "reject",
},
)
assert res.status_code == 200
saved = {r["title_raw"]: r for r in read_matches(cfg.matches_path)}
assert saved["Newcomer"]["match_status"] == "rejected"
def test_own_saves_do_not_count_as_external_changes(tmp_path):
from bggpipe.review import ReviewSession
cfg = make_cfg(tmp_path)
session = ReviewSession(cfg, client=unauthorized_client(tmp_path))
row = next(r for r in session.rows if r["title_raw"] == "Mystery")
session.decide_reject(row)
assert session.reload_if_changed() is False # my own write
write_matches(cfg.matches_path, session.rows + [_row(title_raw="X")])
assert session.reload_if_changed() is True # someone else's
assert any(r["title_raw"] == "X" for r in session.rows)
def test_session_warnings_surface_in_state(tmp_path):
web, cfg = make_client(tmp_path)
# a manual id triggers a lookup against the 401-ing client: the session
# degrades and the warning must reach the payload (the session console
# is a StringIO here — this is the only way the user ever sees it)
res = web.post(
"/api/decision",
json={
"title_raw": "Mystery",
"source_photos": "shelf.jpg",
"action": "manual",
"bgg_id": 42,
},
)
assert res.status_code == 200
warnings = res.json()["warnings"]
assert any("couldn't look up id 42" in w for w in warnings)
assert warnings == web.get("/api/state").json()["warnings"]
def test_duplicate_rows_are_individually_decidable_via_row_ix(tmp_path):
# two editions of one game in one photo: byte-identical rows. The
# ordinal must land each decision on its own row.
from bggpipe.resolve import read_matches as read_m
from bggpipe.resolve import write_matches as write_m
cfg = make_cfg(tmp_path)
dup = _row(title_raw="Twins", match_status="unmatched")
write_m(cfg.matches_path, [dict(dup), dict(dup)])
web = TestClient(create_app(cfg, client=unauthorized_client(tmp_path)))
pending = web.get("/api/state").json()["pending"]
assert [p["title_raw"] for p in pending] == ["Twins", "Twins"]
assert pending[0]["row_ix"] != pending[1]["row_ix"] # identity, not ==
second = pending[1]
web.post(
"/api/decision",
json={
"title_raw": second["title_raw"],
"source_photos": second["source_photos"],
"row_ix": second["row_ix"],
"action": "reject",
},
)
rows = read_m(cfg.matches_path)
assert [r["match_status"] for r in rows] == ["unmatched", "rejected"]
# -- catalog curation: rowless splits and title edits ---------------------
def _rowless_wizwar(cfg) -> None:
"""Add a multi-photo extracted title with NO matches row (still awaiting
resolve — the Wiz-War shape)."""
titles = json.loads(cfg.titles_path.read_text())
titles.append(
{
"title_raw": "Wiz-War",
"confidence": "high",
"publisher_hint": "Fantasy Flight Games",
"edition_hint": "",
"source_photos": ["shelf.jpg", "shelf2.jpg"],
}
)
cfg.titles_path.write_text(json.dumps(titles))
def test_rowless_multiphoto_title_is_splittable(tmp_path):
cfg = make_cfg(tmp_path)
_rowless_wizwar(cfg)
web = TestClient(create_app(cfg, client=unauthorized_client(tmp_path)))
line = next(
c
for c in web.get("/api/state").json()["catalog"]
if c["title_raw"] == "Wiz-War"
)
assert line["status"] == "awaiting_resolve"
assert line["can_split"] is True
res = web.post(
"/api/split",
json={
"title_raw": "Wiz-War",
"source_photos": "shelf.jpg;shelf2.jpg",
"row_ix": None,
},
)
assert res.status_code == 200
copies = [c for c in res.json()["catalog"] if c["title_raw"] == "Wiz-War"]
assert [c["photos"] for c in copies] == [["shelf.jpg"], ["shelf2.jpg"]]
assert all(c["can_split"] is False for c in copies)
# the decision persisted: the store holds it (photo-scoped) and
# titles.json is split
(record,) = json.loads(cfg.title_splits_path.read_text())
assert record == {"title": "Wiz-War", "photos": ["shelf.jpg", "shelf2.jpg"]}
per_photo = [
e["source_photos"]
for e in json.loads(cfg.titles_path.read_text())
if e["title_raw"] == "Wiz-War"
]
assert per_photo == [["shelf.jpg"], ["shelf2.jpg"]]
def test_split_still_requires_multiple_photos(tmp_path):
web, _ = make_client(tmp_path)
res = web.post(
"/api/split",
json={
"title_raw": "Fresh Off The Shelf",
"source_photos": "shelf.jpg",
"row_ix": None,
},
)
assert res.status_code == 400
def test_edit_title_corrects_read_and_requeues_row(tmp_path):
web, cfg = make_client(tmp_path)
# Citadels has an ambiguous matches row; correcting its read must drop
# the stale row (it was searched with the old data) and store the fix
res = web.post(
"/api/edit-title",
json={
"title_raw": "Citadels",
"source_photos": "shelf.jpg",
"title_new": "Citadels: Dark City",
"edition": "2nd Edition",
"publisher": "",
"year": "2004",
},
)
assert res.status_code == 200
state = res.json()
titles = [c["title_raw"] for c in state["catalog"]]
assert "Citadels: Dark City" in titles and "Citadels" not in titles
corrected = next(
e
for e in json.loads(cfg.titles_path.read_text())
if e["title_raw"] == "Citadels: Dark City"
)
assert corrected["year_hint"] == 2004
assert corrected["edition_hint"] == "2nd Edition"
assert corrected["publisher_hint"] == "" # empty string cleared the cue
assert not corrected.get("language_hint") # untouched (was unset)
assert not any(r["title_raw"] == "Citadels" for r in read_matches(cfg.matches_path))
(record,) = json.loads(cfg.title_edits_path.read_text())
assert record["match"] == "Citadels"
assert record["publisher_hint"] == ""
def test_edit_title_rejects_empty_and_noop(tmp_path):
web, _ = make_client(tmp_path)
assert (
web.post(
"/api/edit-title",
json={
"title_raw": "Citadels",
"source_photos": "shelf.jpg",
"title_new": " ",
},
).status_code
== 400
)
assert (
web.post(
"/api/edit-title",
json={"title_raw": "Citadels", "source_photos": "shelf.jpg"},
).status_code
== 400
)
assert (
web.post(
"/api/edit-title",
json={
"title_raw": "No Such Game",
"source_photos": "shelf.jpg",
"title_new": "X",
},
).status_code
== 404
)
def test_edit_preserves_vetoed_rows_and_renames_them(tmp_path):
cfg = make_cfg(tmp_path)
rows = read_matches(cfg.matches_path)
rows.append(
_row(
title_raw="Citadels",
match_status="approved",
bgg_id="478",
source_photos="shelf2.jpg",
dedupe_veto="1",
)
)
write_matches(cfg.matches_path, rows)
web = TestClient(create_app(cfg, client=unauthorized_client(tmp_path)))
res = web.post(
"/api/edit-title",
json={
"title_raw": "Citadels",
"source_photos": "shelf.jpg",
"title_new": "Citadels (2016)",
},
)
assert res.status_code == 200
after = read_matches(cfg.matches_path)
# the human-vetoed row survives the cull, renamed to follow the entry;
# the unvetoed ambiguous row is requeued (dropped)
veto = [r for r in after if r.get("dedupe_veto")]
assert len(veto) == 1
assert veto[0]["title_raw"] == "Citadels (2016)"
assert veto[0]["match_status"] == "approved"
assert not any(
r["title_raw"] == "Citadels" and not r.get("dedupe_veto") for r in after
)
def test_drop_rows_photo_narrowing_spares_other_copy(tmp_path):
import io as _io
from rich.console import Console
from bggpipe.review import ReviewSession
cfg = make_cfg(tmp_path)
write_matches(
cfg.matches_path,
[
_row(title_raw="Wiz-War", match_status="auto", source_photos="a.jpg"),
_row(title_raw="Wiz-War", match_status="auto", source_photos="b.jpg"),
],
)
session = ReviewSession(
cfg,
console=Console(file=_io.StringIO()),
input_fn=lambda prompt: "",
client=unauthorized_client(tmp_path),
)
assert session.drop_rows("Wiz-War", ["a.jpg"]) == 1
(survivor,) = read_matches(cfg.matches_path)
assert survivor["source_photos"] == "b.jpg"
def test_split_and_edit_refuse_while_pipeline_rewrites(tmp_path):
import threading
from bggpipe.jobs import JobRunner
cfg = make_cfg(tmp_path)
_rowless_wizwar(cfg)
release = threading.Event()
started = threading.Event()
def blocking_extract():
started.set()
release.wait(timeout=5)
jobs = JobRunner()
web = TestClient(
create_app(
cfg,
client=unauthorized_client(tmp_path),
stages={"extract": blocking_extract},
jobs=jobs,
)
)
assert web.post("/api/run/extract").status_code == 200
assert started.wait(timeout=5)
try:
split = web.post(
"/api/split",
json={
"title_raw": "Wiz-War",
"source_photos": "shelf.jpg;shelf2.jpg",
"row_ix": None,
},
)
edit = web.post(
"/api/edit-title",
json={
"title_raw": "Wiz-War",
"source_photos": "shelf.jpg;shelf2.jpg",
"title_new": "Wiz-War!",
},
)
finally:
release.set()
assert split.status_code == 409
assert edit.status_code == 409
# neither curation store was written under the in-flight rewrite
assert not cfg.title_splits_path.exists()
assert not cfg.title_edits_path.exists()
def test_renaming_a_split_copy_keeps_split_protection(tmp_path):
cfg = make_cfg(tmp_path)
_rowless_wizwar(cfg)
web = TestClient(create_app(cfg, client=unauthorized_client(tmp_path)))
assert (
web.post(
"/api/split",
json={
"title_raw": "Wiz-War",
"source_photos": "shelf.jpg;shelf2.jpg",
"row_ix": None,
},
).status_code
== 200
)
# rename BOTH copies to the same corrected title, one at a time — the
# store must keep protecting them or the rebuild re-merges the copies
for photo in ("shelf.jpg", "shelf2.jpg"):
assert (
web.post(
"/api/edit-title",
json={
"title_raw": "Wiz-War",
"source_photos": photo,
"title_new": "Wiz War 2000",
},
).status_code
== 200
)
per_photo = [
e["source_photos"]
for e in json.loads(cfg.titles_path.read_text())
if e["title_raw"] == "Wiz War 2000"
]
assert per_photo == [["shelf.jpg"], ["shelf2.jpg"]] # still two copies
def test_single_photo_rowless_entry_is_not_splittable(tmp_path):
web, _ = make_client(tmp_path)
line = next(
c
for c in web.get("/api/state").json()["catalog"]
if c["title_raw"] == "Fresh Off The Shelf"
)
assert line["can_split"] is False
def test_remove_title_is_durable_and_drops_vetoed_rows(tmp_path):
cfg = make_cfg(tmp_path)
rows = read_matches(cfg.matches_path)
rows.append(
_row(
title_raw="Citadels",
match_status="approved",
bgg_id="478",
source_photos="shelf.jpg",
dedupe_veto="1",
)
)
write_matches(cfg.matches_path, rows)
web = TestClient(create_app(cfg, client=unauthorized_client(tmp_path)))
res = web.post(
"/api/remove-title",
json={"title_raw": "Citadels", "source_photos": "shelf.jpg"},
)
assert res.status_code == 200
assert "Citadels" not in [c["title_raw"] for c in res.json()["catalog"]]
# removal is explicit: the veto'd row goes too (unlike edits)
assert not any(r["title_raw"] == "Citadels" for r in read_matches(cfg.matches_path))
(record,) = json.loads(cfg.title_removals_path.read_text())
assert record == {"title": "Citadels", "photos": ["shelf.jpg"]}
# durable: a fresh replay keeps it gone; the sibling title is untouched
from bggpipe.extract import replay_titles
replay_titles(cfg)
titles = {e["title_raw"] for e in json.loads(cfg.titles_path.read_text())}
assert titles == {"Fresh Off The Shelf"}
# unknown titles still 404
assert (
web.post(
"/api/remove-title",
json={"title_raw": "No Such Game", "source_photos": "x.jpg"},
).status_code
== 404
)
def test_help_page_serves_and_is_in_nav(tmp_path):
web, _ = make_client(tmp_path)
res = web.get("/help")
assert res.status_code == 200
assert "Fixing the catalog" in res.text
# every page's nav includes the Help entry
assert 'href="/help"' in web.get("/").text
Reference in New Issue View Git Blame Copy Permalink