eric/bggpipe
1
0
Fork 0
Code Issues Pull Requests Releases 2 Activity
Files
d4e619611a3863d2d1192bfa29c53654ceba7464
bggpipe/tests/test_webreview.py
T
Eric WagonerandClaude Fable 5 d4e619611a Audit round 6 (mobile + --lan): access key replaces trust-by-allowlist 
Five blind reviewers over 2936d21; 13 confirmed fixes. The design
change: the Host allowlist was CSRF armor being asked to do access
control. --lan now mints a per-run access key — the printed URLs carry
?k=..., the first visit sets a cookie — required on EVERY request,
reads included (shelf photos and pipeline state are private). That
closes DNS-rebinding read exfiltration (GETs were exempt from the old
guard), closes any-LAN-device mutations via a forged localhost Host,
and frees phones from allowlist accuracy — multi-interface machines,
DHCP renewals, and failed IP discovery no longer strand writes. A
foreign Origin is still refused even with the key.

Guard hardening: Host parsed via url.hostname (ports, IPv6 brackets,
case) instead of a manual split; refusals now echo one stderr line
(they were invisible at log_level=warning) and the LAN 403 names the
remedy; startup warns when no LAN IP could be determined instead of
printing hostname-only URLs as if verified.

Silent failures: the queue page no longer freezes blank forever when a
render throws (LAST was recorded before render; one malformed CSV cell
would blank all three ledgers and blame the network) — all three
change-detection pages record LAST only after a successful render, and
the queue null-guards source_photos.

Mobile: touch-size the review/ticket/merge buttons the finger-sized
rule lost to on specificity. Style: the meta-cell builder is one shared
metaLine() helper; the Help page no longer claims localhost-only;
dead -webkit prefix dropped; --lan help text in house style.

Tests: token gating (reads and writes, cookie handoff, foreign-Origin
refusal), the Origin-present + Host-with-port path every real browser
mutation takes (was fully uncovered), run_web_review's lan branch via
monkeypatched uvicorn, and a lan_hosts test that actually pins the
lowercase/non-empty/v4-only invariants the guard depends on.

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

981 lines
32 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_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)
wr.run_web_review(cfg, port=9999, lan=True)
out = capsys.readouterr().out
assert captured["host"] == "0.0.0.0"
assert "?k=" in out # every printed URL carries the access key
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