Blame for scripts/record_fixtures.py

"""Record the Phase 0 lookups as replay cassettes — the ONLY live-network step in this repo.

Run by hand from the repo root, keyless, when fixtures are first made or after a reviewed
ToolUniverse bump (ADR-0001 §7). It runs each scenario through the real endpoint code and the
real, allow-listed ToolUniverse, and stores every raw response the endpoint received, keyed by
tool and arguments, in ``tests/fixtures/cassettes/<scenario>.json``. The test suite replays
those responses with sockets blocked, so what the tests check is exactly what the sources said
on the recording day. Re-recording is a reviewed change: diff the cassettes before committing.

Schema validation is switched off while recording (the schemas are written FROM these
responses); replay validates every response against the pinned schemas.

    python scripts/record_fixtures.py            # every scenario
    python scripts/record_fixtures.py q5_trials_cd40
"""

from __future__ import annotations

import datetime
import json
import sys
import time
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))

from src import config, direct_calls, schemas
from src.cache import ResponseCache
from src.endpoints import ENDPOINTS, _common
from src.tu_client import (
    ToolClient,
    build_tooluniverse,
    installed_tooluniverse_version,
)

CASSETTES = ROOT / "tests" / "fixtures" / "cassettes"
# The 55 pORG genes, as pdac-subtype-agent ships them (src/resources/signatures/pORG.tsv,
# commit 2451c18) — the same list Phase 0's Q4 annotated.
PORG_SOURCE = "pdac-subtype-agent src/resources/signatures/pORG.tsv @ 2451c18"
PORG = [
    "TNFSF18",
    "CST2",
    "RPRD1B",
    "BARD1",
    "SORD2P",
    "ZNF217",
    "CHMP4B",
    "SLFN13",
    "MMEL1",
    "SPATA24",
    "OSER1",
    "PHACTR3",
    "PKIB",
    "ARHGAP12",
    "RBM41",
    "MPZL3",
    "GDE1",
    "FGD6",
    "PLAAT2",
    "STAU1",
    "ZG16B",
    "RIOK3",
    "ALOX5",
    "CDS1",
    "PIK3CB",
    "HPGD",
    "HIST1H1T",
    "GID8",
    "NUDT21",
    "PAK1",
    "CYP4F3",
    "ARL17A",
    "PLAAT3",
    "SRP9",
    "ANKRD18B",
    "CCL20",
    "CFB",
    "SPOPL",
    "CTNNBL1",
    "NFE2L3",
    "ACSL5",
    "CIAO2A",
    "SCOC",
    "SAA4",
    "PPP1R3C",
    "WFDC13",
    "FUNDC1",
    "ADNP",
    "DLEU7",
    "NME7",
    "EFNB2",
    "LRRIQ1",
    "MAP3K2",
    "LNPK",
    "ZDHHC13",
]

SCENARIOS = {
    "q1_q2_drugs_kras": ("drugs_for_targets", ["KRAS"], "Phase 0 Q1 (KRAS G12D) and Q2 (G12C)"),
    "q3_annotate_elavl1": ("annotate_genes", ["ELAVL1"], "Phase 0 Q3 (HuR function)"),
    "q4_annotate_porg": ("annotate_genes", [",".join(PORG)], "Phase 0 Q4 (the 55 pORG genes)"),
    "q5_trials_cd40": ("trials_for_genes", ["CD40", "", ""], "Phase 0 Q5 (recruiting CD40 trials)"),
    "q6_drugs_jak_stat": ("drugs_for_targets", ["JAK1,JAK2,STAT1"], "Phase 0 Q6 (JAK1/JAK2/STAT1)"),
    "q7_drugs_tnfrsf10b": ("drugs_for_targets", ["TNFRSF10B"], "Phase 0 Q7 (DR5)"),
    "q8_targets_pdac": (
        "targets_for_disease",
        ["pancreatic ductal adenocarcinoma", ""],
        "Phase 0 Q8 (top PDAC targets)",
    ),
    "q9_pathways_isg15_rsad2_oas1": (
        "pathways_for_genes",
        ["ISG15,RSAD2,OAS1"],
        "Phase 0 Q9 (Interferon alpha/beta signaling)",
    ),
    "c1_annotate_kras_tp53_sord2p": (
        "annotate_genes",
        ["KRAS,TP53,SORD2P"],
        "plan C1(c): SORD2P listed as unresolved, not guessed",
    ),
    "edge_unknown_symbol": ("drugs_for_targets", ["NOTAGENE1"], "a symbol HGNC does not know"),
    "edge_legacy_efo": (
        "targets_for_disease",
        ["EFO_0002618", "5"],
        "a legacy EFO id Open Targets dropped",
    ),
}


class Recorder:
    def __init__(self):
        self.tu = build_tooluniverse()
        self.calls: list[dict] = []

    def __call__(self, tool: str, args: dict):
        if direct_calls.is_direct(tool):
            response = direct_calls.run(tool, args, timeout=30)
        else:
            response = self.tu.run_one_function({"name": tool, "arguments": args}, use_cache=False)
        self.calls.append({"tool": tool, "args": dict(args), "response": response})
        return response


def record(name: str) -> Path:
    endpoint, arguments, why = SCENARIOS[name]
    recorder = Recorder()
    client = ToolClient(runner=recorder, cache=ResponseCache(":memory:"))
    _common.set_client(client)
    started = time.time()
    payload = ENDPOINTS[endpoint][0](*arguments)
    used = sorted({config.source_of(c["tool"]) for c in recorder.calls})
    cassette = {
        "_recorded": {
            "scenario": name,
            "why": why,
            "endpoint": endpoint,
            "arguments": arguments,
            "tooluniverse_version": installed_tooluniverse_version(),
            "recorded_on": datetime.datetime.now(datetime.UTC).strftime("%Y-%m-%dT%H:%M:%SZ"),
            "recorded_at_epoch": round(started, 3),
            "recorded_with": "scripts/record_fixtures.py",
            "payload_status_when_recorded": payload.get("status"),
            "licences": {s: config.source(s)["licence"] for s in used},
            **({"porg_source": PORG_SOURCE} if name == "q4_annotate_porg" else {}),
        },
        "calls": recorder.calls,
    }
    path = CASSETTES / f"{name}.json"
    path.write_text(json.dumps(cassette, indent=1, ensure_ascii=False) + "\n", encoding="utf-8")
    print(
        f"{name}: {len(recorder.calls)} calls, payload {payload.get('status')}, "
        f"{len(payload.get('rows') or [])} rows, {time.time() - started:.1f} s -> {path.name}"
    )
    return path


def _check_porg_against_the_subtype_agent() -> None:
    """The list above was once typed from memory and was wrong; compare it to the source file."""
    relative = Path("pdac-subtype-agent/src/resources/signatures/pORG.tsv")
    for candidate in (ROOT.parent / relative, Path.home() / "Documents/GitHub" / relative):
        if candidate.is_file():
            listed = candidate.read_text().split()[1:]
            assert listed == PORG, "PORG differs from pdac-subtype-agent's pORG.tsv"
            return
    raise SystemExit("pdac-subtype-agent is not checked out beside this repo; cannot check PORG")


def main(names: list[str]) -> None:
    assert len(PORG) == 55, len(PORG)
    _check_porg_against_the_subtype_agent()
    CASSETTES.mkdir(parents=True, exist_ok=True)
    schemas.validate = lambda tool, data: None  # recording only: schemas are written from this
    for name in names or list(SCENARIOS):
        record(name)


if __name__ == "__main__":
    main(sys.argv[1:])