Eric WagonerandClaude Fable 5
53ebb15840
Eric asked whether AI residue had crept in since the last credibility
pass. It had, in two forms — both descendants of the audit's
"row-edit flow is collid-exact now" fix, which corrected upload.py but
not everyone who believed the old constraint:
The deferral scaffolding survived its own removal: build_queue still
returned a permanently-empty deferred list, run_upload still unpacked
it and carried a dead echo branch repeating the DISPROVEN claim ("the
row-edit flow can't target a collid"). And diff.py held a vestigial
guard on the same false premise, refusing to upgrade a versionless
copy whenever a sibling carried a version and telling the human to do
it by hand — the collid-exact editor makes that upgrade safe, so the
guard cost a real capability and its test now pins the opposite.
A comment audit (13 findings, verified line-by-line) cleaned the rest:
the webreview docstring counted six pages of seven and claimed
"localhost only" (--lan exists); the nav comment still said "catalog"
for the page renamed Titles; extract's docstring credited "Claude
vision" though any OpenAI-compatible endpoint dispatches; and five
comments narrated retired bugs ("the original bug", "why the old
guess quit after page one", "made the count go negative") — each
rewritten as the present-tense constraint or trimmed, per the
standing rule: comments state what the code can't show, never how it
got that way.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016jXZFSTZQKzAC8fqpWSz9g
1225 lines
40 KiB
Python
1225 lines
40 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,
|
|
annotate_queue,
|
|
build_queue,
|
|
run_upload,
|
|
stale_jobs,
|
|
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_same_game_updates_run_together(tmp_path):
|
|
# update_entry addresses the copy by collid and the edition by radio
|
|
# value, so two copies of one game are safe in a single run
|
|
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", "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:
|
|
"""Stand-in for the picker sub-view. Models the two traps the real one
|
|
sets: paging controls render twice (a hidden mobile duplicate first),
|
|
and paging state SURVIVES closing and reopening the sub-view."""
|
|
|
|
def __init__(self, pages):
|
|
self.pages = pages
|
|
self.page = 0
|
|
self.clicked = None
|
|
self.clicked_index = None
|
|
self.cancelled = False
|
|
self.reopened = 0
|
|
|
|
def locator(self, selector):
|
|
picker = self
|
|
is_next = 'title="Next Page"' in selector
|
|
is_pager = "pagination" in selector
|
|
# anchors: [hidden mobile "1", "1", "2", ...] mirroring the real DOM
|
|
anchors = ["1"] + [str(n + 1) for n in range(len(picker.pages))]
|
|
|
|
class _Loc:
|
|
def __init__(self, index=0):
|
|
self.index = index
|
|
|
|
def all_text_contents(self):
|
|
return picker.pages[picker.page]
|
|
|
|
def count(self):
|
|
if is_next:
|
|
return 2 # hidden mobile + visible desktop
|
|
if is_pager:
|
|
return len(anchors)
|
|
return len(picker.pages[picker.page])
|
|
|
|
def nth(self, i):
|
|
return _Loc(i)
|
|
|
|
@property
|
|
def first(self):
|
|
return _Loc(0)
|
|
|
|
def is_visible(self):
|
|
return not (is_pager and self.index == 0)
|
|
|
|
def text_content(self):
|
|
return anchors[self.index] if is_pager and not is_next else ""
|
|
|
|
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 is_pager and not self.is_visible():
|
|
raise AssertionError("clicked a HIDDEN paging control")
|
|
if is_next:
|
|
picker.page += 1
|
|
elif is_pager:
|
|
picker.page = int(anchors[self.index]) - 1
|
|
else:
|
|
picker.clicked_index = self.index
|
|
picker.clicked = picker.pages[picker.page][self.index]
|
|
|
|
def filter(self, has_text=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.reopened += 1 # note: does NOT reset the page
|
|
|
|
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
|
|
assert dialog.clicked_index == 0 # found on the later page, clicked
|
|
|
|
|
|
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
|
|
|
|
|
|
def test_paging_state_survives_reopen_so_page_one_is_clicked():
|
|
"""The real sub-view reopens wherever it was left, so the second pass
|
|
must navigate back to page 1 explicitly — Sleeping Gods and Gloomhaven
|
|
both "vanished" when it didn't."""
|
|
up, dialog = _picker_uploader(
|
|
[
|
|
["Sleeping Gods (English Gamefound edition) (2023)"],
|
|
["Sleeping Gods (German edition) (2022)"],
|
|
]
|
|
)
|
|
dialog.page = 1 # left on the last page by a previous scan
|
|
picked, why = up._select_version(dialog, "English Gamefound edition")
|
|
assert picked is True
|
|
assert why == ""
|
|
assert dialog.clicked.startswith("Sleeping Gods (English Gamefound")
|
|
|
|
|
|
def test_stale_queue_jobs_are_skipped(tmp_path):
|
|
"""to_add.csv is a snapshot from the last diff. A review decision made
|
|
afterwards — marking a game local, rejecting it, calling the match
|
|
wrong — must win over the stale queue."""
|
|
from bggpipe.upload import stale_jobs
|
|
|
|
queue = [{"bgg_id": "13"}, {"bgg_id": "140509"}, {"bgg_id": "999"}]
|
|
matches = [
|
|
{"bgg_id": "13", "match_status": "auto"},
|
|
{"bgg_id": "140509", "match_status": "local"},
|
|
]
|
|
stale = stale_jobs(queue, matches)
|
|
assert "13" not in stale # still endorsed
|
|
assert "now local" in stale["140509"]
|
|
assert "no longer matched" in stale["999"]
|
|
# unverifiable (no matches.csv at all) never condemns a job
|
|
assert stale_jobs(queue, []) == {}
|
|
|
|
|
|
def test_failure_badge_ignores_jobs_the_human_retired():
|
|
"""A failed job whose review decision has since changed will never run
|
|
again — offering to retry it is a lie the badge kept telling."""
|
|
from bggpipe.upload import outstanding_failures
|
|
|
|
log = [
|
|
{
|
|
"action": "add",
|
|
"bgg_id": "13",
|
|
"collid": "",
|
|
"version_id": "",
|
|
"status": "failed",
|
|
},
|
|
{
|
|
"action": "add",
|
|
"bgg_id": "140509",
|
|
"collid": "",
|
|
"version_id": "",
|
|
"status": "failed",
|
|
},
|
|
]
|
|
queue = [{"bgg_id": "13"}, {"bgg_id": "140509"}]
|
|
matches = [
|
|
{"bgg_id": "13", "match_status": "auto"},
|
|
{"bgg_id": "", "match_status": "local"}, # 140509 was made local
|
|
]
|
|
assert outstanding_failures(log) == 2 # log-only view: both look pending
|
|
assert outstanding_failures(log, queue, matches) == 1 # only the live one
|
|
|
|
|
|
def test_update_targets_collid_and_version_id_exactly():
|
|
"""The collection row's version cell carries its collid; the editor's
|
|
radios carry version ids. Both are addressed by value, so neither the
|
|
copy nor the edition can be picked by resemblance."""
|
|
from bggpipe.upload import PlaywrightUploader, UploadJob
|
|
|
|
seen = {"clicked": [], "text": "Editing"}
|
|
|
|
class _Loc:
|
|
def __init__(self, selector):
|
|
self.selector = selector
|
|
|
|
@property
|
|
def first(self):
|
|
return self
|
|
|
|
def count(self):
|
|
return 1
|
|
|
|
def wait_for(self, **kw):
|
|
return None
|
|
|
|
def click(self):
|
|
seen["clicked"].append(self.selector)
|
|
if "radio" in self.selector:
|
|
seen["text"] = "Avalon Hill second edition" # CE_SaveData ran
|
|
|
|
def text_content(self):
|
|
return seen["text"]
|
|
|
|
class _Page:
|
|
def goto(self, url, **kw):
|
|
seen["url"] = url
|
|
|
|
def locator(self, selector):
|
|
return _Loc(selector)
|
|
|
|
def wait_for_timeout(self, ms):
|
|
pass
|
|
|
|
up = PlaywrightUploader("someone")
|
|
up._page = _Page()
|
|
up._authed = True
|
|
job = UploadJob(
|
|
action="update",
|
|
bgg_id=240,
|
|
name="Britannia",
|
|
collid="53429559",
|
|
version_id="24621",
|
|
version_name="Avalon Hill second edition",
|
|
)
|
|
status, note = up.update_entry(job)
|
|
assert status == "updated" and note == ""
|
|
assert "objectid=240" in seen["url"]
|
|
assert any('onclick*="53429559"' in s for s in seen["clicked"]) # the copy
|
|
assert any('value="24621"' in s for s in seen["clicked"]) # the edition
|
|
|
|
|
|
def test_run_upload_skips_jobs_the_review_retired(tmp_path, capsys):
|
|
"""The queue is a snapshot from the last diff; a review decision taken
|
|
afterwards (local, rejected, re-matched) outranks it. This must hold
|
|
through run_upload's actual wiring, not just stale_jobs in isolation —
|
|
a filter that pruned only to_add would still pass the unit test while
|
|
uploading a retired version update."""
|
|
from bggpipe.resolve import MATCH_COLUMNS
|
|
|
|
cfg = _cfg(tmp_path)
|
|
_seed_data(
|
|
tmp_path,
|
|
to_add=[_add_row(bgg_id="140509", name="Dungeons & Dragons")],
|
|
to_update=[_update_row(collid="9", bgg_id="240", name="Britannia")],
|
|
)
|
|
blank = dict.fromkeys(MATCH_COLUMNS, "")
|
|
_write_csv(
|
|
tmp_path / "matches.csv",
|
|
MATCH_COLUMNS,
|
|
[
|
|
# 140509 was re-reviewed: it's the RPG blue box, kept local
|
|
{**blank, "title_raw": "D&D", "match_status": "local", "bgg_id": ""},
|
|
# 240 vanished from matches entirely (removed from the catalog)
|
|
{**blank, "title_raw": "WINGSPAN", "match_status": "auto", "bgg_id": "1"},
|
|
],
|
|
)
|
|
fake = FakeUploader()
|
|
run_upload(cfg, uploader=fake, sleep=lambda s: None, now=_now)
|
|
assert fake.calls == []
|
|
out = capsys.readouterr().out
|
|
assert "Skipping 2" in out
|
|
# retired jobs leave no trace to be "already done" later
|
|
assert not (tmp_path / "upload_log.csv").exists()
|
|
|
|
# control: an endorsed job still runs
|
|
_write_csv(
|
|
tmp_path / "matches.csv",
|
|
MATCH_COLUMNS,
|
|
[
|
|
{**blank, "title_raw": "X", "match_status": "auto", "bgg_id": "140509"},
|
|
{
|
|
**blank,
|
|
"title_raw": "Y",
|
|
"match_status": "approved",
|
|
"bgg_id": "240",
|
|
"version_id": "25", # per-copy endorsement: version must agree
|
|
},
|
|
],
|
|
)
|
|
run_upload(cfg, uploader=fake, sleep=lambda s: None, now=_now)
|
|
assert [(j.action, j.bgg_id) for j in fake.calls] == [
|
|
("add", "140509"),
|
|
("update", "240"),
|
|
]
|
|
|
|
|
|
def test_rejected_credentials_raise_and_persist_no_session(tmp_path, monkeypatch):
|
|
"""If the site still offers Sign In after the form was submitted, the
|
|
login FAILED — treating it as success would mark the run authed and
|
|
persist an anonymous storage state that poisons every later run."""
|
|
from bggpipe.upload import LoginError, PlaywrightUploader
|
|
|
|
monkeypatch.setenv("BGG_USERNAME", "someone")
|
|
monkeypatch.setenv("BGG_PASSWORD", "wrong-pw")
|
|
state_calls = []
|
|
|
|
class _Loc:
|
|
def __init__(self, n=1):
|
|
self._n = n
|
|
|
|
@property
|
|
def first(self):
|
|
return self
|
|
|
|
def count(self):
|
|
return self._n
|
|
|
|
def is_visible(self):
|
|
return True
|
|
|
|
def wait_for(self, **kw):
|
|
return None
|
|
|
|
def fill(self, value):
|
|
pass
|
|
|
|
def click(self):
|
|
pass
|
|
|
|
class _Page:
|
|
title = staticmethod(lambda: "BoardGameGeek")
|
|
|
|
def goto(self, url, **kw):
|
|
pass
|
|
|
|
def locator(self, selector):
|
|
return _Loc()
|
|
|
|
def get_by_role(self, role, **kw):
|
|
return _Loc()
|
|
|
|
def wait_for_url(self, pred, **kw):
|
|
return None # the URL left /login (site redirects even on failure)
|
|
|
|
def wait_for_timeout(self, ms):
|
|
pass
|
|
|
|
class _Context:
|
|
def storage_state(self, path):
|
|
state_calls.append(path)
|
|
|
|
up = PlaywrightUploader("someone", storage_state=tmp_path / "state.json")
|
|
up._page = _Page()
|
|
up._context = _Context()
|
|
# signed-out before AND after submitting the form: Sign In stays visible
|
|
monkeypatch.setattr(up, "_signed_out", lambda: True)
|
|
with pytest.raises(LoginError, match="credentials rejected"):
|
|
up._ensure_logged_in()
|
|
assert up._authed is False
|
|
assert state_calls == [] # no anonymous session saved for reuse
|
|
|
|
|
|
def test_update_entry_failure_paths_leave_the_entry_untouched():
|
|
"""The three guarded exits of the collection-cell editor: no cell for
|
|
the collid, the wanted version id never offered (must press Escape —
|
|
the editor is open on a REAL entry), and a save that never settles."""
|
|
from bggpipe.upload import PlaywrightUploader, UploadJob
|
|
|
|
class _Timeout(Exception):
|
|
pass
|
|
|
|
def make_page(cell_count=1, radio_appears=True, settles=True):
|
|
pressed = []
|
|
|
|
class _Loc:
|
|
def __init__(self, selector):
|
|
self.selector = selector
|
|
|
|
@property
|
|
def first(self):
|
|
return self
|
|
|
|
def count(self):
|
|
return cell_count if "collection_version" in self.selector else 1
|
|
|
|
def wait_for(self, **kw):
|
|
if "radio" in self.selector and not radio_appears:
|
|
raise _Timeout("radio never appeared")
|
|
|
|
def click(self):
|
|
pass
|
|
|
|
def text_content(self):
|
|
return "Editing" if not settles else "Avalon Hill second"
|
|
|
|
class _Keyboard:
|
|
def press(self, key):
|
|
pressed.append(key)
|
|
|
|
class _Page:
|
|
keyboard = _Keyboard()
|
|
|
|
def goto(self, url, **kw):
|
|
pass
|
|
|
|
def locator(self, selector):
|
|
return _Loc(selector)
|
|
|
|
def wait_for_timeout(self, ms):
|
|
pass
|
|
|
|
return _Page(), pressed
|
|
|
|
def uploader(page):
|
|
up = PlaywrightUploader("someone")
|
|
up._page = page
|
|
up._authed = True
|
|
up._timeout_error = _Timeout
|
|
return up
|
|
|
|
job = UploadJob(
|
|
action="update",
|
|
bgg_id=240,
|
|
name="Britannia",
|
|
collid="53429559",
|
|
version_id="24621",
|
|
version_name="AH",
|
|
)
|
|
|
|
page, _ = make_page(cell_count=0)
|
|
with pytest.raises(RuntimeError, match="re-run diff"):
|
|
uploader(page).update_entry(job)
|
|
|
|
page, pressed = make_page(radio_appears=False)
|
|
with pytest.raises(RuntimeError, match="entry left untouched"):
|
|
uploader(page).update_entry(job)
|
|
assert pressed == ["Escape"] # the open editor was closed, not abandoned
|
|
|
|
page, _ = make_page(settles=False)
|
|
with pytest.raises(RuntimeError, match="never left its editing state"):
|
|
uploader(page).update_entry(job)
|
|
|
|
|
|
def test_scrub_flattens_multiline_errors_for_the_csv_log():
|
|
text = "Call log:\n - waiting for locator\n - retrying click"
|
|
assert "\n" not in _scrub(text)
|
|
assert _scrub(text) == "Call log: - waiting for locator - retrying click"
|
|
|
|
|
|
def test_stale_jobs_retires_the_copy_not_just_the_game():
|
|
"""Rejecting ONE of two queued editions must not ride along on the
|
|
surviving edition's endorsement."""
|
|
queue = [
|
|
{"bgg_id": "589", "version_id": "100"},
|
|
{"bgg_id": "589", "version_id": "200"},
|
|
]
|
|
matches = [
|
|
{"bgg_id": "589", "match_status": "approved", "version_id": "100"},
|
|
{"bgg_id": "589", "match_status": "rejected", "version_id": "200"},
|
|
]
|
|
stale = stale_jobs(queue, matches)
|
|
assert "589" in stale and "no longer endorses that copy" in stale["589"]
|
|
# both endorsed: nothing stale
|
|
matches[1]["match_status"] = "approved"
|
|
assert stale_jobs(queue, matches) == {}
|
|
|
|
|
|
def test_annotate_queue_marks_one_done_per_completion():
|
|
"""Vetoed duplicate copies share a job key; one success is one copy."""
|
|
queue = [
|
|
{"bgg_id": "13", "version_id": ""},
|
|
{"bgg_id": "13", "version_id": ""},
|
|
]
|
|
log = [_log_row(bgg_id="13", status="added")]
|
|
states = [r["state"] for r in annotate_queue(queue, "add", log)]
|
|
assert states == ["done", ""]
|
|
log.append(_log_row(bgg_id="13", status="added"))
|
|
states = [r["state"] for r in annotate_queue(queue, "add", log)]
|
|
assert states == ["done", "done"]
|
|
|
|
|
|
def test_add_game_reports_already_present_on_owned_page():
|
|
"""A game page showing "In Collections" instead of "Add To" is positive
|
|
evidence a previous attempt landed — retrying the add would duplicate."""
|
|
from bggpipe.upload import PlaywrightUploader, UploadJob
|
|
|
|
class _Btn:
|
|
def __init__(self, present):
|
|
self._present = present
|
|
|
|
@property
|
|
def first(self):
|
|
return self
|
|
|
|
def count(self):
|
|
return 1 if self._present else 0
|
|
|
|
def is_visible(self):
|
|
return self._present
|
|
|
|
class _Page:
|
|
def goto(self, url, **kw):
|
|
pass
|
|
|
|
def get_by_role(self, role, name=None, **kw):
|
|
return _Btn(name == "In Collections")
|
|
|
|
def wait_for_timeout(self, ms):
|
|
pass
|
|
|
|
up = PlaywrightUploader("someone")
|
|
up._page = _Page()
|
|
up._authed = True
|
|
job = UploadJob(action="add", bgg_id="240", name="Britannia")
|
|
status, note = up.add_game(job)
|
|
assert status == "already_present"
|
|
assert "previous attempt" in note
|
|
|
|
# a SECOND copy can't go through the unverified In Collections flow
|
|
job2 = UploadJob(action="add", bgg_id="240", name="Britannia", second_copy=True)
|
|
with pytest.raises(RuntimeError, match="add this copy by hand"):
|
|
up.add_game(job2)
|
|
|
|
|
|
def test_update_entry_rejects_a_settle_on_the_wrong_version():
|
|
"""The editor closing proves nothing: if the cell re-renders its OLD
|
|
content the AJAX save failed and "updated" would be a false record."""
|
|
from bggpipe.upload import PlaywrightUploader, UploadJob
|
|
|
|
class _Loc:
|
|
def __init__(self, selector):
|
|
self.selector = selector
|
|
|
|
@property
|
|
def first(self):
|
|
return self
|
|
|
|
def count(self):
|
|
return 1
|
|
|
|
def wait_for(self, **kw):
|
|
pass
|
|
|
|
def click(self):
|
|
pass
|
|
|
|
def text_content(self):
|
|
return "Some Other Edition (2001)" # the OLD content, not ours
|
|
|
|
class _Page:
|
|
def goto(self, url, **kw):
|
|
pass
|
|
|
|
def locator(self, selector):
|
|
return _Loc(selector)
|
|
|
|
def wait_for_timeout(self, ms):
|
|
pass
|
|
|
|
up = PlaywrightUploader("someone")
|
|
up._page = _Page()
|
|
up._authed = True
|
|
job = UploadJob(
|
|
action="update",
|
|
bgg_id="240",
|
|
name="Britannia",
|
|
collid="9",
|
|
version_id="24621",
|
|
version_name="Avalon Hill second edition",
|
|
)
|
|
with pytest.raises(RuntimeError, match="did not land"):
|
|
up.update_entry(job)
|
|
|
|
|
|
def test_version_drift_skips_the_doomed_readd(tmp_path, capsys):
|
|
"""A game the log shows added under a different version is already on
|
|
BGG — re-adding can only duplicate it, so the queue skips it."""
|
|
cfg = _cfg(tmp_path)
|
|
_seed_data(
|
|
tmp_path,
|
|
to_add=[_add_row(bgg_id="1", name="Wingspan", version_id="99")],
|
|
log=[_log_row(bgg_id="1", version_id="55", status="added")],
|
|
)
|
|
fake = FakeUploader()
|
|
run_upload(cfg, uploader=fake, sleep=lambda s: None, now=_now)
|
|
assert fake.calls == []
|
|
assert "correct the version by hand" in capsys.readouterr().out
|
|
|
|
|
|
def test_abort_needs_identical_messages_not_just_same_class(tmp_path):
|
|
"""Three DIFFERENT per-game failures are three unlucky games, not a
|
|
systemic outage; only identical messages abort the run."""
|
|
cfg = _cfg(tmp_path)
|
|
rows = [_add_row(bgg_id=str(i), name=f"Game {i}") for i in (1, 2, 3, 4)]
|
|
_seed_data(tmp_path, to_add=rows)
|
|
|
|
class _VariedFail(FakeUploader):
|
|
def add_game(self, job):
|
|
self.calls.append(job)
|
|
raise RuntimeError(f"distinct problem with {job.name}")
|
|
|
|
varied = _VariedFail()
|
|
run_upload(cfg, uploader=varied, sleep=lambda s: None, now=_now)
|
|
assert len(varied.calls) == 4 # no abort: every message differed
|