Files
bggpipe/tests/test_upload.py
T
Eric WagonerandClaude Fable 5 6f2f2dac53 Row clicks survive whitespace; the failure badge stops inflating
Sleeping Gods failed where three siblings passed: the second-pass row
click matched captured text with Playwright's has_text regex, which
tests raw textContent — tabs and newlines included — against a capture
that was whitespace-normalized. Rows whose markup happened to be tidy
matched; that one didn't. The picker now re-finds the row by NORMALIZED
text and clicks it by index, which also survives the list re-rendering
in a different order between openings.

And the UI's failure count only ever grew: upload_log.csv is an
append-only audit trail, so a retry that succeeds leaves its old
'failed' line in place. outstanding_failures() counts the LAST status
per job key — the same rule _plan_jobs already uses to decide what to
skip — so a landed retry clears the badge. On Eric's log: 6 'failed'
rows, 1 job actually outstanding.

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

734 lines
24 KiB
Python

"""Upload-stage tests: queue building, idempotency via upload_log.csv, the
stub-fixture guard, per-game failure isolation, pacing, and verify — all
against a fake uploader. No browser, no network."""
from __future__ import annotations
import csv
import random
from pathlib import Path
import pytest
import typer
from bggpipe.config import Config
from bggpipe.models import CollectionItem
from bggpipe.upload import (
UPLOAD_LOG_COLUMNS,
LoginError,
UploadJob,
_scrub,
build_queue,
run_upload,
verify_uploads,
)
def _now() -> str:
return "2026-08-01T00:00:00+00:00"
def _cfg(tmp_path: Path) -> Config:
return Config(bgg_username="tester", data_dir=tmp_path)
def _write_csv(path: Path, columns: list[str], rows: list[dict]) -> None:
with path.open("w", newline="") as f:
writer = csv.DictWriter(f, fieldnames=columns)
writer.writeheader()
writer.writerows(rows)
def _add_row(bgg_id="1", name="Wingspan", version_id="", version_name=""):
return {
"bgg_id": bgg_id,
"bgg_name": name,
"year": "2019",
"type": "boardgame",
"version_id": version_id,
"version_name": version_name,
"title_raw": name.upper(),
"source_photos": "x.jpg",
}
def _update_row(collid="9", bgg_id="2", name="Britannia", vid="25", vname="AH ed."):
return {
"collid": collid,
"bgg_id": bgg_id,
"bgg_name": name,
"version_id": vid,
"version_name": vname,
}
def _log_row(action="add", bgg_id="1", collid="", version_id="", status="added"):
return {
"action": action,
"bgg_id": bgg_id,
"collid": collid,
"name": "Game",
"version_id": version_id,
"status": status,
"timestamp": _now(),
"error": "",
}
def _seed_data(tmp_path: Path, to_add=None, to_update=None, log=None) -> None:
_write_csv(
tmp_path / "to_add.csv",
list(_add_row().keys()),
to_add if to_add is not None else [],
)
_write_csv(
tmp_path / "to_update.csv",
list(_update_row().keys()),
to_update if to_update is not None else [],
)
if log is not None:
_write_csv(tmp_path / "upload_log.csv", UPLOAD_LOG_COLUMNS, log)
class FakeUploader:
"""Records jobs; raises for names listed in `failures`."""
def __init__(self, failures: set[str] | None = None):
self.calls: list[UploadJob] = []
self.failures = failures or set()
def _handle(self, job: UploadJob, status: str) -> tuple[str, str]:
self.calls.append(job)
if job.name in self.failures:
raise RuntimeError("dialog never appeared")
return status, ""
def add_game(self, job: UploadJob) -> tuple[str, str]:
return self._handle(job, "added")
def update_entry(self, job: UploadJob) -> tuple[str, str]:
return self._handle(job, "updated")
# -- build_queue --------------------------------------------------------
def test_build_queue_skips_logged_successes():
jobs, done, failed, _ = build_queue(
[_add_row(bgg_id="1"), _add_row(bgg_id="2", name="Catan")],
[_update_row(collid="9")],
[
_log_row(action="add", bgg_id="1", status="added"),
_log_row(action="update", bgg_id="2", collid="9", status="updated"),
],
)
assert [j.bgg_id for j in jobs] == ["2"]
assert done == 2
assert failed == 0
def test_build_queue_second_copy_is_a_distinct_job():
# Same game, different version: a separate physical copy, so a
# logged add of one version must not swallow the other.
jobs, done, _, _ = build_queue(
[
_add_row(bgg_id="1", version_id="10", version_name="First ed."),
_add_row(bgg_id="1", version_id="11", version_name="Second ed."),
],
[],
[_log_row(action="add", bgg_id="1", version_id="10", status="added")],
)
assert [j.version_id for j in jobs] == ["11"]
assert done == 1
def test_build_queue_failures_need_retry_flag():
log = [_log_row(action="add", bgg_id="1", status="failed")]
jobs, _, skipped, _ = build_queue([_add_row(bgg_id="1")], [], log)
assert jobs == [] and skipped == 1
jobs, _, skipped, _ = build_queue(
[_add_row(bgg_id="1")], [], log, retry_failed=True
)
assert len(jobs) == 1 and skipped == 0
def test_build_queue_latest_log_entry_wins():
# failed then added on retry -> done, not retriable
log = [
_log_row(action="add", bgg_id="1", status="failed"),
_log_row(action="add", bgg_id="1", status="added"),
]
jobs, done, _, _ = build_queue([_add_row(bgg_id="1")], [], log, retry_failed=True)
assert jobs == [] and done == 1
# -- stub-fixture guard -------------------------------------------------
def test_real_run_refuses_while_stub_marker_exists(tmp_path):
cfg = _cfg(tmp_path)
cfg.cache_dir.mkdir(parents=True)
(cfg.cache_dir / "STUB_FIXTURES.marker").write_text("stub")
_seed_data(tmp_path, to_add=[_add_row()])
with pytest.raises(typer.Exit):
run_upload(cfg, uploader=FakeUploader(), sleep=lambda s: None, now=_now)
def test_dry_run_allowed_with_stub_marker_and_writes_nothing(tmp_path, capsys):
cfg = _cfg(tmp_path)
cfg.cache_dir.mkdir(parents=True)
(cfg.cache_dir / "STUB_FIXTURES.marker").write_text("stub")
_seed_data(tmp_path, to_add=[_add_row()], to_update=[_update_row()])
run_upload(cfg, dry_run=True, now=_now)
out = capsys.readouterr().out
assert "WARNING" in out and "SYNTHETIC" in out
assert "would add Wingspan" in out
assert "collid 9" in out
assert not (tmp_path / "upload_log.csv").exists()
# -- run_upload with a fake browser -------------------------------------
def test_run_logs_every_attempt_and_continues_past_failures(tmp_path):
cfg = _cfg(tmp_path)
_seed_data(
tmp_path,
to_add=[_add_row(bgg_id="1"), _add_row(bgg_id="2", name="Catan")],
to_update=[_update_row(collid="9")],
)
fake = FakeUploader(failures={"Catan"})
results = run_upload(cfg, uploader=fake, sleep=lambda s: None, now=_now)
assert [r["status"] for r in results] == ["added", "failed", "updated"]
logged = list(csv.DictReader((tmp_path / "upload_log.csv").open()))
assert len(logged) == 3
assert logged[1]["error"] == "RuntimeError: dialog never appeared"
assert logged[2]["collid"] == "9"
def test_rerun_skips_completed_work(tmp_path):
cfg = _cfg(tmp_path)
_seed_data(tmp_path, to_add=[_add_row(bgg_id="1"), _add_row(bgg_id="2", name="C")])
fake = FakeUploader()
run_upload(cfg, uploader=fake, sleep=lambda s: None, now=_now)
assert len(fake.calls) == 2
again = FakeUploader()
results = run_upload(cfg, uploader=again, sleep=lambda s: None, now=_now)
assert again.calls == [] and results == []
def test_pacing_sleeps_2_to_4s_between_games_only(tmp_path):
cfg = _cfg(tmp_path)
_seed_data(tmp_path, to_add=[_add_row(bgg_id=str(i)) for i in range(1, 5)])
sleeps: list[float] = []
run_upload(
cfg,
uploader=FakeUploader(),
sleep=sleeps.append,
rng=random.Random(42),
now=_now,
)
assert len(sleeps) == 3 # between games, not before the first
assert all(2.0 <= s <= 4.0 for s in sleeps)
def test_limit_caps_the_queue(tmp_path):
cfg = _cfg(tmp_path)
_seed_data(tmp_path, to_add=[_add_row(bgg_id=str(i)) for i in range(1, 5)])
fake = FakeUploader()
run_upload(cfg, uploader=fake, limit=2, sleep=lambda s: None, now=_now)
assert len(fake.calls) == 2
# -- credential hygiene -------------------------------------------------
def test_scrub_removes_credentials_from_error_text(monkeypatch):
monkeypatch.setenv("BGG_USERNAME", "erics-user")
monkeypatch.setenv("BGG_PASSWORD", "s3cret-pw")
text = 'fill("erics-user") then fill("s3cret-pw") timed out'
assert "s3cret-pw" not in _scrub(text)
assert "erics-user" not in _scrub(text)
# -- verify -------------------------------------------------------------
def _item(object_id, coll_id, name="Game", version_id=None):
return CollectionItem(
object_id=object_id,
coll_id=coll_id,
name=name,
subtype="boardgame",
own=True,
year=None,
version_id=version_id,
)
def test_verify_flags_missing_and_confirms_present():
log = [
_log_row(action="add", bgg_id="1", status="added"),
_log_row(action="add", bgg_id="2", status="added"),
_log_row(
action="update", bgg_id="3", collid="30", version_id="7", status="updated"
),
]
collection = [_item(1, 10), _item(3, 30, version_id=7)]
problems = verify_uploads(log, collection)
assert len(problems) == 1
assert "not in collection" in problems[0]
def test_verify_checks_version_on_adds_and_updates():
log = [
_log_row(action="add", bgg_id="1", version_id="99", status="added"),
_log_row(
action="update", bgg_id="3", collid="30", version_id="7", status="updated"
),
]
collection = [_item(1, 10, version_id=11), _item(3, 30, version_id=8)]
problems = verify_uploads(log, collection)
assert len(problems) == 2
def test_verify_ignores_failed_rows():
log = [_log_row(action="add", bgg_id="5", status="failed")]
assert verify_uploads(log, []) == []
def test_fresh_clone_marker_blocks_upload_without_cache_dir(tmp_path):
# A fresh clone has the committed data/STUB_DATA.marker but no
# gitignored bgg_cache/ at all — upload must still refuse.
cfg = _cfg(tmp_path)
(tmp_path / "STUB_DATA.marker").write_text("stub-derived CSVs")
_seed_data(tmp_path, to_add=[_add_row()])
with pytest.raises(typer.Exit):
run_upload(cfg, uploader=FakeUploader(), sleep=lambda s: None, now=_now)
# -- failure isolation and resume ---------------------------------------
def test_login_error_aborts_without_poisoning_the_log(tmp_path):
class BrokenLogin(FakeUploader):
def add_game(self, job):
raise LoginError("Cloudflare is challenging this browser")
cfg = _cfg(tmp_path)
_seed_data(tmp_path, to_add=[_add_row(bgg_id=str(i)) for i in range(1, 4)])
results = run_upload(cfg, uploader=BrokenLogin(), sleep=lambda s: None, now=_now)
assert results == [] # nothing logged: next run retries everything
assert not (tmp_path / "upload_log.csv").exists()
def test_three_identical_failures_abort_as_systemic(tmp_path):
cfg = _cfg(tmp_path)
_seed_data(tmp_path, to_add=[_add_row(bgg_id=str(i)) for i in range(1, 6)])
fake = FakeUploader(failures={"Wingspan"}) # every job shares the name
results = run_upload(cfg, uploader=fake, sleep=lambda s: None, now=_now)
assert len(results) == 3 # aborted after the third identical failure
logged = list(csv.DictReader((tmp_path / "upload_log.csv").open()))
assert len(logged) == 3 # jobs 4-5 left unlogged and retryable
def test_added_no_version_is_done_and_verify_tolerates_it(tmp_path):
class NoVersionPicker(FakeUploader):
def add_game(self, job):
self.calls.append(job)
return "added_no_version", "version not in picker; added without version"
cfg = _cfg(tmp_path)
_seed_data(tmp_path, to_add=[_add_row(version_id="99", version_name="4th ed.")])
run_upload(cfg, uploader=NoVersionPicker(), sleep=lambda s: None, now=_now)
# done: re-running must NOT re-add (a duplicate collection entry)
again = FakeUploader()
assert run_upload(cfg, uploader=again, sleep=lambda s: None, now=_now) == []
assert again.calls == []
# verify: game present without the version is the EXPECTED outcome
log = list(csv.DictReader((tmp_path / "upload_log.csv").open()))
assert verify_uploads(log, [_item(1, 10)]) == []
def test_missing_to_add_csv_is_a_loud_precondition_failure(tmp_path):
cfg = _cfg(tmp_path) # no diff outputs seeded at all
with pytest.raises(typer.Exit):
run_upload(cfg, uploader=FakeUploader(), sleep=lambda s: None, now=_now)
def test_second_update_for_same_game_is_deferred(tmp_path):
# the row-edit flow can't target a collid, so only one update per game
# per run is safe
cfg = _cfg(tmp_path)
_seed_data(
tmp_path,
to_update=[
_update_row(collid="9", bgg_id="2"),
_update_row(collid="10", bgg_id="2", vid="26", vname="2nd ed."),
],
)
fake = FakeUploader()
results = run_upload(cfg, uploader=fake, sleep=lambda s: None, now=_now)
assert [r["collid"] for r in results] == ["9"]
# after the first lands, the next run picks up the deferred one
again = FakeUploader()
results = run_upload(cfg, uploader=again, sleep=lambda s: None, now=_now)
assert [j.collid for j in again.calls] == ["10"]
def test_real_run_without_credentials_exits_before_any_browser(tmp_path, monkeypatch):
monkeypatch.delenv("BGG_USERNAME", raising=False)
monkeypatch.delenv("BGG_PASSWORD", raising=False)
cfg = _cfg(tmp_path)
_seed_data(tmp_path, to_add=[_add_row()])
with pytest.raises(typer.Exit):
run_upload(cfg, sleep=lambda s: None, now=_now) # uploader=None: real path
assert not (tmp_path / "upload_log.csv").exists()
def test_same_key_second_copy_survives_limit_and_interrupts(tmp_path):
# two vetoed duplicate copies share ("add", bgg_id, version): one logged
# success must complete exactly ONE of them, not both
cfg = _cfg(tmp_path)
twin = _add_row(bgg_id="13", name="Catan", version_id="123", version_name="3rd")
_seed_data(tmp_path, to_add=[dict(twin), dict(twin)])
first = FakeUploader()
run_upload(cfg, uploader=first, limit=1, sleep=lambda s: None, now=_now)
assert len(first.calls) == 1
second = FakeUploader()
run_upload(cfg, uploader=second, sleep=lambda s: None, now=_now)
assert len(second.calls) == 1 # the second copy, not zero, not two
third = FakeUploader()
assert run_upload(cfg, uploader=third, sleep=lambda s: None, now=_now) == []
def test_consecutive_failure_counter_resets_on_success(tmp_path):
class FlakyPairs(FakeUploader):
def add_game(self, job):
self.calls.append(job)
if job.bgg_id in ("1", "2", "4", "5"):
raise RuntimeError("dialog never appeared")
return "added", ""
cfg = _cfg(tmp_path)
_seed_data(tmp_path, to_add=[_add_row(bgg_id=str(i)) for i in range(1, 7)])
fake = FlakyPairs()
run_upload(cfg, uploader=fake, sleep=lambda s: None, now=_now)
assert len(fake.calls) == 6 # fail,fail,ok,fail,fail,ok — never aborts
def test_empty_game_name_is_refused_not_uploaded(tmp_path):
cfg = _cfg(tmp_path)
_seed_data(tmp_path, to_add=[_add_row(bgg_id="42", name="")])
fake = FakeUploader()
results = run_upload(cfg, uploader=fake, sleep=lambda s: None, now=_now)
assert results == [] and fake.calls == []
def test_verify_shortfall_reported_once_per_game(tmp_path):
# two DONE adds (different versions) of one game, one copy on BGG:
# the shortfall is reported once per game, not once per logged add
log = [
_log_row(action="add", bgg_id="7", version_id="1", status="added"),
_log_row(action="add", bgg_id="7", version_id="2", status="added"),
]
problems = verify_uploads(log, [_item(7, 70, version_id=1)])
shortfalls = [p for p in problems if "add(s) logged" in p]
assert len(shortfalls) == 1
class _VerifyClient:
def __init__(self, collection):
self.collection = collection
self.calls: list[dict] = []
def collection_full(self, username, *, refresh=False):
self.calls.append({"username": username, "refresh": refresh})
return self.collection
def test_run_upload_verify_wiring(tmp_path, capsys):
# verify=True must re-fetch the LIVE collection (refresh) and cross-check
cfg = _cfg(tmp_path)
_seed_data(tmp_path, to_add=[_add_row(bgg_id="1", name="Wingspan")])
run_upload(cfg, uploader=FakeUploader(), sleep=lambda s: None, now=_now)
client = _VerifyClient([_item(1, 10, name="Wingspan")])
run_upload(
cfg,
uploader=FakeUploader(),
verify=True,
client=client,
sleep=lambda s: None,
now=_now,
)
assert client.calls == [{"username": "tester", "refresh": True}]
assert "Verification OK" in capsys.readouterr().out
def test_verify_marks_second_copy_adds_unverifiable():
log = [
{**_log_row(action="add", bgg_id="13", status="added"), "second_copy": "1"},
]
problems = verify_uploads(log, [_item(13, 900, version_id=7)])
(problem,) = problems
assert "can't be verified" in problem and "confirm by eye" in problem
def test_drift_warning_suppressed_for_multi_edition_partial_run(tmp_path, capsys):
# run 1 added edition A; edition B is still queued alongside A's row:
# that's a two-edition game mid-way, not a re-review drift
cfg = _cfg(tmp_path)
_seed_data(
tmp_path,
to_add=[
_add_row(bgg_id="13", name="Catan", version_id="1", version_name="A"),
_add_row(bgg_id="13", name="Catan", version_id="2", version_name="B"),
],
log=[_log_row(action="add", bgg_id="13", version_id="1", status="added")],
)
run_upload(cfg, uploader=FakeUploader(), sleep=lambda s: None, now=_now)
assert "manual correction" not in capsys.readouterr().out
# -- sign-in detection: never guess from a single absence ----------------
class _FakePage:
"""Counts per selector, replayed one poll at a time — the header
hydrates late, so the first reads legitimately see nothing."""
def __init__(self, script):
self.script = list(script)
self.polls = 0
def locator(self, selector):
counts = self.script[min(self.polls, len(self.script) - 1)]
class _Loc:
def count(self):
return counts["out" if "Sign In" in selector else "in"]
return _Loc()
def wait_for_timeout(self, ms):
self.polls += 1
def _uploader_with(script):
from bggpipe.upload import PlaywrightUploader
up = PlaywrightUploader("someone")
up._page = _FakePage(script)
return up
def test_signed_out_waits_through_hydration():
# neither control for two polls (the window the old check misread as
# "already signed in"), then the Sign In control appears
up = _uploader_with([{"out": 0, "in": 0}, {"out": 0, "in": 0}, {"out": 1, "in": 0}])
assert up._signed_out() is True
def test_signed_in_is_detected_positively():
up = _uploader_with([{"out": 0, "in": 0}, {"out": 0, "in": 1}])
assert up._signed_out() is False
def test_undeterminable_state_raises_instead_of_guessing(monkeypatch):
from bggpipe.upload import LoginError
ticks = iter([0, 1, 2, 31]) # force the deadline to pass
monkeypatch.setattr("time.monotonic", lambda: next(ticks))
up = _uploader_with([{"out": 0, "in": 0}])
with pytest.raises(LoginError, match="could not tell"):
up._signed_out()
# -- version picker: pagination + never-guess matching -------------------
class _FakePicker:
"""Minimal stand-in for the picker sub-view: pages of row texts, a
Next anchor that disables on the last page."""
def __init__(self, pages):
self.pages = pages
self.page = 0
self.clicked = None
self.clicked_index = None
self.cancelled = False
self.reopened = 0
# -- locator plumbing the uploader uses ---------------------------
def locator(self, selector):
picker = self
class _Loc:
def __init__(self, index=0):
self.index = index
def all_text_contents(self):
return picker.pages[picker.page]
def count(self):
if "pagination" in selector:
return 2 # BGG renders a mobile AND a desktop control
return len(picker.pages[picker.page])
def nth(self, i):
return _Loc(i)
def is_visible(self):
# among PAGING controls, index 0 is the mobile
# (visible-xs-*) duplicate — present but not clickable
return "pagination" not in selector or self.index != 0
@property
def first(self):
return _Loc(0)
def wait_for(self, **kw):
return None
def evaluate(self, _js):
last = picker.page >= len(picker.pages) - 1
return "ng-scope disabled" if last else "ng-scope"
def click(self):
if "pagination" in selector and not self.is_visible():
raise AssertionError("clicked a HIDDEN paging control")
if "Next Page" in selector:
picker.page += 1
elif "summary-item-thumbnail" in selector:
picker.clicked_index = self.index
picker.clicked = picker.pages[picker.page][self.index]
def filter(self, has_text=None):
picker.clicked = has_text.pattern if has_text else None
return self
return _Loc()
def get_by_role(self, role, name=None, **kw):
picker = self
class _Btn:
@property
def first(self):
return self
def click(self):
if name == "Cancel":
picker.cancelled = True
elif name == "Set version/edition":
picker.page = 0 # the sub-view reopens on page 1
picker.reopened += 1
return _Btn()
def _picker_uploader(pages):
from bggpipe.upload import PlaywrightUploader
up = PlaywrightUploader("someone")
class _Page:
def wait_for_timeout(self, ms):
pass
up._page = _Page()
return up, _FakePicker(pages)
def test_version_found_on_a_later_page():
up, dialog = _picker_uploader(
[
["Sleeping Gods (Czech edition) (2023)"],
["Sleeping Gods (English Gamefound edition) (2023)"],
]
)
picked, why = up._select_version(dialog, "English Gamefound edition")
assert picked is True # the old code gave up after page one
assert dialog.clicked_index == 0 # clicked by index, not by regex
assert why == ""
def test_api_name_with_printing_qualifier_matches_loosely():
up, dialog = _picker_uploader([["Mysterium (English edition) (2018)"]])
picked, why = up._select_version(dialog, "English edition 2018-2")
assert picked is True
assert "matched loosely" in why
def test_ambiguous_versions_are_refused_not_guessed():
up, dialog = _picker_uploader(
[
[
"Munchkin (English edition) (2009)",
"Munchkin (English edition) (2015)",
]
]
)
picked, why = up._select_version(dialog, "English edition 2009")
assert picked is False
assert "refusing to guess" in why
assert dialog.cancelled is True
def test_absent_version_reports_pages_scanned():
up, dialog = _picker_uploader([["Catan (German edition) (2015)"]])
picked, why = up._select_version(dialog, "English edition")
assert picked is False
assert "not offered" in why and "1 page(s)" in why
def test_paging_never_clicks_a_hidden_mobile_control():
"""BGG renders First/Prev/Next twice — a desktop set and a mobile set
that is invisible on a desktop viewport. Clicking the hidden one hangs
until timeout (it did, on Munchkin Big Box and Tang Garden)."""
up, dialog = _picker_uploader(
[
["Munchkin (German edition) (2010)"],
["Munchkin (English edition) (2009)"],
]
)
picked, _ = up._select_version(dialog, "English edition 2009")
assert picked is True # no AssertionError from the fake = no hidden click
# the initial open plus ONE reopen — never a "First Page" click
assert dialog.reopened == 2
def test_outstanding_failures_ignores_retried_jobs():
"""upload_log.csv is append-only: a failure line stays forever, so the
UI badge must count LAST status per job, not every 'failed' ever."""
from bggpipe.upload import outstanding_failures
rows = [
{
"action": "add",
"bgg_id": "1",
"collid": "",
"version_id": "",
"status": "failed",
},
{
"action": "add",
"bgg_id": "1",
"collid": "",
"version_id": "",
"status": "added",
},
{
"action": "add",
"bgg_id": "2",
"collid": "",
"version_id": "",
"status": "failed",
},
]
assert outstanding_failures(rows) == 1 # game 1 was retried and landed
assert outstanding_failures([]) == 0