Eric Wagoner
33cbfffbd6
/photos/view/{name} shows the photo large in the shell, every title
read from it with status chips and BGG matches, its reshoot tickets
with working dismiss, prev/next navigation with arrow keys, position
in the gallery, and a link to the raw full-size file. Gallery and
catalog photo links point here now (review's shots keep linking to the
raw image — zooming spine text is their whole purpose). The status
chip renderer moves to app.js so the catalog and photo pages can't
drift; render_page learns an `active` override so a detail page keeps
its nav section lit.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
636 lines
19 KiB
Python
636 lines
19 KiB
Python
"""Dashboard/API tests: job lifecycle, pipeline status, photo upload —
|
|
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
|
|
|
|
from bggpipe.config import Config
|
|
from bggpipe.jobs import JobRunner
|
|
from bggpipe.webreview import create_app
|
|
|
|
|
|
def _cfg(tmp_path) -> Config:
|
|
cfg = Config(data_dir=tmp_path / "data", photos_dir=tmp_path / "photos")
|
|
cfg.photos_dir.mkdir(parents=True)
|
|
cfg.data_dir.mkdir(parents=True)
|
|
return cfg
|
|
|
|
|
|
def _app(cfg, stages=None, jobs=None) -> TestClient:
|
|
return TestClient(create_app(cfg, stages=stages or {}, jobs=jobs))
|
|
|
|
|
|
# -- JobRunner ----------------------------------------------------------
|
|
|
|
|
|
def test_job_captures_output_and_finishes():
|
|
runner = JobRunner()
|
|
|
|
def stage():
|
|
typer.echo("line one")
|
|
typer.echo("line two")
|
|
|
|
assert runner.start("extract", stage)
|
|
runner.wait()
|
|
snap = runner.snapshot()
|
|
assert snap["status"] == "done"
|
|
assert snap["log"] == ["line one", "line two"]
|
|
|
|
|
|
def test_job_failure_is_reported_not_swallowed():
|
|
runner = JobRunner()
|
|
runner.start("resolve", lambda: (_ for _ in ()).throw(RuntimeError("boom")))
|
|
runner.wait()
|
|
snap = runner.snapshot()
|
|
assert snap["status"] == "failed"
|
|
assert "RuntimeError: boom" in snap["error"]
|
|
|
|
|
|
def test_typer_exit_code_counts_as_failure():
|
|
runner = JobRunner()
|
|
|
|
def stage():
|
|
raise typer.Exit(code=1)
|
|
|
|
runner.start("diff", stage)
|
|
runner.wait()
|
|
assert runner.snapshot()["status"] == "failed"
|
|
|
|
|
|
def test_single_slot_rejects_second_job():
|
|
runner = JobRunner()
|
|
release = threading.Event()
|
|
assert runner.start("extract", release.wait)
|
|
assert not runner.start("resolve", lambda: None) # slot busy
|
|
release.set()
|
|
runner.wait()
|
|
assert runner.start("resolve", lambda: None) # slot free again
|
|
runner.wait()
|
|
|
|
|
|
# -- /api/run + /api/job ------------------------------------------------
|
|
|
|
|
|
def test_run_stage_lifecycle_via_api(tmp_path):
|
|
cfg = _cfg(tmp_path)
|
|
ran = []
|
|
jobs = JobRunner()
|
|
web = _app(cfg, stages={"extract": lambda: ran.append(1)}, jobs=jobs)
|
|
|
|
res = web.post("/api/run/extract")
|
|
assert res.status_code == 200
|
|
jobs.wait()
|
|
assert ran == [1]
|
|
assert web.get("/api/job").json()["status"] == "done"
|
|
|
|
|
|
def test_unknown_stage_404s(tmp_path):
|
|
web = _app(_cfg(tmp_path))
|
|
assert web.post("/api/run/frobnicate").status_code == 404
|
|
|
|
|
|
def test_busy_runner_409s(tmp_path):
|
|
cfg = _cfg(tmp_path)
|
|
release = threading.Event()
|
|
jobs = JobRunner()
|
|
web = _app(
|
|
cfg, stages={"extract": release.wait, "resolve": lambda: None}, jobs=jobs
|
|
)
|
|
assert web.post("/api/run/extract").status_code == 200
|
|
assert web.post("/api/run/resolve").status_code == 409
|
|
release.set()
|
|
jobs.wait()
|
|
|
|
|
|
def test_upload_defaults_to_dry_run(tmp_path):
|
|
cfg = _cfg(tmp_path)
|
|
calls = []
|
|
jobs = JobRunner()
|
|
|
|
def upload(dry_run=True, limit=None):
|
|
calls.append({"dry_run": dry_run, "limit": limit})
|
|
|
|
web = _app(cfg, stages={"upload": upload}, jobs=jobs)
|
|
web.post("/api/run/upload") # no body: the safe direction
|
|
jobs.wait()
|
|
web.post("/api/run/upload", json={"dry_run": False, "limit": 2})
|
|
jobs.wait()
|
|
assert calls == [
|
|
{"dry_run": True, "limit": None},
|
|
{"dry_run": False, "limit": 2},
|
|
]
|
|
|
|
|
|
# -- /api/pipeline ------------------------------------------------------
|
|
|
|
|
|
def test_pipeline_reports_counts_and_never_values(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("BGG_USERNAME", "supersecretname")
|
|
monkeypatch.delenv("BGG_API_TOKEN", raising=False)
|
|
cfg = _cfg(tmp_path)
|
|
(cfg.photos_dir / "a.jpg").write_bytes(b"x")
|
|
web = _app(cfg)
|
|
payload = web.get("/api/pipeline").json()
|
|
assert payload["photos"] == 1
|
|
assert payload["env"]["BGG_USERNAME"] is True
|
|
assert payload["env"]["BGG_API_TOKEN"] is False
|
|
assert "supersecretname" not in web.get("/api/pipeline").text
|
|
|
|
|
|
def test_pipeline_flags_stub_data(tmp_path):
|
|
cfg = _cfg(tmp_path)
|
|
(cfg.data_dir / "STUB_DATA.marker").write_text("stub")
|
|
assert _app(cfg).get("/api/pipeline").json()["stub_data"] is True
|
|
|
|
|
|
# -- /api/photos --------------------------------------------------------
|
|
|
|
|
|
def test_photo_upload_saves_and_invalidates_raw_cache(tmp_path):
|
|
cfg = _cfg(tmp_path)
|
|
cfg.extract_raw_dir.mkdir(parents=True)
|
|
stale = cfg.extract_raw_dir / "shelf.jpg.json"
|
|
stale.write_text("{}")
|
|
web = _app(cfg)
|
|
res = web.post(
|
|
"/api/photos", files={"files": ("shelf.jpg", b"\xff\xd8jpegdata", "image/jpeg")}
|
|
)
|
|
assert res.status_code == 200
|
|
assert (cfg.photos_dir / "shelf.jpg").read_bytes() == b"\xff\xd8jpegdata"
|
|
assert not stale.exists() # re-upload means re-extract
|
|
|
|
|
|
def test_photo_upload_rejects_non_photos_and_path_tricks(tmp_path):
|
|
cfg = _cfg(tmp_path)
|
|
web = _app(cfg)
|
|
res = web.post("/api/photos", files={"files": ("notes.txt", b"hi", "text/plain")})
|
|
assert res.status_code == 400
|
|
res = web.post(
|
|
"/api/photos",
|
|
files={"files": ("../../escape.jpg", b"x", "image/jpeg")},
|
|
)
|
|
if res.status_code == 200: # client may strip the path; the name must be bare
|
|
assert (cfg.photos_dir / "escape.jpg").exists()
|
|
assert not (tmp_path / "escape.jpg").exists()
|
|
|
|
|
|
# -- pages --------------------------------------------------------------
|
|
|
|
|
|
def test_dashboard_and_review_pages_serve(tmp_path):
|
|
web = _app(_cfg(tmp_path))
|
|
assert "Pipeline" in web.get("/").text
|
|
assert "bggpipe" in web.get("/review").text
|
|
|
|
|
|
# -- design system + navigation -----------------------------------------
|
|
|
|
|
|
def test_stylesheet_is_served_and_linked_by_both_pages(tmp_path):
|
|
web = _app(_cfg(tmp_path))
|
|
css = web.get("/static/app.css")
|
|
assert css.status_code == 200
|
|
assert css.headers["content-type"].startswith("text/css")
|
|
assert "--accent" in css.text # the token layer, not an empty file
|
|
for path in ("/", "/review"):
|
|
assert 'href="/static/app.css"' in web.get(path).text
|
|
|
|
|
|
def test_both_pages_carry_navigation_and_skip_link(tmp_path):
|
|
web = _app(_cfg(tmp_path))
|
|
for path, current in (("/", 'href="/"'), ("/review", 'href="/review"')):
|
|
html = web.get(path).text
|
|
assert 'nav aria-label="Primary"' in html
|
|
assert f'<a {current} aria-current="page"' in html.replace("\n", " ") or (
|
|
current in html and 'aria-current="page"' in html
|
|
)
|
|
assert 'class="skip"' in html
|
|
|
|
|
|
def test_activity_region_announces_politely(tmp_path):
|
|
html = _app(_cfg(tmp_path)).get("/").text
|
|
assert 'aria-live="polite"' in html
|
|
|
|
|
|
# -- six-page shell -----------------------------------------------------
|
|
|
|
|
|
def test_every_page_serves_with_shared_shell(tmp_path):
|
|
web = _app(_cfg(tmp_path))
|
|
for path, marker in (
|
|
("/", "Pipeline"),
|
|
("/photos", "Reshoot"),
|
|
("/review", "Review"),
|
|
("/catalog", "Catalog"),
|
|
("/queue", "Upload queue"),
|
|
("/library", "Library"),
|
|
):
|
|
html = web.get(path).text
|
|
assert marker in html, path
|
|
assert 'nav aria-label="Primary"' in html, path
|
|
assert 'aria-current="page"' in html, path
|
|
assert "logo-full.jpg" in html, path # Juniper's portrait in the rail
|
|
assert "art by Juniper" in html, path
|
|
assert "trademarks of" in html, path # attribution notice
|
|
|
|
|
|
def test_photos_list_reports_extraction_state(tmp_path):
|
|
import json as _json
|
|
|
|
cfg = _cfg(tmp_path)
|
|
(cfg.photos_dir / "done.jpg").write_bytes(b"x")
|
|
(cfg.photos_dir / "fresh.jpg").write_bytes(b"x")
|
|
cfg.extract_raw_dir.mkdir(parents=True)
|
|
(cfg.extract_raw_dir / "done.jpg.json").write_text(
|
|
_json.dumps({"titles": [{"title_raw": "Catan"}], "unidentified": [{}, {}]})
|
|
)
|
|
listing = {p["name"]: p for p in _app(cfg).get("/api/photos-list").json()}
|
|
assert listing["done.jpg"] == {
|
|
"name": "done.jpg",
|
|
"extracted": True,
|
|
"titles": 1,
|
|
"unidentified": 2,
|
|
}
|
|
assert listing["fresh.jpg"]["extracted"] is False
|
|
|
|
|
|
def test_queue_endpoint_serves_all_three_ledgers(tmp_path):
|
|
cfg = _cfg(tmp_path)
|
|
(cfg.to_add_path).write_text("bgg_id,bgg_name\n13,Catan\n")
|
|
q = _app(cfg).get("/api/queue").json()
|
|
assert q["to_add"] == [{"bgg_id": "13", "bgg_name": "Catan"}]
|
|
assert q["to_update"] == [] and q["log"] == []
|
|
|
|
|
|
def test_library_serves_games_sorted_or_empty(tmp_path):
|
|
import json as _json
|
|
|
|
cfg = _cfg(tmp_path)
|
|
web = _app(cfg)
|
|
assert web.get("/api/library").json() == []
|
|
cfg.games_path.write_text(
|
|
_json.dumps(
|
|
{
|
|
"13": {"name": "Catan", "year": 1995},
|
|
"266192": {"name": "Wingspan", "year": 2019},
|
|
"1": {"name": "aliens", "year": 2000},
|
|
}
|
|
)
|
|
)
|
|
names = [g["name"] for g in web.get("/api/library").json()]
|
|
assert names == ["aliens", "Catan", "Wingspan"] # casefold sort
|
|
|
|
|
|
def test_pipeline_reports_reshoot_count(tmp_path):
|
|
import json as _json
|
|
|
|
cfg = _cfg(tmp_path)
|
|
cfg.unidentified_path.write_text(
|
|
_json.dumps({"a.jpg": [{"location": "top shelf"}]})
|
|
)
|
|
assert _app(cfg).get("/api/pipeline").json()["reshoot"] == 1
|
|
|
|
|
|
def test_photos_list_tolerates_bare_array_raw_cache(tmp_path):
|
|
import json as _json
|
|
|
|
cfg = _cfg(tmp_path)
|
|
(cfg.photos_dir / "old.jpg").write_bytes(b"x")
|
|
cfg.extract_raw_dir.mkdir(parents=True)
|
|
(cfg.extract_raw_dir / "old.jpg.json").write_text(
|
|
_json.dumps([{"title_raw": "Catan"}, {"title_raw": "Risk"}])
|
|
)
|
|
(item,) = _app(cfg).get("/api/photos-list").json()
|
|
assert item["extracted"] is True and item["titles"] == 2
|
|
|
|
|
|
def test_non_photo_files_are_invisible(tmp_path):
|
|
cfg = _cfg(tmp_path)
|
|
(cfg.photos_dir / "shelf.jpg").write_bytes(b"x")
|
|
(cfg.photos_dir / ".DS_Store").write_bytes(b"junk")
|
|
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
|
|
|
|
|
|
def test_photo_detail_page_serves_with_photos_nav_active(tmp_path):
|
|
web = _app(_cfg(tmp_path))
|
|
html = web.get("/photos/view/shelf.jpg").text
|
|
assert "all photos" in html
|
|
# the Photos nav entry stays highlighted on the detail page
|
|
import re
|
|
|
|
(current,) = re.findall(r'<a href="([^"]+)" aria-current="page"', html)
|
|
assert current == "/photos"
|
|
assert 'src="/static/app.js"' in html
|