eric/bggpipe
1
0
Fork 0
Code Issues Pull Requests Releases 2 Activity
Files
95fd18ca5c20aa5d626829764fd2dd84826345ee
bggpipe/tests/test_webreview.py
T
Eric WagonerandClaude Fable 5 95fd18ca5c --lan startup: lead with the default-route address, demote the rest 
A multi-homed machine (VM bridges, Ethernet + Wi-Fi) has several
addresses and the server cannot know which network the phone is on —
but the OS's default route is the right answer nearly always. The
banner now prints one "on your phone" URL from the route probe, with
the other interfaces on an if-that-doesn't-answer line; when the probe
fails, the settings hint plus candidates.

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

1028 lines
34 KiB
Python
Raw 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 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
)
def test_lan_allowed_hosts_admit_network_but_not_strangers(tmp_path):
cfg = make_cfg(tmp_path)
app = create_app(
cfg,
client=unauthorized_client(tmp_path),
allowed_hosts={"192.168.1.5", "erics-mac.local"},
)
body = {"title_raw": "Citadels", "source_photos": "shelf.jpg", "confirm": True}
lan = TestClient(app, base_url="http://192.168.1.5")
assert lan.post("/api/edit-title", json=body).status_code == 200
# a hostname NOT on the allowlist (DNS rebinding shape) is still refused
stranger = TestClient(app, base_url="http://attacker.example")
assert stranger.post("/api/edit-title", json=body).status_code == 403
# and without the opt-in, the LAN host is refused too
plain = create_app(cfg, client=unauthorized_client(tmp_path))
assert (
TestClient(plain, base_url="http://192.168.1.5")
.post("/api/edit-title", json=body)
.status_code
== 403
)
def test_lan_hosts_are_lowercase_nonempty_and_v4_only(monkeypatch):
import socket as socket_mod
from bggpipe.webreview import lan_hosts
monkeypatch.setattr(socket_mod, "gethostname", lambda: "Erics-Mac.Example.COM")
monkeypatch.setattr(
socket_mod,
"getaddrinfo",
lambda *a, **k: (_ for _ in ()).throw(OSError("no dns")),
)
hosts = lan_hosts()
assert "erics-mac.example.com" in hosts
assert "erics-mac.local" in hosts
assert "" not in hosts # an empty entry would admit Host-less mutations
assert all(":" not in h for h in hosts) # no v6 forms, no ports
def test_lan_token_gates_every_request(tmp_path):
cfg = make_cfg(tmp_path)
app = create_app(cfg, client=unauthorized_client(tmp_path), lan_token="sekret")
phone = TestClient(app, base_url="http://192.168.1.99:8377")
# reads are gated too: shelf photos and state are private
assert phone.get("/api/state").status_code == 403
assert phone.get("/api/state?k=wrong").status_code == 403
first = phone.get("/titles?k=sekret")
assert first.status_code == 200
assert "bggpipe_key" in first.cookies # the URL key becomes a cookie
# cookie carries the session: mutations work from ANY host the phone
# used (no allowlist dependence — DHCP/multi-interface safe), with a
# same-origin Origin header and a port, like a real phone browser
res = phone.post(
"/api/edit-title",
json={"title_raw": "Citadels", "source_photos": "shelf.jpg", "confirm": True},
headers={"origin": "http://192.168.1.99:8377"},
)
assert res.status_code == 200
# a foreign Origin is still refused even with the key
assert (
phone.post(
"/api/dismiss",
json={"photo": "shelf.jpg"},
headers={"origin": "http://evil.example"},
).status_code
== 403
)
def test_lan_loopback_stays_keyless_but_guarded(tmp_path):
cfg = make_cfg(tmp_path)
app = create_app(cfg, client=unauthorized_client(tmp_path), lan_token="sekret")
local = TestClient(
app, base_url="http://127.0.0.1:8377", client=("127.0.0.1", 50000)
)
# the desktop browser (old tabs, the auto-opened one) needs no key
assert local.get("/api/pipeline").status_code == 200
assert (
local.post(
"/api/edit-title",
json={
"title_raw": "Citadels",
"source_photos": "shelf.jpg",
"confirm": True,
},
headers={"origin": "http://127.0.0.1:8377"},
).status_code
== 200
)
# but loopback keylessness never extends to foreign Hosts (rebinding)
# or foreign Origins (classic CSRF from a local browser)
rebound = TestClient(
app, base_url="http://evil.example:8377", client=("127.0.0.1", 50000)
)
assert rebound.get("/api/state").status_code == 403
assert (
local.post(
"/api/dismiss",
json={"photo": "shelf.jpg"},
headers={"origin": "http://evil.example"},
).status_code
== 403
)
def test_localhost_mutations_pass_with_origin_and_port(tmp_path):
# the path every real browser takes: Origin present + port in Host —
# regressing the header parsing must fail loudly here
cfg = make_cfg(tmp_path)
app = create_app(cfg, client=unauthorized_client(tmp_path))
web = TestClient(app, base_url="http://localhost:8377")
res = web.post(
"/api/edit-title",
json={"title_raw": "Citadels", "source_photos": "shelf.jpg", "confirm": True},
headers={"origin": "http://localhost:8377"},
)
assert res.status_code == 200
assert (
TestClient(app, base_url="http://attacker.example:8377")
.post(
"/api/edit-title",
json={"title_raw": "Citadels", "source_photos": "shelf.jpg"},
)
.status_code
== 403
)
def test_run_web_review_lan_branch_binds_and_warns(tmp_path, monkeypatch, capsys):
import bggpipe.webreview as wr
captured = {}
monkeypatch.setattr("uvicorn.run", lambda app, **kw: captured.update(kw, app=app))
cfg = make_cfg(tmp_path)
monkeypatch.setattr(wr, "_route_ip", lambda: "192.168.1.5")
monkeypatch.setattr(
wr,
"lan_hosts",
lambda: {"127.0.0.1", "192.168.1.5", "192.168.64.1", "erics-mac.local"},
)
wr.run_web_review(cfg, port=9999, lan=True)
out = capsys.readouterr().out
assert captured["host"] == "0.0.0.0"
# the default-route address leads; other interfaces are fallbacks
assert "on your phone, open: http://192.168.1.5:9999/?k=" in out
fallback = next(line for line in out.splitlines() if "try:" in line)
assert "192.168.64.1" in fallback and "erics-mac.local" in fallback
# no phone can reach loopback: only the desktop line mentions it
assert out.count("http://127.0.0.1:9999") == 1
assert "Use only on a network you trust" in out
wr.run_web_review(cfg, port=9999, lan=False)
assert captured["host"] == "127.0.0.1"
Reference in New Issue View Git Blame Copy Permalink