eric/bggpipe
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
Files
86d434a400ade06f0e9c392fb3d11183840cf5f6
bggpipe/tests/test_review.py
T
Eric Wagoner 851b34e369 Split into copies: the symmetric partner of the merge veto 
Three identical boxes in three photos are indistinguishable from one
box photographed three times, so extract's dedupe folds them into one
entry — correct for overlapping shots, wrong for a shelf holding three
editions of a favorite game. The catalog now offers "split into copies"
on multi-photo rows: the row explodes into one row per photo, each
dedupe_veto-flagged so no future resolve re-merges them, each keeping
its match but reopening its own edition slot (candidates preserved when
present). Resolve's provenance-follow skips split rows (their photo
sets are human-authored), the catalog renders surplus split copies as
their own lines with a "copy" chip, and diff's vetoed-duplicate logic
turns them into the extra collection entries they are.

Applied to the real data: Wiz-War is now three copies across IMG_4502/
4504/4528 — one claims the owned collection entry, two queue as new
second-copy adds.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-08-02 19:10:26 -04:00

539 lines
17 KiB
Python
Raw Blame History

"""Review-TUI tests against a synthetic matches.csv, driven by scripted
input. BGG responses replay from the fixture cache; nothing hits the
network (the transport raises if it would)."""
from __future__ import annotations
import io
import json
from pathlib import Path
import httpx
import pytest
from rich.console import Console
from bggpipe.bgg_client import BGGClient
from bggpipe.config import Config
from bggpipe.resolve import read_matches, write_matches
from bggpipe.review import ReviewSession, run_review
from bggpipe.webreview import DismissStore
FIXTURES = Path(__file__).parent / "fixtures" / "bgg_cache"
def _no_network(request: httpx.Request) -> httpx.Response:
raise AssertionError(f"test hit the network: {request.url}")
def fixture_client() -> BGGClient:
return BGGClient(cache_dir=FIXTURES, transport=httpx.MockTransport(_no_network))
def unauthorized_client(tmp_path) -> BGGClient:
"""Client whose every request 401s — the no-token-yet world."""
transport = httpx.MockTransport(
lambda req: httpx.Response(401, text="Unauthorized")
)
return BGGClient(cache_dir=tmp_path / "empty_cache", transport=transport)
def scripted(*answers):
it = iter(answers)
def input_fn(prompt: str) -> str:
try:
return next(it)
except StopIteration:
raise EOFError from None
return input_fn
def quiet_console() -> Console:
return Console(file=io.StringIO(), width=100)
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": "hand-typed-test-list",
}
row.update(overrides)
return row
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,
},
]
)
WINGSPAN_VERSION_CANDIDATES = json.dumps(
[
{
"version_id": 465063,
"name": "English first edition",
"year": 2019,
"publishers": ["Stonemaier Games"],
"languages": ["English"],
"score": 5,
},
{
"version_id": 521212,
"name": "English fourth printing",
"year": 2020,
"publishers": ["Stonemaier Games"],
"languages": ["English"],
"score": 5,
},
]
)
def _cfg(tmp_path) -> Config:
return Config(data_dir=tmp_path / "data")
def _setup(tmp_path, rows) -> Config:
cfg = _cfg(tmp_path)
write_matches(cfg.matches_path, rows)
return cfg
def test_ambiguous_pick_approves_candidate(tmp_path):
cfg = _setup(
tmp_path,
[
_row(
title_raw="Citadels",
match_status="ambiguous",
candidates_json=CITADELS_CANDIDATES,
)
],
)
run_review(
cfg, console=quiet_console(), input_fn=scripted("2"), client=fixture_client()
)
(row,) = read_matches(cfg.matches_path)
assert row["match_status"] == "approved"
assert row["bgg_id"] == "205398"
assert row["year"] == "2016"
def test_reject_and_skip(tmp_path):
cfg = _setup(
tmp_path,
[
_row(title_raw="Blorvath", match_status="unmatched"),
_row(
title_raw="Citadels",
match_status="ambiguous",
candidates_json=CITADELS_CANDIDATES,
),
],
)
run_review(
cfg,
console=quiet_console(),
input_fn=scripted("r", "s"),
client=fixture_client(),
)
rows = {r["title_raw"]: r for r in read_matches(cfg.matches_path)}
assert rows["Blorvath"]["match_status"] == "rejected"
assert rows["Citadels"]["match_status"] == "ambiguous" # skipped, still pending
def test_unmatched_manual_id_degrades_without_token(tmp_path):
cfg = _setup(tmp_path, [_row(title_raw="Some Rare Game", match_status="unmatched")])
run_review(
cfg,
console=quiet_console(),
input_fn=scripted("m 99999"),
client=unauthorized_client(tmp_path),
)
(row,) = read_matches(cfg.matches_path)
assert row["match_status"] == "approved"
assert row["bgg_id"] == "99999"
assert row["bgg_name"] == "" # lookup blocked; id recorded anyway
def test_unmatched_research_from_fixture_cache_with_version(tmp_path):
cfg = _setup(tmp_path, [_row(title_raw="Wingspan", match_status="unmatched")])
cfg.titles_path.parent.mkdir(parents=True, exist_ok=True)
cfg.titles_path.write_text(
json.dumps(
[
{
"title_raw": "Wingspan",
"publisher_hint": "Stonemaier Games",
"year_hint": 2019,
"language_hint": "English",
"source_photos": ["x.jpg"],
}
]
)
)
run_review(
cfg,
console=quiet_console(),
input_fn=scripted("f Wingspan", "1"),
client=fixture_client(),
)
(row,) = read_matches(cfg.matches_path)
assert row["match_status"] == "approved"
assert row["bgg_id"] == "266192"
# cues + fixture versions -> resolved to the 2019 Stonemaier English edition
assert row["version_status"] == "version_auto"
assert row["version_id"] == "465063"
def test_resumable_quit_midway_then_continue(tmp_path):
rows = [
_row(
title_raw="Citadels",
match_status="ambiguous",
candidates_json=CITADELS_CANDIDATES,
),
_row(title_raw="Blorvath", match_status="unmatched"),
]
cfg = _setup(tmp_path, rows)
# first sitting: one decision, then input runs out (EOF == quit)
run_review(
cfg, console=quiet_console(), input_fn=scripted("1"), client=fixture_client()
)
by_title = {r["title_raw"]: r for r in read_matches(cfg.matches_path)}
assert by_title["Citadels"]["match_status"] == "approved"
assert by_title["Blorvath"]["match_status"] == "unmatched" # untouched
# second sitting resumes exactly where we left off
run_review(
cfg, console=quiet_console(), input_fn=scripted("r"), client=fixture_client()
)
by_title = {r["title_raw"]: r for r in read_matches(cfg.matches_path)}
assert by_title["Blorvath"]["match_status"] == "rejected"
def test_version_pass_pick_and_unknown(tmp_path):
cfg = _setup(
tmp_path,
[
_row(
title_raw="Wingspan",
match_status="auto",
bgg_id="266192",
version_status="version_ambiguous",
version_candidates_json=WINGSPAN_VERSION_CANDIDATES,
),
_row(
title_raw="Catan",
match_status="auto",
bgg_id="13",
version_status="version_ambiguous",
version_candidates_json=WINGSPAN_VERSION_CANDIDATES,
),
],
)
run_review(
cfg,
console=quiet_console(),
input_fn=scripted("y", "1", "u"),
client=fixture_client(),
)
by_title = {r["title_raw"]: r for r in read_matches(cfg.matches_path)}
assert by_title["Wingspan"]["version_status"] == "version_approved"
assert by_title["Wingspan"]["version_id"] == "465063"
assert by_title["Catan"]["version_status"] == "version_unknown"
assert by_title["Catan"]["version_id"] == ""
def test_version_pass_is_skippable(tmp_path):
cfg = _setup(
tmp_path,
[
_row(
title_raw="Wingspan",
match_status="auto",
bgg_id="266192",
version_status="version_ambiguous",
version_candidates_json=WINGSPAN_VERSION_CANDIDATES,
)
],
)
run_review(
cfg, console=quiet_console(), input_fn=scripted("n"), client=fixture_client()
)
(row,) = read_matches(cfg.matches_path)
assert row["version_status"] == "version_ambiguous" # untouched, review later
# -- concurrent-rewrite and degradation safety --------------------------
def test_veto_merge_persists_against_future_dedupe(tmp_path):
cfg = _setup(
tmp_path,
[
_row(title_raw="CATAN", match_status="merged", merged_into="Catan"),
_row(title_raw="Catan", match_status="auto", bgg_id="13"),
],
)
session = ReviewSession(
cfg,
console=quiet_console(),
input_fn=scripted(),
client=unauthorized_client(tmp_path),
)
merged = next(r for r in session.rows if r["match_status"] == "merged")
session.veto_merge(merged)
saved = {r["title_raw"]: r for r in read_matches(cfg.matches_path)}
assert saved["CATAN"]["match_status"] == "approved"
assert saved["CATAN"]["dedupe_veto"] == "1" # survives resolve re-runs
def test_failed_save_never_leaves_memory_ahead_of_disk(tmp_path, monkeypatch):
cfg = _setup(tmp_path, [_row(title_raw="Mystery", match_status="unmatched")])
session = ReviewSession(
cfg,
console=quiet_console(),
input_fn=scripted(),
client=unauthorized_client(tmp_path),
)
row = session.rows[0]
def exploding_write(path, rows):
raise OSError("disk full")
monkeypatch.setattr("bggpipe.review.write_matches", exploding_write)
with pytest.raises(OSError):
session.decide_reject(row)
# memory was rolled back to what disk actually holds
assert session.rows[0]["match_status"] == "unmatched"
assert session.decisions == 0
def test_save_merges_own_decision_over_concurrent_external_rewrite(tmp_path):
cfg = _setup(tmp_path, [_row(title_raw="Mystery", match_status="unmatched")])
session = ReviewSession(
cfg,
console=quiet_console(),
input_fn=scripted(),
client=unauthorized_client(tmp_path),
)
row = session.rows[0]
# resolve appends a new row in another terminal AFTER our session loaded
external = read_matches(cfg.matches_path)
external.append(_row(title_raw="Newcomer", match_status="auto", bgg_id="7"))
write_matches(cfg.matches_path, external)
session.decide_reject(row)
saved = {r["title_raw"]: r for r in read_matches(cfg.matches_path)}
assert saved["Mystery"]["match_status"] == "rejected" # our decision
assert "Newcomer" in saved # their row survived too
def test_manual_id_unknown_to_bgg_warns_instead_of_crashing(tmp_path):
empty_things = BGGClient(
cache_dir=tmp_path / "cache",
transport=httpx.MockTransport(
lambda req: httpx.Response(200, text='<items total="0"></items>')
),
sleep=lambda s: None,
)
cfg = _setup(tmp_path, [_row(title_raw="Mystery", match_status="unmatched")])
session = ReviewSession(
cfg, console=quiet_console(), input_fn=scripted(), client=empty_things
)
session.decide_manual(session.rows[0], 999999)
assert session.rows[0]["match_status"] == "approved"
assert session.rows[0]["bgg_id"] == "999999"
assert any("no game with id 999999" in w for w in session.warnings)
def test_every_tui_decision_after_external_rewrite_is_saved(tmp_path):
# run() iterates row references snapshotted before any reload; every
# decision must be re-adopted into the current list or it would be
# counted but never written.
cfg = _setup(
tmp_path,
[
_row(title_raw="Alpha", match_status="unmatched"),
_row(title_raw="Beta", match_status="unmatched"),
_row(title_raw="Gamma", match_status="unmatched"),
],
)
session = ReviewSession(
cfg,
console=quiet_console(),
input_fn=scripted(),
client=unauthorized_client(tmp_path),
)
stale_refs = list(session.pending_rows()) # what run() iterates
external = read_matches(cfg.matches_path)
external.append(_row(title_raw="Newcomer", match_status="auto", bgg_id="7"))
write_matches(cfg.matches_path, external)
for ref in stale_refs: # decision 1 reloads; 2 and 3 are orphaned refs
session.decide_reject(ref)
saved = {r["title_raw"]: r for r in read_matches(cfg.matches_path)}
assert [saved[t]["match_status"] for t in ("Alpha", "Beta", "Gamma")] == (
["rejected"] * 3
)
assert "Newcomer" in saved # the external row survived too
assert session.decisions == 3
def test_fill_version_uses_the_approved_rows_own_cues(tmp_path):
# two same-title entries are two editions: the version lookup must use
# the row's own photo-keyed cues, never a title-keyed last-wins dict
entry_good = {
"title_raw": "Wingspan",
"publisher_hint": "Stonemaier", # matches the fixture's version
"year_hint": 2019,
"source_photos": ["good.jpg"],
}
entry_bad = {
"title_raw": "Wingspan",
"publisher_hint": "Nobody Media", # matches nothing
"edition_hint": "Imaginary edition",
"source_photos": ["bad.jpg"],
}
cfg = _setup(
tmp_path,
[
_row(
title_raw="Wingspan",
match_status="ambiguous",
source_photos="good.jpg",
candidates_json=json.dumps(
[{"bgg_id": 266192, "name": "Wingspan", "year": 2019}]
),
)
],
)
(cfg.data_dir / "titles.json").write_text(json.dumps([entry_good, entry_bad]))
session = ReviewSession(
cfg, console=quiet_console(), input_fn=scripted(), client=fixture_client()
)
row = session.rows[0]
session.decide_pick(row, {"bgg_id": 266192, "name": "Wingspan", "year": 2019})
# the row's own photo (good.jpg) must select entry_good's cues
assert row["version_status"] == "version_auto"
assert row["version_id"] == "465063"
def test_dismiss_failure_keeps_ticket_visible(tmp_path, monkeypatch):
store = DismissStore(tmp_path / "dismissed.json")
def exploding(path, text):
raise OSError("disk full")
monkeypatch.setattr("bggpipe.webreview.atomic_write_text", exploding)
with pytest.raises(OSError):
store.add("photo|loc|txt|art")
assert store.keys == set() # memory never claims what disk doesn't hold
def test_corrupt_dismiss_file_is_quarantined_not_fatal(tmp_path):
path = tmp_path / "dismissed.json"
path.write_text('["torn')
with pytest.warns(UserWarning, match="unreadable"):
store = DismissStore(path)
assert store.keys == set()
assert (tmp_path / "dismissed.json.corrupt").exists()
def test_split_row_makes_per_photo_vetoed_copies(tmp_path):
from bggpipe.resolve import read_matches
cfg = _setup(
tmp_path,
[
_row(
title_raw="Wiz-War",
match_status="approved",
bgg_id="104710",
bgg_name="Wiz-War",
source_photos="a.jpg;b.jpg;c.jpg",
version_candidates_json='[{"version_id": 1, "name": "8th"}]',
)
],
)
session = ReviewSession(
cfg,
console=quiet_console(),
input_fn=scripted(),
client=unauthorized_client(tmp_path),
)
copies = session.split_row(session.rows[0])
saved = read_matches(cfg.matches_path)
assert [r["source_photos"] for r in saved] == ["a.jpg", "b.jpg", "c.jpg"]
assert all(r["dedupe_veto"] == "1" for r in saved)
assert all(r["bgg_id"] == "104710" for r in saved) # match survives
# each copy picks its own edition: candidates kept, status reopened
assert all(r["version_status"] == "version_ambiguous" for r in saved)
assert all(r["version_id"] == "" for r in saved)
assert len(copies) == 3
def test_split_copies_survive_resolve_rerun(tmp_path):
import json as _json
from bggpipe.resolve import read_matches, run_resolve
cfg = _setup(
tmp_path,
[
_row(
title_raw="Wiz-War",
match_status="approved",
bgg_id="104710",
source_photos="a.jpg;b.jpg;c.jpg",
)
],
)
(cfg.data_dir / "titles.json").write_text(
_json.dumps(
[{"title_raw": "Wiz-War", "source_photos": ["a.jpg", "b.jpg", "c.jpg"]}]
)
)
session = ReviewSession(
cfg,
console=quiet_console(),
input_fn=scripted(),
client=unauthorized_client(tmp_path),
)
session.split_row(session.rows[0])
run_resolve(cfg, client=fixture_client())
saved = read_matches(cfg.matches_path)
assert len(saved) == 3 # not re-merged, not re-resolved
assert [r["source_photos"] for r in saved] == ["a.jpg", "b.jpg", "c.jpg"]
Reference in New Issue View Git Blame Copy Permalink