| |
| """Checks for the publish gate (the curator's rule, 2026-09-29: nothing publishes |
| anywhere without his approval; everything else is drafted and ready to go). |
| Offline: git and the network are stubbed; every write lands in one temporary |
| directory that is removed at the end, pass or fail. Run from the repo root: |
| .venv/bin/python intake/scripts/test_publish_gate.py |
| |
| Covers: the registry, the public-review stage and the push all refuse without |
| approval and save the entry as ready to publish; register --live carries the |
| approval; the Zenodo watcher holds a newly published record, pings the curator |
| once, and never tells the author; the Zenodo-community accept holds before it |
| touches the site; the legacy DOI-route accept returns nothing to publish; no |
| service unit or cron line sets the approval. |
| """ |
| from __future__ import annotations |
|
|
| import inspect |
| import json |
| import os |
| import shutil |
| import subprocess |
| import sys |
| import tempfile |
| from pathlib import Path |
|
|
| ROOT = Path(__file__).resolve().parents[2] |
| sys.path.insert(0, str(ROOT)) |
|
|
| failures: list[str] = [] |
|
|
|
|
| def check(cond: bool, name: str) -> None: |
| print((" ok " if cond else " FAIL ") + name) |
| if not cond: |
| failures.append(name) |
|
|
|
|
| os.environ.pop("ICSAC_PUBLISH_APPROVED", None) |
| tmp = Path(tempfile.mkdtemp(prefix="publish-gate-")) |
| try: |
| import publications as pub |
|
|
| site = tmp / "site" |
| (site / "src" / "data").mkdir(parents=True) |
| reg = site / "src" / "data" / "accepted.json" |
| reg.write_text("[]") |
| pub.WEBSITE_REPO = str(site) |
| pub.REGISTRY_PATH = str(reg) |
| pub.READY_DIR = str(tmp / "ready") |
| git_calls: list = [] |
| real_run = subprocess.run |
| pub.subprocess.run = lambda cmd, **k: (git_calls.append(cmd), subprocess.CompletedProcess(cmd, 0, "", ""))[1] |
|
|
| proto = {"title": "A held paper", "authors": ["A. Author"], "doi": "10.5555/icsac.2026.990", |
| "source": "submission-pdf", "sub_id": "ICSAC-SUB-00990"} |
|
|
| print("1. the writers refuse without approval") |
| check(pub.upsert_entry(dict(proto)) == {}, "registry: no entry without approval") |
| check(json.loads(reg.read_text()) == [], "registry: accepted.json untouched") |
| ready = Path(pub.READY_DIR) / "ICSAC-SUB-00990.json" |
| check(ready.exists() and json.loads(ready.read_text())["proto"]["doi"] == proto["doi"], |
| "the entry is saved as ready to publish") |
| check(pub.stage_public_review_for_slug("ICSAC-SUB-00990", "a-held-paper", str(tmp)) == (None, None) |
| and not (site / "src" / "data" / "public-reviews").exists(), "public review: nothing staged") |
| pub.commit_and_push("publications: held") |
| check(git_calls == [], "push: no git command runs") |
|
|
| print("2. approval opens it") |
| e = pub.upsert_entry(dict(proto), approved=True) |
| check(bool(e.get("slug")) and len(json.loads(reg.read_text())) == 1, "approved=True writes the entry") |
| pub.commit_and_push("publications: approved", approved=True) |
| check(any(c[:2] == ("git", "add") for c in git_calls), "approved=True pushes") |
| os.environ["ICSAC_PUBLISH_APPROVED"] = "1" |
| check(pub.publish_approved(), "a hand-run with ICSAC_PUBLISH_APPROVED=1 is approved") |
| os.environ.pop("ICSAC_PUBLISH_APPROVED") |
| check(not pub.publish_approved(), "without it, not approved") |
|
|
| print("3. register --live is the approved path") |
| import crossref_deposit as cd |
| src = inspect.getsource(cd._push_publications) |
| check(src.count("approved=True") == 3, "_push_publications passes approval to all three writers") |
|
|
| print("4. the Zenodo watcher holds, pings once, tells no author") |
| import publish_watcher as pw |
| sub_dir = tmp / "subs" / "ICSAC-SUB-00991" |
| sub_dir.mkdir(parents=True) |
| (sub_dir / "state.json").write_text(json.dumps({"deposit_record_id": "991"})) |
| (sub_dir / "submission.json").write_text(json.dumps({"title": "W", "creators": [{"name": "Doe, Jo"}], |
| "form": {"email": "a@example.com", "name": "Jo Doe"}})) |
| pw._list_awaiting_publish = lambda: [sub_dir] |
| pw._get_deposit = lambda rid: {"submitted": True, "state": "done", "doi": "10.5281/zenodo.991", |
| "record_id": 991, "metadata": {"title": "W"}} |
| told = [] |
| pw._notify_author_published = lambda *a, **k: told.append(a) |
| import notify |
| pings = [] |
| notify.send_to_curator = lambda msg, **k: pings.append(msg) |
| s1 = pw.poll_drafts() |
| st = json.loads((sub_dir / "state.json").read_text()) |
| check(s1.get("held") == 1 and s1.get("published") == 0, "the record is held, not published") |
| check(len(pings) == 1 and "READY TO PUBLISH" in pings[0], "the curator is told once") |
| check(told == [] and st.get("state") != "published" and st.get("publish_held_at"), |
| "no author email, state not marked published") |
| s2 = pw.poll_drafts() |
| check(s2.get("held") == 1 and len(pings) == 1, "the next tick stays quiet") |
|
|
| print("5. the Zenodo-community accept holds before touching the site") |
| import action |
| action._fetch_record = lambda rid: {"metadata": {}} |
| action._extract_registry_entry = lambda rid, md, source: {"title": "C", "authors": ["C"], "doi": "10.5281/zenodo.992", |
| "source": source, "record_id": rid} |
| touched = [] |
| action._publish_public_review = lambda rid: touched.append("review") |
| action._refresh_panel_stats = lambda: touched.append("stats") |
| before = reg.read_text() |
| action.register_accepted_paper("992") |
| check(touched == [] and reg.read_text() == before, "no review, no stats, no registry write") |
| check((Path(pub.READY_DIR) / "992.json").exists() and any("992" in p for p in pings), |
| "saved as ready to publish, and the curator is told") |
|
|
| print("6. the legacy DOI-route accept") |
| from intake import submission_worker as sw |
| src = inspect.getsource(sw._register_doi_accept) |
| check("if not entry:" in src and "return None" in src, "a held entry returns nothing to publish") |
|
|
| print("7. nothing automatic carries the approval") |
| units = list(Path("/etc/systemd/system").glob("*.service")) |
| check(all("ICSAC_PUBLISH_APPROVED" not in u.read_text(errors="replace") for u in units), |
| f"no service unit sets ICSAC_PUBLISH_APPROVED ({len(units)} units read)") |
| try: |
| cron = real_run(["crontab", "-l"], capture_output=True, text=True, timeout=10).stdout |
| except Exception: |
| cron = "" |
| check("ICSAC_PUBLISH_APPROVED" not in cron, "no cron line sets it") |
| finally: |
| shutil.rmtree(tmp, ignore_errors=True) |
|
|
| print() |
| if failures: |
| print(f"FAILED: {len(failures)}") |
| for f in failures: |
| print(" - " + f) |
| sys.exit(1) |
| print("ALL GREEN") |
|
|
