Files
bggpipe/tests/test_webreview.py
Eric Wagoner 1d700db012 Shaky reads become visible and clearable on the Titles page
The badge counted lines nothing on the page identified. Each shaky line
(vision confidence below high, nothing verified yet — the badge's exact
predicate, served per-line so page and badge can't disagree) now shows
a dashed gold "shaky read" chip, a filterbar toggle shows just them,
and a "✓ looks right" button confirms a correct read in one click —
recorded as a confidence-high edit record, so verification is as
durable as any other curation and, unlike a real edit, re-queues
nothing. Saving an unchanged edit form on a shaky line counts as
confirming it. Help's chip legend explains the chip and its three ways
to clear (confirm, edit, or a BGG match).

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

872 lines
28 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
# -- 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 titles" in res.text
# every page's nav includes the Help entry
assert 'href="/help"' in web.get("/").text
def test_shaky_reads_badge_counts_and_drains_on_edit(tmp_path):
cfg = make_cfg(tmp_path)
titles = json.loads(cfg.titles_path.read_text())
titles.append(
{
"title_raw": "Hebarceos",
"confidence": "low",
"source_photos": ["shelf.jpg"],
}
)
cfg.titles_path.write_text(json.dumps(titles))
web = TestClient(create_app(cfg, client=unauthorized_client(tmp_path)))
assert web.get("/api/pipeline").json()["shaky_reads"] == 1
# an edit means a human read the line: it is verified, the badge drains
assert (
web.post(
"/api/edit-title",
json={
"title_raw": "Hebarceos",
"source_photos": "shelf.jpg",
"title_new": "Herbaceous",
},
).status_code
== 200
)
assert web.get("/api/pipeline").json()["shaky_reads"] == 0
def test_nav_order_matches_workflow(tmp_path):
web, _ = make_client(tmp_path)
html = web.get("/").text
order = [
html.index(f'href="{href}"')
for href in (
"/",
"/photos",
"/titles",
"/review",
"/queue",
"/library",
"/help",
)
]
assert order == sorted(order)
# the old address still lands on the page
assert web.get("/catalog", follow_redirects=False).headers["location"] == "/titles"
def test_confirm_marks_shaky_read_verified_without_requeue(tmp_path):
cfg = make_cfg(tmp_path)
titles = json.loads(cfg.titles_path.read_text())
titles.append(
{
"title_raw": "Blurry Spine",
"confidence": "low",
"source_photos": ["shelf.jpg"],
}
)
cfg.titles_path.write_text(json.dumps(titles))
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"] == "Blurry Spine"
)
assert line["shaky"] is True
rows_before = read_matches(cfg.matches_path)
res = web.post(
"/api/edit-title",
json={
"title_raw": "Blurry Spine",
"source_photos": "shelf.jpg",
"confirm": True,
},
)
assert res.status_code == 200
line = next(c for c in res.json()["catalog"] if c["title_raw"] == "Blurry Spine")
assert line["shaky"] is False # verified: chip and badge both clear
assert web.get("/api/pipeline").json()["shaky_reads"] == 0
# confirm changed no data, so nothing was re-queued
assert read_matches(cfg.matches_path) == rows_before
(record,) = json.loads(cfg.title_edits_path.read_text())
assert record["confidence"] == "high"
# a change-free save without confirm is still refused
assert (
web.post(
"/api/edit-title",
json={"title_raw": "Blurry Spine", "source_photos": "shelf.jpg"},
).status_code
== 400
)