intake/scripts/test_code_data_link.py

#!/usr/bin/env python3
"""Checks for the code and data link policy (2026-09-28). Offline: no network,
no mail, no pings; every write lands in one temporary directory that is
removed at the end, pass or fail. Run from the repo root with the venv:
    .venv/bin/python intake/scripts/test_code_data_link.py

Covers: the claim check on real papers and on made-up sentences; the form
answer through the real /api/submit handler (signed like the proxy signs it);
the worker holding a paper before the panel and the curator override; the
revise email the hold produces; the accept guard; the Zenodo relation; the
registry field.
"""
from __future__ import annotations

import glob
import hashlib
import hmac
import json
import pathlib
import shutil
import sys
import tempfile
import time
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)


PAPER_PDF = Path.home() / "icsac-submissions" / "ICSAC-SUB-00008" / "paper.pdf"
PAPERS = Path.home() / "Desktop" / "icsac" / "icsacinstitute.org" / "public" / "papers"
TEST_ORCID = "0000-0002-1825-0097"

tmp = Path(tempfile.mkdtemp(prefix="codelink-"))
real_home = pathlib.Path.home
try:
    import code_availability as ca
    import submission_intake as ingest

    # ── 1. the claim check ───────────────────────────────────────────────────
    print("1. claim check")
    vim_text = ingest.extract_pdf_text(str(PAPER_PDF))
    r = ca.assess({}, vim_text)
    check(r["missing_link"], "the paper: claims code/data, no link anywhere -> held")
    check(any("accompanying archive" in c for c in r["claims"]), "the paper: the archive sentence is quoted")
    link = "https://github.com/example/generator"
    check(not ca.assess({"code_data": {"available": True, "url": link}}, vim_text)["missing_link"],
          "the paper + form link -> not held")
    check(not ca.assess({"related_identifiers": [{"identifier": link, "relation": "isSupplementedBy"}]},
                        vim_text)["missing_link"], "the paper + 'is supplemented by' identifier -> not held")
    check(not ca.assess({"related_identifiers": [{"identifier": link, "relation": "isSupplementTo"}]},
                        vim_text)["missing_link"], "the paper + older 'is supplement to' identifier -> not held")
    check(ca.assess({"related_identifiers": [{"identifier": link, "relation": "cites"}]},
                    vim_text)["missing_link"], "the paper + a 'cites' identifier -> still held")

    na = "Methods.\nWe prove a theorem.\n\nData availability: Not applicable.\n\nReferences\nSmith, J. (2020)."
    check(ca.find_claims(ca.body_text(na)) == [], "'Data availability: Not applicable.' is no claim")
    none = "Data availability statement\nNo new data were generated or analysed in this study."
    check(ca.find_claims(none) == [], "'No new data were generated' is no claim")
    plain = "The data show a clear trend, and the code of conduct was followed. Figure 2 plots the data."
    check(ca.find_claims(plain) == [], "ordinary mentions of data and code are no claim")
    req = "Results.\nData are available on request from the author.\n"
    check(ca.assess({}, req)["missing_link"], "'available on request' with no link -> held")
    body_link = "Code availability. The scripts are available at https://github.com/a/b for anyone."
    check(not ca.assess({}, body_link)["missing_link"], "claim with a repository link in the body -> not held")
    refs_only = ("Code availability. The code is available from the author.\n\nReferences\n"
                 "Doe, A. (2021). Tool. https://github.com/doe/tool\n")
    check(ca.assess({}, refs_only)["missing_link"], "a link only in the reference list does not count")
    gap = "Complete source code and validated data files are available in the GitHub Repository."
    check(len(ca.find_claims(gap)) == 1, "words between 'code' and 'are available' still read as a claim")
    for p in sorted(glob.glob(str(PAPERS / "*.pdf"))):
        if p.endswith("-icsac.pdf") or not Path(p).stem.isdigit():
            continue   # founding papers are the Zenodo-record PDFs; intake papers below
        rr = ca.assess({}, ingest.extract_pdf_text(p))
        check(not rr["missing_link"], f"founding paper {Path(p).stem}: not held "
              f"({len(rr['claims'])} claim(s), {len(rr['links']['in_paper'])} link(s))")
    # The first intake paper (2026-09-29): its Declarations promise an archive with no
    # link in the text; the link came from the author's reply and rides on code_data.
    vim = PAPERS / "ICSAC-SUB-00008.pdf"
    if vim.exists():
        vt = ingest.extract_pdf_text(str(vim))
        check(ca.assess({}, vt)["missing_link"], "the published paper on its text alone: held (the claim has no link)")
        check(not ca.assess({"code_data": {"available": True, "url": "https://doi.org/10.5281/zenodo.23042786"}}, vt)["missing_link"],
              "the published paper with its author's archive link: not held")

    # ── 2. the form answer through the real handler ──────────────────────────
    print("2. /api/submit (T2, signed like the proxy)")
    import httpx
    from fastapi.testclient import TestClient
    from intake import intake_server as iss
    iss.HMAC_SECRET = b"codelink-test-secret"
    iss.TEST_ORCID_WHITELIST = frozenset({TEST_ORCID})
    subs = tmp / "subs"
    iss.SUBMISSIONS_ROOT = subs
    iss.QUEUE_DIR = subs / "queue"
    iss.COUNTER_FILE = subs / ".counter"
    iss.TEST_SUBMISSIONS_ROOT = subs / "test"
    iss.TEST_QUEUE_DIR = subs / "test" / "queue"
    audit_iss: list[dict] = []
    iss._audit_append = lambda entry, test_mode=False: audit_iss.append(entry)
    _n = iter(range(1790000000, 1790000100))
    iss._allocate_test_sub_id = lambda: f"ICSAC-SUB-TEST-{next(_n)}"
    client = TestClient(iss.app)
    pdf_bytes = PAPER_PDF.read_bytes()

    def submit(**extra) -> tuple[int, dict]:
        data = {
            "mode": "upload", "name": "Test Author", "email": "author@example.com",
            "orcid": TEST_ORCID, "coi": "true", "exclusivity_acknowledged": "true",
            "deposit_consent": "true", "title": "A test paper on boundaries",
            "abstract": "An abstract long enough to pass the fifty-character minimum for uploads.",
            "keywords": "testing, boundaries", "license": "cc-by-4.0", "resource_type": "preprint",
            "publication_date": "2026-09-28", "subject": "", "funding": "",
            "creators": json.dumps([{"name": "Test Author", "orcid": TEST_ORCID}]),
            "related_identifiers": "[]", **extra,
        }
        req = httpx.Request("POST", "http://intake/api/submit", data=data,
                            files={"pdf": ("paper.pdf", pdf_bytes, "application/pdf")})
        body = req.read()
        ts = str(int(time.time()))
        # Signature v2 (2026-09-29): ts, ORCID, name (none sent), tier, then the body.
        sig = hmac.new(iss.HMAC_SECRET, f"icsac-v2\n{ts}\n{TEST_ORCID}\n\nt2\n".encode() + body,
                       hashlib.sha256).hexdigest()
        resp = client.post("/api/submit", content=body, headers={
            "content-type": req.headers["content-type"],
            "x-icsac-signature": f"v2={sig}", "x-icsac-timestamp": ts,
            "x-icsac-auth-orcid": TEST_ORCID, "x-icsac-test-tier": "t2",
        })
        try:
            return resp.status_code, resp.json()
        except Exception:
            return resp.status_code, {"raw": resp.text[:300]}

    def record(resp: dict) -> dict:
        sid = resp.get("sub_id") or resp.get("id") or ""
        return json.loads((subs / "test" / sid / "submission.json").read_text()) if sid else {}

    st, resp = submit(code_data_available="yes", code_data_url="https://github.com/example/generator")
    check(st in (200, 202) and record(resp).get("code_data") == {"available": True, "url": "https://github.com/example/generator"},
          f"yes + link -> stored ({st})")
    linked_id = resp.get("sub_id", "")
    st, resp = submit(code_data_available="yes", code_data_url="")
    check(st == 400 and "link" in json.dumps(resp), f"yes without a link -> 400 ({st})")
    for bad in ("ftp://example.org/x", "javascript:alert(1)", "https://github.com/a b",
                "http://localhost/x", "https://user@example.org/x", "https://example.org/" + "a" * 490):
        st, resp = submit(code_data_available="yes", code_data_url=bad)
        check(st == 400, f"bad link refused: {bad[:40]} ({st})")
    st, resp = submit(code_data_available="maybe")
    check(st == 400, f"an answer other than yes/no -> 400 ({st})")
    st, resp = submit(code_data_available="no", code_data_url="https://github.com/example/x")
    check(st in (200, 202) and record(resp).get("code_data") == {"available": False, "url": None},
          f"no -> stored as no, stray link dropped ({st})")
    st, resp = submit()
    held_id = resp.get("sub_id", "")
    check(st in (200, 202) and record(resp).get("code_data") == {"available": None, "url": None},
          f"no answer (older form) -> stored as unknown ({st})")
    st, resp = submit()
    held2_id = resp.get("sub_id", "")
    check("isSupplementedBy" in iss.RELATION_TYPES, "'is supplemented by' is an accepted relation")
    st, resp = submit(related_identifiers=json.dumps([{"identifier": "https://github.com/a/b",
                                                       "relation": "isSupplementedBy"}]))
    check(st in (200, 202), f"a related identifier 'is supplemented by' is accepted ({st})")
    check(all(e.get("event") != "code_data_link_missing" for e in audit_iss), "intake itself holds nothing")

    # ── 3. the worker: hold before the panel, override, linked paper ─────────
    print("3. worker")
    from intake import submission_worker as worker
    worker.SUBMISSIONS_ROOT = subs
    worker.QUEUE_DIR = subs / "queue"
    worker.TEST_SUBMISSIONS_ROOT = subs / "test"
    worker.TEST_QUEUE_DIR = subs / "test" / "queue"
    worker.TEST_REVIEWS_DIR = tmp / "reviews-test"
    audit_w: list[dict] = []
    worker._audit = lambda entry, test_mode=False: audit_w.append(entry)
    alerts: list = []
    worker._alert_remote = lambda *a, **k: alerts.append(a)
    pings: list = []
    worker.notify.send_telegram = lambda *a, **k: pings.append(a)
    worker.notify.send_to_curator = lambda *a, **k: pings.append(a)
    panel_calls: list[str] = []

    def fake_panel(review_data):
        panel_calls.append(review_data.get("title", ""))
        return "", {"recommendation": "PAUSED_AI_FAILURE"}
    worker.review.review_paper = fake_panel

    def state(sid: str) -> dict:
        return json.loads((subs / "test" / sid / "state.json").read_text())

    worker._process_locked(held_id)
    s = state(held_id)
    check(s.get("state") == "awaiting_decision" and s.get("pending_recommendation") == "REVISE_AND_RESUBMIT"
          and s.get("precheck") == "code_data_link_missing", "no link: held for a revise before the panel")
    check(panel_calls == [], "no link: the panel did not run")
    cda = json.loads((subs / "test" / held_id / "code_availability.json").read_text())
    check(cda["missing_link"] and cda["claims"], "no link: code_availability.json records the claims")
    check(any(e.get("event") == "code_data_link_missing" for e in audit_w), "no link: audited")
    check(pings == [] and alerts == [], "T2: no curator ping, no alert")

    worker._process_locked(linked_id)
    check(panel_calls == ["A test paper on boundaries"], "form link: the panel runs")
    check(state(linked_id).get("state") == "paused_panel_failure", "form link: normal path continues")

    from intake import code_check_override as cco
    rc = cco.main(["code-check-override.sh", held_id])
    s = state(held_id)
    check(rc == 0 and s.get("code_check_override") and s.get("precheck") == "overridden",
          "override: recorded, precheck cleared")
    check((subs / "test" / "queue" / held_id).exists(), "override: queued again")
    check(cco.main(["code-check-override.sh", held_id]) == 3, "override twice: refused")
    worker._process_locked(held_id)
    check(len(panel_calls) == 2, "override: the panel runs on the next pass")

    real_assess = worker.code_availability.assess
    def broken(*a, **k):
        raise RuntimeError("simulated check failure")
    worker.code_availability.assess = broken
    st_, resp_ = submit()
    broken_id = resp_.get("sub_id", "")
    worker._process_locked(broken_id)
    worker.code_availability.assess = real_assess
    check(len(panel_calls) == 3, "a broken check fails open: the panel still runs")
    check(any(e.get("event") == "code_check_failed" for e in audit_w), "a broken check is audited")
    check(pings == [] and alerts == [], "T2: a broken check pings no one (test tier)")

    # ── 4. decisions on a held paper: the revise email, the accept guard ─────
    print("4. decisions")
    worker._process_locked(held2_id)
    check(state(held2_id).get("precheck") == "code_data_link_missing", "second paper held")
    from intake import apply_decision as ad
    ad.SUBMISSIONS_ROOT = subs
    ad.TEST_SUBMISSIONS_ROOT = subs / "test"
    ad._audit = lambda entry, test_mode=False: audit_w.append(entry)
    ad.notify.send_to_curator = lambda *a, **k: pings.append(a)
    pathlib.Path.home = classmethod(lambda cls: tmp / "home")
    rc = ad.main(["apply_decision.py", held2_id, "accept"])
    check(rc == 3 and not state(held2_id).get("decision"), "accept on a held paper: refused, nothing recorded")
    rc = ad.main(["apply_decision.py", held2_id, "revise"])
    pathlib.Path.home = real_home
    check(rc in (0, None) and state(held2_id).get("decision") == "revise", f"revise: recorded (rc={rc})")
    eml = tmp / "home" / "icsac-submissions" / "test" / "_outbox" / f"{held2_id}.eml"
    check(eml.exists(), "revise: the email is written (T2 outbox)")
    if eml.exists():
        import email as _email
        import email.policy as _policy
        msg = _email.message_from_bytes(eml.read_bytes(), policy=_policy.default)
        text = "".join(p.get_payload(decode=True).decode("utf-8", "replace")
                       for p in msg.walk() if p.get_content_type() == "text/plain")
        norm = " ".join(text.split())
        check("revise and resubmit" in str(msg["Subject"] or "").lower(), f"revise email: subject ({msg['Subject']})")
        check("accompanying archive" in norm, "revise email: quotes the paper's own sentence")
        check("Harvard Dataverse" in norm and "hosted by the author" in norm, "revise email: says where and how it shows")
        check("{{" not in text and "}}" not in text, "revise email: no unfilled fields")
        check("hasn't been sent to the reviewer panel" in norm, "revise email: says the panel did not run")
        check(not any(p.get_content_type() == "application/pdf" for p in msg.walk()), "revise email: no report attached")
        check("reply" not in norm.lower(), "revise email: invites no reply")

    # ── 5. Zenodo relation and the registry field ────────────────────────────
    print("5. deposit + registry")
    import repository_deposit as rd
    sub = {"title": "T", "abstract": "A", "keywords": ["k"], "license": "cc-by-4.0",
           "publication_date": "2026-09-28", "resource_type": "preprint",
           "creators": [{"name": "Author, Test"}], "related_identifiers": [],
           "code_data": {"available": True, "url": "https://github.com/example/generator"}}
    md = rd._build_metadata(sub)
    check({"identifier": "https://github.com/example/generator", "relation": "isSupplementedBy"}
          in md.get("related_identifiers", []), "Zenodo: the link is 'is supplemented by'")
    sub["related_identifiers"] = [{"identifier": "https://github.com/example/generator", "relation": "isSupplementedBy"}]
    md = rd._build_metadata(sub)
    check(len(md["related_identifiers"]) == 1, "Zenodo: no duplicate when the author also listed it")
    import publications as pub
    site = tmp / "site"; (site / "src" / "data").mkdir(parents=True)
    (site / "src" / "data" / "accepted.json").write_text("[]")
    pub.WEBSITE_REPO = str(site); pub.REGISTRY_PATH = str(site / "src" / "data" / "accepted.json")
    check(pub.code_data_url(sub) == "https://github.com/example/generator", "registry: helper reads the form link")
    e = pub.upsert_entry(approved=True, proto={"title": "T", "authors": ["A"], "doi": "10.67697/icsac.2026.999",
                          "source": "submission-pdf", "code_data_url": pub.code_data_url(sub)})
    check(e.get("code_data_url") == "https://github.com/example/generator", "registry: the entry keeps code_data_url")
    check(pub.code_data_url({"code_data": {"available": False, "url": None}}) is None, "registry: 'no' gives no link")
finally:
    pathlib.Path.home = real_home
    shutil.rmtree(tmp, ignore_errors=True)

print(f"\n{'PASS' if not failures else 'FAILED'}: {len(failures)} failure(s)")
sys.exit(0 if not failures else 1)