diff --git a/src/bggpipe/config.py b/src/bggpipe/config.py
index 3eb86bf..dbe3965 100644
--- a/src/bggpipe/config.py
+++ b/src/bggpipe/config.py
@@ -95,6 +95,11 @@ class Config:
def load_config(path: Path | None = None) -> Config:
cfg = Config()
p = path or DEFAULT_CONFIG_PATH
+ if path is not None and not path.exists():
+ # an EXPLICIT --config that doesn't exist must not silently fall
+ # back to defaults — that reads photos/ while the user believes
+ # their prod config is active
+ raise FileNotFoundError(f"--config {path} does not exist")
if p.exists():
raw = tomllib.loads(p.read_text())
known = {
diff --git a/src/bggpipe/diff.py b/src/bggpipe/diff.py
index b649ce0..1e0f165 100644
--- a/src/bggpipe/diff.py
+++ b/src/bggpipe/diff.py
@@ -19,12 +19,18 @@ Outputs both artifacts:
from __future__ import annotations
import os
+from collections import Counter
from dataclasses import dataclass, field
from pathlib import Path
import typer
-from bggpipe.bgg_client import BGGAuthError, BGGClient, client_for
+from bggpipe.bgg_client import (
+ BGGAuthError,
+ BGGClient,
+ BGGQueueTimeout,
+ client_for,
+)
from bggpipe.config import Config
from bggpipe.fsio import atomic_write_csv
from bggpipe.models import (
@@ -44,6 +50,7 @@ TO_ADD_COLUMNS = [
"version_name",
"title_raw",
"source_photos",
+ "second_copy", # "1": game already had copies — verify can't confirm it
]
TO_UPDATE_COLUMNS = ["collid", "bgg_id", "bgg_name", "version_id", "version_name"]
@@ -100,7 +107,10 @@ def compute_diff(rows: list[dict], collection: list[CollectionItem]) -> DiffResu
p for p in row["source_photos"].split(";") if p
)
- def add_row(row: dict, confident: bool) -> dict:
+ queued_adds: Counter[int] = Counter() # per game, across all passes
+
+ def add_row(row: dict, confident: bool, second_copy: bool = False) -> dict:
+ queued_adds[int(row["bgg_id"])] += 1
photos = {p for p in row["source_photos"].split(";") if p}
photos |= merged_photos.get(row["title_raw"], set())
return {
@@ -112,6 +122,7 @@ def compute_diff(rows: list[dict], collection: list[CollectionItem]) -> DiffResu
"version_name": row["version_name"] if confident else "",
"title_raw": row["title_raw"],
"source_photos": ";".join(sorted(photos)),
+ "second_copy": "1" if second_copy else "",
}
recognized: list[dict] = []
@@ -212,7 +223,7 @@ def compute_diff(rows: list[dict], collection: list[CollectionItem]) -> DiffResu
"left untouched"
)
else:
- result.to_add.append(add_row(row, True))
+ result.to_add.append(add_row(row, True, second_copy=True))
result.second_copies.append(
f"{row['title_raw']}: adding as a NEW copy with version "
f"{row['version_name']!r} ({row['version_id']}) — every "
@@ -226,14 +237,22 @@ def compute_diff(rows: list[dict], collection: list[CollectionItem]) -> DiffResu
bgg_id = int(row["bgg_id"])
copies = by_object.get(bgg_id, [])
if not copies:
- result.to_add.append(add_row(row, False))
+ if queued_adds[bgg_id] and not row.get("dedupe_veto"):
+ # an unvetoed bare row is a typo-read sibling: a pass-1 add
+ # for the same absent game already covers the physical box —
+ # queueing it again would upload a duplicate entry
+ result.already_owned.append(row["title_raw"])
+ else:
+ result.to_add.append(
+ add_row(row, False, second_copy=bool(queued_adds[bgg_id]))
+ )
continue
remaining = unconsumed(bgg_id)
if remaining:
consumed_collids.add(remaining[0].coll_id)
result.already_owned.append(row["title_raw"])
elif row.get("dedupe_veto"):
- result.to_add.append(add_row(row, False))
+ result.to_add.append(add_row(row, False, second_copy=True))
result.second_copies.append(
f"{row['title_raw']}: adding as a NEW version-less copy — "
"human-vetoed duplicate, every existing entry claimed by "
@@ -263,7 +282,7 @@ def run_diff(cfg: Config, *, client: BGGClient | None = None) -> DiffResult:
client = client or client_for(cfg)
try:
collection = client.collection_full(cfg.bgg_username, refresh=True)
- except BGGAuthError as err:
+ except (BGGAuthError, BGGQueueTimeout) as err:
# a present-but-invalid token must not traceback when the
# snapshot fallback is sitting right there
typer.echo(f"Live fetch failed ({err}) — using snapshot files.")
diff --git a/src/bggpipe/enrich.py b/src/bggpipe/enrich.py
index aa56c12..eaa388f 100644
--- a/src/bggpipe/enrich.py
+++ b/src/bggpipe/enrich.py
@@ -19,7 +19,12 @@ import json
import typer
-from bggpipe.bgg_client import BGGAuthError, BGGClient, client_for
+from bggpipe.bgg_client import (
+ BGGAuthError,
+ BGGClient,
+ BGGQueueTimeout,
+ client_for,
+)
from bggpipe.config import Config
from bggpipe.fsio import atomic_write_text
from bggpipe.models import is_confident_version, is_recognized
@@ -80,7 +85,7 @@ def run_enrich(
try:
for thing in client.things_full(batch, refresh=refresh):
fetched[thing["bgg_id"]] = thing
- except BGGAuthError:
+ except (BGGAuthError, BGGQueueTimeout):
blocked = True
break
@@ -90,6 +95,20 @@ def run_enrich(
games[key] = {**fetched[bgg_id], "version": version}
updated += 1
+ # prune keys no current target claims: a row whose version was approved
+ # after a bare-key run (or was later rejected) must not leave an orphan
+ # entry in the frontend seed data. Only safe when nothing was blocked —
+ # a token-less run knows too little to declare anything stale.
+ if not blocked:
+ current = {key for key, _, _ in targets}
+ stale = [k for k in games if k not in current]
+ for k in stale:
+ del games[k]
+ if stale:
+ typer.echo(
+ f" pruned {len(stale)} stale entr{'y' if len(stale) == 1 else 'ies'}"
+ )
+
games_path.parent.mkdir(parents=True, exist_ok=True)
atomic_write_text(
games_path, json.dumps(games, indent=2, ensure_ascii=False) + "\n"
diff --git a/src/bggpipe/extract.py b/src/bggpipe/extract.py
index cb33f31..9e075ce 100644
--- a/src/bggpipe/extract.py
+++ b/src/bggpipe/extract.py
@@ -265,7 +265,13 @@ def rebuild_artifacts(
entries: list[dict] = []
unidentified: dict[str, list[dict]] = {}
for raw_file in sorted(raw_dir.glob("*.json")):
- data = json.loads(raw_file.read_text())
+ try:
+ data = json.loads(raw_file.read_text())
+ except json.JSONDecodeError as err:
+ raise ValueError(
+ f"{raw_file} is corrupt ({err}) — delete it (or re-upload "
+ "the photo) and run extract again"
+ ) from err
if isinstance(data, list): # bare-array shape
entries.extend(data)
continue
@@ -315,17 +321,37 @@ def run_extract(
vision = vision or default_vision(cfg.model)
failed: list[str] = []
+ consecutive: tuple[str, int] = ("", 0)
+ attempted = 0
for photo in photos:
raw_path = raw_dir / f"{photo.name}.json"
if raw_path.exists() and not only and not force:
- typer.echo(f" {photo.name}: already extracted, skipping")
- continue
+ try:
+ json.loads(raw_path.read_text())
+ typer.echo(f" {photo.name}: already extracted, skipping")
+ continue
+ except json.JSONDecodeError:
+ typer.echo(f" {photo.name}: cache corrupt — re-extracting")
+ attempted += 1
try:
result = extract_photo(photo, vision)
except Exception as err: # one bad photo must not block the rest
failed.append(photo.name)
typer.echo(f" {photo.name}: FAILED ({err}) — continuing")
+ kind = type(err).__name__
+ consecutive = (
+ kind,
+ consecutive[1] + 1 if kind == consecutive[0] else 1,
+ )
+ if consecutive[1] >= 3:
+ typer.echo(
+ " aborting — 3 identical consecutive failures look "
+ "systemic (bad API key? wrong model in config.toml?), "
+ "not per-photo; nothing more will be attempted"
+ )
+ break
continue
+ consecutive = ("", 0)
atomic_write_text(
raw_path, json.dumps(result, indent=2, ensure_ascii=False) + "\n"
)
@@ -352,6 +378,10 @@ def run_extract(
f"\n{len(failed)} photo(s) failed extraction (re-run to retry): "
+ ", ".join(failed)
)
+ if len(failed) == attempted:
+ # every attempted photo failed: this run accomplished nothing
+ # and a wrapper (or the web job runner) must not report success
+ raise typer.Exit(code=1)
if unidentified:
typer.echo(
diff --git a/src/bggpipe/fsio.py b/src/bggpipe/fsio.py
index b510377..6f99678 100644
--- a/src/bggpipe/fsio.py
+++ b/src/bggpipe/fsio.py
@@ -9,12 +9,26 @@ from __future__ import annotations
import csv
import os
+import uuid
from pathlib import Path
+def _tmp_for(path: Path) -> Path:
+ # unique per writer: a FIXED tmp name lets two concurrent processes
+ # interleave writes into one inode and replace garbage into place
+ return path.with_name(f"{path.name}.{uuid.uuid4().hex[:8]}.tmp")
+
+
+def atomic_write_bytes(path: Path, data: bytes) -> None:
+ path.parent.mkdir(parents=True, exist_ok=True)
+ tmp = _tmp_for(path)
+ tmp.write_bytes(data)
+ os.replace(tmp, path)
+
+
def atomic_write_text(path: Path, text: str) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
- tmp = path.with_name(path.name + ".tmp")
+ tmp = _tmp_for(path)
tmp.write_text(text)
os.replace(tmp, path)
@@ -23,12 +37,16 @@ def atomic_write_csv(path: Path, columns: list[str], rows: list[dict]) -> int:
"""Atomic CSV rewrite (tmp + os.replace). Returns the written file's
mtime_ns so callers tracking their own writes avoid a re-stat race."""
path.parent.mkdir(parents=True, exist_ok=True)
- tmp = path.with_name(path.name + ".tmp")
+ tmp = _tmp_for(path)
with tmp.open("w", newline="") as f:
writer = csv.DictWriter(
f, fieldnames=columns, extrasaction="ignore", restval=""
)
writer.writeheader()
writer.writerows(rows)
+ # stat the TMP inode before the replace: statting the destination after
+ # could hand back a FOREIGN writer's mtime landing in the gap, and the
+ # caller would record someone else's write as its own
+ mtime = tmp.stat().st_mtime_ns
os.replace(tmp, path)
- return path.stat().st_mtime_ns
+ return mtime
diff --git a/src/bggpipe/init_wizard.py b/src/bggpipe/init_wizard.py
index 63504d7..b2cb4a3 100644
--- a/src/bggpipe/init_wizard.py
+++ b/src/bggpipe/init_wizard.py
@@ -77,20 +77,34 @@ class InitReport:
def _env_file_keys(env_path: Path) -> set[str]:
- """Key names with non-empty values in .env. Values are never read into
- variables that outlive this parse and are never printed."""
+ """Key names with non-empty values in .env. Handles `export KEY=v` and
+ quoted values; a quoted-empty value ("" / '') counts as NOT set. Values
+ never outlive this parse and are never printed."""
if not env_path.exists():
return set()
present = set()
for line in env_path.read_text().splitlines():
line = line.strip()
- if "=" in line and not line.startswith("#"):
- key, _, value = line.partition("=")
- if value.strip():
- present.add(key.strip())
+ if "=" not in line or line.startswith("#"):
+ continue
+ key, _, value = line.partition("=")
+ key = key.strip()
+ if key.startswith("export "):
+ key = key.removeprefix("export ").strip()
+ value = value.strip()
+ if len(value) >= 2 and value[0] == value[-1] and value[0] in "'\"":
+ value = value[1:-1]
+ if value:
+ present.add(key)
return present
+def _quote_env_value(value: str) -> str:
+ """Single-quote for `source`/direnv safety: spaces, $, backslashes and
+ quotes must survive the shell verbatim."""
+ return "'" + value.replace("'", "'\\''") + "'"
+
+
def _default_secret_prompt(label: str) -> str:
return typer.prompt(label, default="", show_default=False, hide_input=True)
@@ -162,11 +176,14 @@ def run_init(
report.keys_missing.append(key)
continue
is_new_file = not env_path.exists()
- with env_path.open("a") as f:
+ # owner-only from the FIRST byte — chmod-after-write leaves a
+ # world-readable window holding a credential
+ fd = os.open(env_path, os.O_CREAT | os.O_WRONLY | os.O_APPEND, 0o600)
+ with os.fdopen(fd, "a") as f:
if is_new_file:
f.write(ENV_HEADER)
- f.write(f"{key}={value}\n")
- env_path.chmod(0o600) # credentials: owner-only, every time
+ f.write(f"{key}={_quote_env_value(value)}\n")
+ env_path.chmod(0o600) # older files created by hand tighten up too
report.keys_written.append(key)
typer.echo(f" {key}: saved to {env_path}")
diff --git a/src/bggpipe/jobs.py b/src/bggpipe/jobs.py
index 5a45c6e..aefc6a0 100644
--- a/src/bggpipe/jobs.py
+++ b/src/bggpipe/jobs.py
@@ -13,11 +13,14 @@ from __future__ import annotations
import io
import threading
import time
+import traceback
from collections.abc import Callable
from contextlib import redirect_stdout
import typer
+MAX_LOG_LINES = 1000 # buffer cap; snapshots serve the last 200
+
class _LineBuffer(io.TextIOBase):
def __init__(self) -> None:
@@ -30,6 +33,8 @@ class _LineBuffer(io.TextIOBase):
self._partial += text
*complete, self._partial = self._partial.split("\n")
self._lines.extend(complete)
+ if len(self._lines) > MAX_LOG_LINES: # bound memory on long runs
+ del self._lines[: len(self._lines) - MAX_LOG_LINES]
return len(text)
def lines(self) -> list[str]:
@@ -68,20 +73,27 @@ class JobRunner:
def _run(self, fn: Callable[[], object]) -> None:
status, error = "done", ""
try:
- with redirect_stdout(self._buffer):
- fn()
- except typer.Exit as exc:
- if exc.exit_code:
- status, error = "failed", f"exited with code {exc.exit_code}"
- except SystemExit as exc:
- if exc.code:
- status, error = "failed", f"exited with code {exc.code}"
- except Exception as exc: # surfaced in the UI, never swallowed
- status, error = "failed", f"{type(exc).__name__}: {exc}"
- with self._lock:
- self._status = status
- self._error = error
- self._finished = time.time()
+ try:
+ with redirect_stdout(self._buffer):
+ fn()
+ except typer.Exit as exc:
+ if exc.exit_code:
+ status, error = "failed", f"exited with code {exc.exit_code}"
+ except SystemExit as exc:
+ if exc.code:
+ status, error = "failed", f"exited with code {exc.code}"
+ except BaseException as exc: # incl. GreenletExit et al: the
+ # runner must NEVER stay "running" — that wedges every
+ # future stage behind a 409 until a server restart
+ status, error = "failed", f"{type(exc).__name__}: {exc}"
+ # the short error names the exception; the log carries the
+ # traceback, or a failure is undiagnosable from the UI
+ self._buffer.write("\n" + traceback.format_exc())
+ finally:
+ with self._lock:
+ self._status = status
+ self._error = error
+ self._finished = time.time()
def snapshot(self) -> dict:
with self._lock:
diff --git a/src/bggpipe/resolve.py b/src/bggpipe/resolve.py
index aa3d216..13d6d38 100644
--- a/src/bggpipe/resolve.py
+++ b/src/bggpipe/resolve.py
@@ -475,6 +475,23 @@ def dedupe_matches(rows: list[dict], titles: list[TitleEntry]) -> list[MergeEven
bgg_id=bgg_id,
)
)
+ # A survivor can itself lose in a later run; rows pointing at it would
+ # form a chain diff's one-level photo hop can't follow. Rewrite every
+ # merged row to its terminal survivor.
+ status_by_title = {r["title_raw"]: r for r in rows}
+ for row in rows:
+ if row["match_status"] != "merged":
+ continue
+ target, hops = row["merged_into"], 0
+ while (
+ hops < 10
+ and (nxt := status_by_title.get(target)) is not None
+ and nxt["match_status"] == "merged"
+ and nxt["merged_into"]
+ ):
+ target = nxt["merged_into"]
+ hops += 1
+ row["merged_into"] = target
return events
diff --git a/src/bggpipe/static/app.js b/src/bggpipe/static/app.js
index 6ee7f40..78e33b0 100644
--- a/src/bggpipe/static/app.js
+++ b/src/bggpipe/static/app.js
@@ -16,8 +16,11 @@ function showBanner(html) {
}
function errorBanner(detail) {
- showBanner(`lost contact with the server ` +
- `(${esc(detail)}) — check its terminal
`);
+ // an HTTP status means the server answered — the problem is its data
+ const httpish = /^\d{3}$/.test(String(detail));
+ showBanner(`${httpish
+ ? `the server hit an error (HTTP ${esc(detail)}) — a data file may be broken`
+ : `lost contact with the server (${esc(detail)})`} — check its terminal
`);
}
async function fetchJSON(url) {
diff --git a/src/bggpipe/templates/pages/catalog.html b/src/bggpipe/templates/pages/catalog.html
index 46f8778..686b945 100644
--- a/src/bggpipe/templates/pages/catalog.html
+++ b/src/bggpipe/templates/pages/catalog.html
@@ -51,5 +51,5 @@ async function refresh() {
document.getElementById("catsearch").addEventListener("input", render);
refresh().catch(err => errorBanner(err.message || err));
-pollLoop(refresh, 5000);
+pollLoop(refresh, 5000, () => showBanner(""));
diff --git a/src/bggpipe/templates/pages/library.html b/src/bggpipe/templates/pages/library.html
index 45135be..38394fb 100644
--- a/src/bggpipe/templates/pages/library.html
+++ b/src/bggpipe/templates/pages/library.html
@@ -57,5 +57,5 @@ async function refresh() {
document.getElementById("libsearch").addEventListener("input", render);
refresh().catch(err => errorBanner(err.message || err));
-pollLoop(refresh, 10000);
+pollLoop(refresh, 10000, () => showBanner(""));
diff --git a/src/bggpipe/templates/pages/photos.html b/src/bggpipe/templates/pages/photos.html
index ec5c530..9a7cc33 100644
--- a/src/bggpipe/templates/pages/photos.html
+++ b/src/bggpipe/templates/pages/photos.html
@@ -87,15 +87,21 @@ async function sendPhotos(files) {
if (!files.length) return;
const form = new FormData();
for (const f of files) form.append("files", f);
- const res = await fetch("/api/photos", {method: "POST", body: form});
+ let res;
+ try {
+ res = await fetch("/api/photos", {method: "POST", body: form});
+ } catch (err) {
+ alert("Upload failed (no response from the server): " + err);
+ return;
+ }
if (!res.ok) {
const detail = await res.json().then(d => d.detail).catch(() => null);
alert("Upload failed: " + (detail ?? res.statusText));
return;
}
- refresh();
+ refresh().catch(() => {}); // the next poll self-heals a refresh hiccup
}
refresh().catch(err => errorBanner(err.message || err));
-pollLoop(refresh, 5000);
+pollLoop(refresh, 5000, () => showBanner(""));
diff --git a/src/bggpipe/templates/pages/pipeline.html b/src/bggpipe/templates/pages/pipeline.html
index 3f7f07f..b1e8f9a 100644
--- a/src/bggpipe/templates/pages/pipeline.html
+++ b/src/bggpipe/templates/pages/pipeline.html
@@ -11,7 +11,7 @@ let running = false;
async function runStage(stage, body) {
const res = await apiPost(`/api/run/${stage}`, body);
- if (res) refresh();
+ if (res) refresh().catch(() => {}); // the poll self-heals a hiccup
}
function stageCard(num, name, facts, actions) {
@@ -28,6 +28,7 @@ function runBtn(stage, label) {
function render() {
const m = P.matches, log = P.upload_log;
+ running = P.job.status === "running"; // buttons below depend on this
const banners = [];
if (P.stub_data) banners.push(
@@ -67,7 +68,6 @@ function render() {
});
const job = P.job;
- running = job.status === "running";
const state = document.getElementById("jobstate");
if (job.status === "idle") state.textContent = "idle";
else state.innerHTML = `${esc(job.stage)}: ${esc(job.status)}` +
diff --git a/src/bggpipe/templates/pages/queue.html b/src/bggpipe/templates/pages/queue.html
index 8cd5df2..933cb1a 100644
--- a/src/bggpipe/templates/pages/queue.html
+++ b/src/bggpipe/templates/pages/queue.html
@@ -53,5 +53,5 @@ async function refresh() {
}
refresh().catch(err => errorBanner(err.message || err));
-pollLoop(refresh, 5000);
+pollLoop(refresh, 5000, () => showBanner(""));
diff --git a/src/bggpipe/templates/pages/review.html b/src/bggpipe/templates/pages/review.html
index 581c5d7..16da8ec 100644
--- a/src/bggpipe/templates/pages/review.html
+++ b/src/bggpipe/templates/pages/review.html
@@ -267,7 +267,10 @@ refresh().catch(err => errorBanner(err.message || err));
let lastGood = null;
pollLoop(async () => {
const fresh = await fetchJSON("/api/state");
- if (STATE && fresh.revision < STATE.revision) return;
+ // discard stale responses — but only within one server lifetime: a
+ // restart resets revision to 0 (boot changes), and refusing forever
+ // would freeze the page
+ if (STATE && fresh.boot === STATE.boot && fresh.revision < STATE.revision) return;
if (JSON.stringify(fresh) !== JSON.stringify(STATE)) {
STATE = fresh;
render();
diff --git a/src/bggpipe/upload.py b/src/bggpipe/upload.py
index 7663107..478c6c4 100644
--- a/src/bggpipe/upload.py
+++ b/src/bggpipe/upload.py
@@ -48,6 +48,7 @@ UPLOAD_LOG_COLUMNS = [
"collid",
"name",
"version_id",
+ "second_copy",
"status",
"timestamp",
"error",
@@ -64,6 +65,7 @@ class UploadJob:
version_id: str = ""
version_name: str = ""
collid: str = ""
+ second_copy: bool = False # game had prior copies: verify can't confirm
@property
def key(self) -> tuple[str, str, str]:
@@ -135,6 +137,7 @@ def build_queue(
name=row["bgg_name"],
version_id=row.get("version_id", ""),
version_name=row.get("version_name", ""),
+ second_copy=bool(row.get("second_copy")),
)
for row in to_add
] + [
@@ -161,11 +164,19 @@ def build_queue(
deferred: list[UploadJob] = []
update_game_seen: set[str] = set()
seen: Counter[tuple[str, str, str]] = Counter()
+ queued_versions: dict[str, set[str]] = {}
+ for job in candidates:
+ if job.action == "add":
+ queued_versions.setdefault(job.bgg_id, set()).add(job.version_id)
for job in candidates:
prior = done_versions.get(
("update", job.collid) if job.action == "update" else ("add", job.bgg_id),
set(),
)
+ if job.action == "add" and prior & queued_versions.get(job.bgg_id, set()):
+ # the done version is still queued alongside this one: a
+ # multi-edition game partway through, not a re-review drift
+ prior = set()
if prior and job.version_id not in prior:
typer.echo(
f" note: {job.name} was previously {job.action}ed with a "
@@ -303,6 +314,7 @@ class PlaywrightUploader:
"page changed — see docs/bgg-upload-flow.md)"
) from err
self._context.storage_state(path=str(self._storage_state))
+ self._storage_state.chmod(0o600) # session cookies: owner-only
self._authed = True
def _open_dialog(self, opener):
@@ -444,6 +456,7 @@ def _process(
"collid": job.collid,
"name": job.name,
"version_id": job.version_id,
+ "second_copy": "1" if job.second_copy else "",
"status": status,
"timestamp": now(),
# the column carries degradation notes on successes too
@@ -488,6 +501,15 @@ def verify_uploads(log_rows: list[dict], collection: list[CollectionItem]) -> li
added_copies[int(row["bgg_id"])] += 1
for row in latest.values():
if row["status"] in ("added", "added_no_version"):
+ if row.get("second_copy"):
+ # the game had copies before this add: collection state can't
+ # distinguish "new entry created" from "existing entry
+ # edited" (the failure the unverified dialog could produce)
+ problems.append(
+ f"{row['name']}: second-copy add can't be verified from "
+ "collection counts — confirm by eye on BGG"
+ )
+ continue
copies = by_object.get(int(row["bgg_id"]), [])
if not copies:
problems.append(f"{row['name']}: logged added but not in collection")
diff --git a/src/bggpipe/webreview.py b/src/bggpipe/webreview.py
index 3770f89..78ac201 100644
--- a/src/bggpipe/webreview.py
+++ b/src/bggpipe/webreview.py
@@ -18,6 +18,7 @@ import io
import json
import os
import threading
+import time
import warnings
import webbrowser
import xml.etree.ElementTree as ET
@@ -26,17 +27,18 @@ from collections.abc import Callable
from functools import partial
from importlib import resources
from pathlib import Path
+from urllib.parse import urlsplit
import typer
from defusedxml.ElementTree import fromstring as _safe_fromstring
from fastapi import FastAPI, HTTPException, UploadFile
-from fastapi.responses import FileResponse, HTMLResponse, Response
+from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, Response
from pydantic import BaseModel
from rich.console import Console
from bggpipe.bgg_client import BGGClient, cached_paths
from bggpipe.config import DEFAULT_REVIEW_PORT, Config
-from bggpipe.fsio import atomic_write_text
+from bggpipe.fsio import atomic_write_bytes, atomic_write_text
from bggpipe.jobs import JobRunner
from bggpipe.models import (
CONFIDENT_VERSION_STATUSES,
@@ -219,13 +221,39 @@ def create_app(
input_fn=lambda prompt: "",
client=client,
)
- thumbnails = load_thumbnails(cfg.cache_dir)
- dismissed = DismissStore(cfg.dismissed_path)
+ with warnings.catch_warnings(record=True) as caught:
+ warnings.simplefilter("always")
+ thumbnails = load_thumbnails(cfg.cache_dir)
+ dismissed = DismissStore(cfg.dismissed_path)
+ startup_notes = [str(w.message) for w in caught]
# FastAPI runs sync endpoints in a threadpool: without this, a polled
# GET /api/state can freshen()-swap session.rows out from under a
# concurrent decision POST, silently dropping the decision.
lock = threading.Lock()
revision = {"n": 0} # bumped on every mutation and reload
+ boot = time.time() # clients detect restarts (revision resets with us)
+ # server-side degradations, shown in-browser (quarantined dismiss file,
+ # unreadable thumbnails, torn artifacts)
+ app_warnings: list[str] = list(startup_notes)
+
+ ALLOWED_HOSTS = {"127.0.0.1", "localhost", "testserver"}
+
+ @app.middleware("http")
+ async def origin_guard(request, call_next):
+ # A hostile webpage can fire preflight-free cross-origin POSTs at a
+ # localhost server (bodyless run triggers, multipart photo posts).
+ # Mutations must come from us: same-host, and no foreign Origin.
+ if request.method not in ("GET", "HEAD", "OPTIONS"):
+ host = (request.headers.get("host") or "").split(":")[0]
+ origin = request.headers.get("origin")
+ origin_host = urlsplit(origin).hostname if origin else None
+ if host not in ALLOWED_HOSTS or (
+ origin_host is not None and origin_host not in ALLOWED_HOSTS
+ ):
+ return JSONResponse(
+ {"detail": "cross-origin request refused"}, status_code=403
+ )
+ return await call_next(request)
def freshen() -> None:
"""Serve every request from the current file state: an extract or
@@ -239,7 +267,18 @@ def create_app(
available = photo_names()
sightings = []
if cfg.unidentified_path.exists():
- for photo, entries in json.loads(cfg.unidentified_path.read_text()).items():
+ try:
+ data = json.loads(cfg.unidentified_path.read_text())
+ except (json.JSONDecodeError, OSError) as err:
+ # one torn file must not 500 every page and badge at once
+ note = (
+ f"{cfg.unidentified_path.name} unreadable ({err}) — "
+ "reshoot list unavailable"
+ )
+ if note not in app_warnings:
+ app_warnings.append(note)
+ return []
+ for photo, entries in data.items():
for s in entries:
if _sighting_key(photo, s) in dismissed.keys:
continue
@@ -248,6 +287,17 @@ def create_app(
)
return sightings
+ def read_games() -> dict:
+ if not cfg.games_path.exists():
+ return {}
+ try:
+ return json.loads(cfg.games_path.read_text())
+ except (json.JSONDecodeError, OSError) as err:
+ note = f"{cfg.games_path.name} unreadable ({err}) — library unavailable"
+ if note not in app_warnings:
+ app_warnings.append(note)
+ return {}
+
def find_row(title_raw: str, source_photos: str, row_ix: int | None = None) -> dict:
freshen()
row = session.find_row(title_raw, source_photos, row_ix)
@@ -345,7 +395,8 @@ def create_app(
]
return {
"revision": revision["n"],
- "warnings": session.warnings[-10:],
+ "boot": boot,
+ "warnings": (session.warnings + app_warnings)[-10:],
"pending": [row_payload(r) for r in session.pending_rows()],
"versions": [version_payload(r) for r in session.version_rows()],
"merges": merges,
@@ -453,9 +504,7 @@ def create_app(
@app.get("/api/library")
def api_library() -> list[dict]:
- if not cfg.games_path.exists():
- return []
- games = json.loads(cfg.games_path.read_text())
+ games = read_games()
return sorted(games.values(), key=lambda g: (g.get("name") or "").casefold())
def _csv_count(path: Path) -> int:
@@ -476,10 +525,9 @@ def create_app(
import csv as _csv
log_counts = Counter(row["status"] for row in _csv.DictReader(f))
- games = 0
- if cfg.games_path.exists():
- games = len(json.loads(cfg.games_path.read_text()))
+ games = len(read_games())
return {
+ "boot": boot,
# key NAMES only — never values (credentials stay out of
# every payload and log)
"env": {
@@ -532,12 +580,13 @@ def create_app(
raise HTTPException(400, f"not a photo: {name or '(unnamed)'}")
target = cfg.photos_dir / name
data = await upload_file.read()
- target.write_bytes(data)
- # a re-uploaded photo means "re-extract this one": drop its raw
- # cache (regenerable) so the next extract run picks it up
+ # a re-uploaded photo means "re-extract this one": the raw cache
+ # goes FIRST, so a crash can never leave the new photo paired
+ # with the old photo's extraction
stale = cfg.extract_raw_dir / f"{name}.json"
if stale.exists():
stale.unlink()
+ atomic_write_bytes(target, data)
saved.append(name)
return {"saved": saved, "photos": len(photo_names())}
@@ -552,7 +601,20 @@ def create_app(
revision["n"] += 1
return _decide(body)
+ def _refuse_if_rewriting() -> None:
+ snap = jobs.snapshot()
+ if snap["status"] == "running" and snap["stage"] in ("extract", "resolve"):
+ # the job holds a start-of-run snapshot and will atomically
+ # rewrite matches.csv/titles.json at the end — a decision saved
+ # now would be silently reverted by that rewrite
+ raise HTTPException(
+ 409,
+ f"a {snap['stage']} run is rewriting the pipeline data — "
+ "wait for it to finish, then decide",
+ )
+
def _decide(body: DecisionBody) -> dict:
+ _refuse_if_rewriting()
row = find_row(body.title_raw, body.source_photos, body.row_ix)
if body.action == "pick":
candidates = json.loads(row["candidates_json"] or "[]")
@@ -579,6 +641,7 @@ def create_app(
return _version(body)
def _version(body: VersionBody) -> dict:
+ _refuse_if_rewriting()
row = find_row(body.title_raw, body.source_photos, body.row_ix)
if body.action == "unknown":
session.decide_version(row, None)
@@ -595,6 +658,7 @@ def create_app(
def api_veto_merge(body: VetoBody) -> dict:
with lock:
revision["n"] += 1
+ _refuse_if_rewriting()
row = find_row(body.title_raw, body.source_photos, body.row_ix)
if row["match_status"] != "merged":
raise HTTPException(400, "row is not merged")
diff --git a/tests/test_config.py b/tests/test_config.py
index 35aef79..dd737a1 100644
--- a/tests/test_config.py
+++ b/tests/test_config.py
@@ -8,8 +8,11 @@ from bggpipe.config import Config, load_config
def test_defaults_when_no_file(tmp_path, monkeypatch):
+ # no config.toml in cwd and no explicit path -> defaults (an EXPLICIT
+ # missing path errors instead; see the dedicated test)
monkeypatch.delenv("BGG_USERNAME", raising=False)
- cfg = load_config(tmp_path / "missing.toml")
+ monkeypatch.chdir(tmp_path)
+ cfg = load_config()
assert cfg == Config()
assert cfg.cache_dir == Path("data/bgg_cache")
@@ -44,3 +47,10 @@ def test_unknown_toml_keys_warn(tmp_path, monkeypatch):
p.write_text('photo_dir = "oops"\n')
with pytest.warns(UserWarning, match="photo_dir"):
load_config(p)
+
+
+def test_explicit_missing_config_errors_instead_of_silent_defaults(tmp_path):
+ import pytest
+
+ with pytest.raises(FileNotFoundError, match="does not exist"):
+ load_config(tmp_path / "nope.toml")
diff --git a/tests/test_diff.py b/tests/test_diff.py
index 56fe930..9f708f8 100644
--- a/tests/test_diff.py
+++ b/tests/test_diff.py
@@ -364,3 +364,36 @@ def test_live_diff_falls_back_to_snapshots_on_auth_failure(tmp_path, monkeypatch
write_matches(cfg.matches_path, [_match("5 MINUTE DUNGEON", "207830")])
result = run_diff(cfg, client=_LiveClient([], fail_auth=True))
assert result.already_owned == ["5 MINUTE DUNGEON"] # snapshots served
+
+
+def test_bare_sibling_of_confident_add_is_not_a_duplicate_upload():
+ # photo A reads "Wingspan" (confident version), photo B misreads
+ # "Wingspam" (bare) resolving to the same absent game: ONE add, not two
+ rows = [
+ _match("Wingspan", "266192", vstatus="version_auto", vid="1", vname="1st"),
+ _match("Wingspam", "266192"),
+ ]
+ result = compute_diff(rows, [])
+ assert len(result.to_add) == 1
+ assert result.already_owned == ["Wingspam"]
+
+
+def test_vetoed_bare_sibling_still_adds_for_absent_game():
+ vetoed = {**_match("Wingspan", "266192"), "dedupe_veto": "1"}
+ rows = [
+ _match("Wingspan", "266192", vstatus="version_auto", vid="1", vname="1st"),
+ vetoed,
+ ]
+ result = compute_diff(rows, [])
+ assert len(result.to_add) == 2
+ assert result.to_add[1]["second_copy"] == "1"
+
+
+def test_second_copy_flag_travels_on_exhausted_adds():
+ rows = [
+ _match("Catan", "13", vstatus="version_auto", vid="123", vname="3rd"),
+ _match("Catan", "13", vstatus="version_auto", vid="123", vname="3rd"),
+ ]
+ result = compute_diff(rows, [_item(13, 900, version_id=123)])
+ (added,) = result.to_add
+ assert added["second_copy"] == "1"
diff --git a/tests/test_extract.py b/tests/test_extract.py
index 4cf957e..601902c 100644
--- a/tests/test_extract.py
+++ b/tests/test_extract.py
@@ -317,3 +317,35 @@ def test_low_confidence_reads_are_surfaced(tmp_path, capsys):
assert "Low-confidence reads" in out
assert "'Patchwork' (medium)" in out
assert "'Catan'" not in out.split("Low-confidence reads")[1]
+
+
+def test_corrupt_raw_cache_is_reextracted_not_skipped(tmp_path):
+ cfg = _cfg(tmp_path)
+ cfg.photos_dir.mkdir()
+ _write_image(cfg.photos_dir / "shelf.jpg")
+ vision, calls = _vision_stub('[{"title_raw": "Catan", "confidence": "high"}]')
+ run_extract(cfg, vision=vision)
+ raw = cfg.extract_raw_dir / "shelf.jpg.json"
+ raw.write_text("{torn") # corrupt the cache
+ run_extract(cfg, vision=vision)
+ assert len(calls) == 2 # re-extracted, not skipped
+ json.loads(raw.read_text()) # cache healed
+
+
+def test_systemic_failures_abort_and_exit_nonzero(tmp_path):
+ import pytest
+ import typer as _typer
+
+ calls = []
+
+ def broken_vision(image_b64, media_type):
+ calls.append(1)
+ raise RuntimeError("invalid x-api-key")
+
+ cfg = _cfg(tmp_path)
+ cfg.photos_dir.mkdir()
+ for i in range(6):
+ _write_image(cfg.photos_dir / f"p{i}.jpg")
+ with pytest.raises(_typer.Exit):
+ run_extract(cfg, vision=broken_vision)
+ assert len(calls) == 3 # aborted after 3 identical failures
diff --git a/tests/test_fsio.py b/tests/test_fsio.py
new file mode 100644
index 0000000..6cb3698
--- /dev/null
+++ b/tests/test_fsio.py
@@ -0,0 +1,49 @@
+"""fsio carries the project's central resumability promise: a kill or
+crash mid-write must never leave a torn file. These pin that promise
+directly — six modules depend on it."""
+
+from __future__ import annotations
+
+import os
+import stat
+
+import pytest
+
+from bggpipe.fsio import atomic_write_bytes, atomic_write_csv, atomic_write_text
+
+
+def test_failed_write_leaves_original_intact(tmp_path, monkeypatch):
+ target = tmp_path / "artifact.json"
+ target.write_text("precious")
+ # a read-only directory makes the tmp-file write fail
+ os.chmod(tmp_path, stat.S_IRUSR | stat.S_IXUSR)
+ try:
+ with pytest.raises(OSError):
+ atomic_write_text(target, "replacement")
+ finally:
+ os.chmod(tmp_path, 0o755)
+ assert target.read_text() == "precious"
+ assert not list(tmp_path.glob("*.tmp")) # no leftovers
+
+
+def test_returned_mtime_matches_the_written_file(tmp_path):
+ target = tmp_path / "rows.csv"
+ mtime = atomic_write_csv(target, ["a", "b"], [{"a": "1", "b": "2"}])
+ # ReviewSession records this as "my own write" — it must be the mtime
+ # the file actually carries, or external-change detection breaks
+ assert mtime == target.stat().st_mtime_ns
+
+
+def test_bytes_variant_round_trips(tmp_path):
+ target = tmp_path / "photo.jpg"
+ atomic_write_bytes(target, b"\xff\xd8jpeg")
+ assert target.read_bytes() == b"\xff\xd8jpeg"
+ atomic_write_bytes(target, b"\xff\xd8jpeg2") # overwrite is atomic too
+ assert target.read_bytes() == b"\xff\xd8jpeg2"
+
+
+def test_concurrent_writers_use_distinct_tmp_names(tmp_path):
+ from bggpipe.fsio import _tmp_for
+
+ target = tmp_path / "x.csv"
+ assert _tmp_for(target) != _tmp_for(target) # no shared-inode interleave
diff --git a/tests/test_init.py b/tests/test_init.py
index 0576b12..41678c5 100644
--- a/tests/test_init.py
+++ b/tests/test_init.py
@@ -43,8 +43,8 @@ def test_creates_dirs_config_and_env_from_nothing(tmp_path, monkeypatch):
assert (tmp_path / "photos").is_dir() and (tmp_path / "data").is_dir()
assert (tmp_path / "config.toml").exists()
env = (tmp_path / ".env").read_text()
- assert "ANTHROPIC_API_KEY=sk-test-123" in env
- assert "BGG_USERNAME=eric" in env
+ assert "ANTHROPIC_API_KEY='sk-test-123'" in env
+ assert "BGG_USERNAME='eric'" in env
assert report.keys_written == ["ANTHROPIC_API_KEY", "BGG_USERNAME"]
assert set(report.keys_missing) == {"BGG_PASSWORD", "BGG_API_TOKEN"}
@@ -117,3 +117,33 @@ def test_secret_values_never_appear_in_output(tmp_path, monkeypatch, capsys):
_clear_env(monkeypatch)
_run(tmp_path, answers={"BGG_PASSWORD": "s3cret-value-xyz"})
assert "s3cret-value-xyz" not in capsys.readouterr().out
+
+
+def test_values_are_shell_quoted_for_source(tmp_path, monkeypatch):
+ # a password with spaces, $, and quotes must survive `source .env`
+ _clear_env(monkeypatch)
+ _run(tmp_path, answers={"BGG_PASSWORD": "pa$s wo'rd"})
+ env = (tmp_path / ".env").read_text()
+ assert "BGG_PASSWORD='pa$s wo'\\''rd'" in env
+
+
+def test_env_parsing_negatives(tmp_path, monkeypatch):
+ # commented, quoted-empty, and export-prefixed lines must parse sanely
+ _clear_env(monkeypatch)
+ (tmp_path / ".env").write_text(
+ "# BGG_PASSWORD=commented-out\n"
+ 'ANTHROPIC_API_KEY=""\n'
+ "export BGG_USERNAME='eric'\n"
+ )
+ report = _run(tmp_path)
+ assert "BGG_USERNAME" in report.keys_ready # export form recognized
+ assert "ANTHROPIC_API_KEY" in report.keys_missing # quoted-empty ≠ set
+ assert "BGG_PASSWORD" in report.keys_missing # comments don't count
+
+
+def test_env_file_is_owner_only_from_creation(tmp_path, monkeypatch):
+ import os as _os
+
+ _clear_env(monkeypatch)
+ _run(tmp_path, answers={"BGG_API_TOKEN": "tok-123"})
+ assert _os.stat(tmp_path / ".env").st_mode & 0o777 == 0o600
diff --git a/tests/test_resolve.py b/tests/test_resolve.py
index 3beeaff..8e0b90a 100644
--- a/tests/test_resolve.py
+++ b/tests/test_resolve.py
@@ -660,3 +660,15 @@ def test_reordered_titles_json_cannot_mispair_editions(client, tmp_path):
assert by_photos["b.jpg"]["match_status"] == "auto" # kept its resolution
assert "a.jpg" in by_photos # newcomer resolved as its own row
assert len(rows) == 2
+
+
+def test_merged_into_chains_resolve_to_terminal_survivor():
+ # X merged into Y in a prior run; this run merges Y into W — X must
+ # point at W, or diff's one-level photo hop loses X's provenance
+ x = _mrow("Wingspam", "266192", "x.jpg", status="merged")
+ x["merged_into"] = "Wingspan Typo"
+ y = _mrow("Wingspan Typo", "266192", "y.jpg", name="Wingspan")
+ w = _mrow("Wingspan", "266192", "w.jpg", name="Wingspan")
+ dedupe_matches([x, y, w], [])
+ assert y["match_status"] == "merged" and y["merged_into"] == "Wingspan"
+ assert x["merged_into"] == "Wingspan" # chain collapsed
diff --git a/tests/test_upload.py b/tests/test_upload.py
index ce7c025..aa9d818 100644
--- a/tests/test_upload.py
+++ b/tests/test_upload.py
@@ -466,3 +466,28 @@ def test_run_upload_verify_wiring(tmp_path, capsys):
)
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
diff --git a/tests/test_web_dashboard.py b/tests/test_web_dashboard.py
index f7b42a5..48a3e5f 100644
--- a/tests/test_web_dashboard.py
+++ b/tests/test_web_dashboard.py
@@ -4,6 +4,7 @@ stage functions are injected so nothing slow or networked ever runs."""
from __future__ import annotations
import threading
+import time
import typer
from fastapi.testclient import TestClient
@@ -315,3 +316,308 @@ def test_non_photo_files_are_invisible(tmp_path):
web = _app(cfg)
assert [p["name"] for p in web.get("/api/photos-list").json()] == ["shelf.jpg"]
assert web.get("/photos/.DS_Store").status_code == 404
+
+
+# -- jobs: live-progress and termination contracts ----------------------
+
+
+def test_running_snapshot_shows_partial_line_then_finishes():
+ import typer as _typer
+
+ runner = JobRunner()
+ release = threading.Event()
+
+ def stage():
+ print("progress: 40%", end="", flush=True) # no newline yet
+ release.wait()
+ _typer.echo(" done")
+
+ runner.start("extract", stage)
+ for _ in range(100):
+ snap = runner.snapshot()
+ if snap["log"]:
+ break
+ time.sleep(0.01)
+ assert snap["status"] == "running"
+ assert snap["log"] == ["progress: 40%"]
+ release.set()
+ runner.wait()
+ assert runner.snapshot()["log"] == ["progress: 40% done"]
+
+
+def test_log_serves_last_200_lines_and_buffer_is_bounded():
+ import typer as _typer
+
+ from bggpipe.jobs import MAX_LOG_LINES
+
+ runner = JobRunner()
+
+ def stage():
+ for i in range(MAX_LOG_LINES + 300):
+ _typer.echo(f"line {i}")
+
+ runner.start("extract", stage)
+ runner.wait()
+ log = runner.snapshot()["log"]
+ assert len(log) == 200
+ assert log[-1] == f"line {MAX_LOG_LINES + 299}"
+
+
+def test_zero_exit_codes_count_as_done():
+ import typer as _typer
+
+ for exc in (_typer.Exit(), SystemExit(0)):
+ runner = JobRunner()
+ runner.start("diff", lambda exc=exc: (_ for _ in ()).throw(exc))
+ runner.wait()
+ assert runner.snapshot()["status"] == "done"
+
+
+def test_nonzero_systemexit_is_failed():
+ runner = JobRunner()
+ runner.start("diff", lambda: (_ for _ in ()).throw(SystemExit(2)))
+ runner.wait()
+ snap = runner.snapshot()
+ assert snap["status"] == "failed" and "2" in snap["error"]
+
+
+def test_base_exception_cannot_wedge_the_runner():
+ class Rude(BaseException):
+ pass
+
+ runner = JobRunner()
+ runner.start("upload", lambda: (_ for _ in ()).throw(Rude("greenlet died")))
+ runner.wait()
+ snap = runner.snapshot()
+ assert snap["status"] == "failed"
+ assert "Rude" in snap["error"]
+ assert any("Traceback" in line for line in snap["log"]) # diagnosable
+ assert runner.start("extract", lambda: None) # slot is free again
+ runner.wait()
+
+
+# -- concurrency and restart contracts ----------------------------------
+
+
+def test_revision_bumps_on_mutations_not_reads(tmp_path):
+ from bggpipe.resolve import write_matches
+
+ cfg = _cfg(tmp_path)
+ write_matches(
+ cfg.matches_path,
+ [
+ {
+ "title_raw": "Mystery",
+ "bgg_id": "",
+ "bgg_name": "",
+ "year": "",
+ "type": "",
+ "match_status": "unmatched",
+ "version_id": "",
+ "version_name": "",
+ "version_status": "",
+ "candidates_json": "[]",
+ "version_candidates_json": "[]",
+ "source_photos": "x.jpg",
+ }
+ ],
+ )
+ web = _app(cfg)
+ first = web.get("/api/state").json()
+ assert "boot" in first # clients detect restarts by boot change
+ second = web.get("/api/state").json()
+ assert second["revision"] == first["revision"] # reads never bump
+
+ res = web.post(
+ "/api/decision",
+ json={"title_raw": "Mystery", "source_photos": "x.jpg", "action": "reject"},
+ )
+ assert res.json()["revision"] > first["revision"] # mutations bump
+
+ # external rewrite bumps on the next read
+ rows = []
+ write_matches(cfg.matches_path, rows)
+ bumped = web.get("/api/state").json()
+ assert bumped["revision"] > res.json()["revision"]
+
+
+def test_torn_data_files_degrade_with_warnings_not_500(tmp_path):
+ cfg = _cfg(tmp_path)
+ cfg.unidentified_path.write_text("{torn")
+ cfg.games_path.write_text("[not even close")
+ web = _app(cfg)
+ state = web.get("/api/state")
+ pipeline = web.get("/api/pipeline")
+ library = web.get("/api/library")
+ assert state.status_code == pipeline.status_code == library.status_code == 200
+ assert library.json() == []
+ assert any("unreadable" in w for w in state.json()["warnings"])
+
+
+def test_decisions_locked_out_while_matches_rewriting_job_runs(tmp_path):
+ from bggpipe.resolve import read_matches, write_matches
+
+ cfg = _cfg(tmp_path)
+ write_matches(
+ cfg.matches_path,
+ [
+ {
+ "title_raw": "Mystery",
+ "bgg_id": "",
+ "bgg_name": "",
+ "year": "",
+ "type": "",
+ "match_status": "unmatched",
+ "version_id": "",
+ "version_name": "",
+ "version_status": "",
+ "candidates_json": "[]",
+ "version_candidates_json": "[]",
+ "source_photos": "x.jpg",
+ }
+ ],
+ )
+ release = threading.Event()
+ jobs = JobRunner()
+ web = _app(cfg, stages={"resolve": release.wait}, jobs=jobs)
+ web.post("/api/run/resolve")
+ res = web.post(
+ "/api/decision",
+ json={"title_raw": "Mystery", "source_photos": "x.jpg", "action": "reject"},
+ )
+ assert res.status_code == 409
+ assert "resolve run is rewriting" in res.json()["detail"]
+ # the decision was NOT applied
+ assert read_matches(cfg.matches_path)[0]["match_status"] == "unmatched"
+ release.set()
+ jobs.wait()
+
+
+def test_cross_origin_mutations_are_refused(tmp_path):
+ web = _app(_cfg(tmp_path))
+ # foreign Origin on a same-host request: refused
+ res = web.post("/api/run/extract", headers={"origin": "https://evil.example"})
+ assert res.status_code == 403
+ # foreign Host (DNS rebinding): refused
+ res = web.post(
+ "/api/photos",
+ headers={"host": "evil.example"},
+ files={"files": ("a.jpg", b"x", "image/jpeg")},
+ )
+ assert res.status_code == 403
+ # reads are unaffected
+ assert (
+ web.get("/api/pipeline", headers={"origin": "https://evil.example"}).status_code
+ == 200
+ )
+
+
+def test_concurrent_decisions_and_polls_never_lose_a_decision(tmp_path):
+ from bggpipe.resolve import read_matches, write_matches
+
+ cfg = _cfg(tmp_path)
+ base = {
+ "bgg_id": "",
+ "bgg_name": "",
+ "year": "",
+ "type": "",
+ "match_status": "unmatched",
+ "version_id": "",
+ "version_name": "",
+ "version_status": "",
+ "candidates_json": "[]",
+ "version_candidates_json": "[]",
+ "source_photos": "x.jpg",
+ }
+ rows = [{**base, "title_raw": f"Game{i}"} for i in range(8)]
+ write_matches(cfg.matches_path, rows)
+ web = _app(cfg)
+
+ def decide(i):
+ web.post(
+ "/api/decision",
+ json={
+ "title_raw": f"Game{i}",
+ "source_photos": "x.jpg",
+ "action": "reject",
+ },
+ )
+
+ def poll():
+ web.get("/api/state")
+
+ threads = [threading.Thread(target=decide, args=(i,)) for i in range(8)]
+ threads += [threading.Thread(target=poll) for _ in range(8)]
+ for t in threads:
+ t.start()
+ for t in threads:
+ t.join()
+ saved = read_matches(cfg.matches_path)
+ assert all(r["match_status"] == "rejected" for r in saved)
+
+
+def test_decision_for_vanished_row_is_404_and_touches_nothing(tmp_path):
+ from bggpipe.resolve import read_matches, write_matches
+
+ cfg = _cfg(tmp_path)
+ row = {
+ "title_raw": "Kept",
+ "bgg_id": "",
+ "bgg_name": "",
+ "year": "",
+ "type": "",
+ "match_status": "unmatched",
+ "version_id": "",
+ "version_name": "",
+ "version_status": "",
+ "candidates_json": "[]",
+ "version_candidates_json": "[]",
+ "source_photos": "x.jpg",
+ }
+ write_matches(cfg.matches_path, [row])
+ web = _app(cfg)
+ res = web.post(
+ "/api/decision",
+ json={"title_raw": "Gone", "source_photos": "x.jpg", "action": "reject"},
+ )
+ assert res.status_code == 404
+ assert read_matches(cfg.matches_path)[0]["match_status"] == "unmatched"
+
+
+def test_pipeline_reports_badge_fields(tmp_path):
+ from bggpipe.resolve import write_matches
+
+ cfg = _cfg(tmp_path)
+ base = {
+ "bgg_id": "13",
+ "bgg_name": "Catan",
+ "year": "",
+ "type": "boardgame",
+ "version_id": "",
+ "version_name": "",
+ "candidates_json": "[]",
+ "version_candidates_json": "[]",
+ "source_photos": "x.jpg",
+ }
+ write_matches(
+ cfg.matches_path,
+ [
+ {
+ **base,
+ "title_raw": "A",
+ "match_status": "ambiguous",
+ "version_status": "",
+ },
+ {
+ **base,
+ "title_raw": "B",
+ "match_status": "auto",
+ "version_status": "version_ambiguous",
+ "version_id": "1",
+ },
+ ],
+ )
+ cfg.to_add_path.write_text("bgg_id,bgg_name\n1,X\n2,Y\n")
+ p = _app(cfg).get("/api/pipeline").json()
+ assert p["pending_review"] == 2 # one match + one edition decision
+ assert p["to_add"] == 2 # header excluded