tests/test_transfer_gse65858.py

"""Synthetic-cohort tests for ``validate/transfer_gse65858.py``.

No network. No engine import. A GSE65858-shaped fixture (symbols ×
samples matrix + `sample_id`/`hpv_status` clinical frame) is written
into a tmp_path directory; the transfer function points at it via the
`processed_dir` override.
"""

from __future__ import annotations

import numpy as np
import pandas as pd
import pytest

from data_pipeline import schema
from validate.transfer_gse65858 import transfer_score


N = 120
N_POS = 40   # HPV+ count
RNG_SEED = 0


def _write_fixture(tmp_path, *, signal_gene: str, extra_genes: list[str],
                   signal_strength: float = 3.0):
    """Write a synthetic (symbols x samples) parquet + clinical parquet
    into `tmp_path`. The signal gene is drawn so HPV+ patients score
    reliably higher; the extra genes are pure noise."""
    rng = np.random.default_rng(RNG_SEED)
    sample_ids = [f"GSMsynth{i:04d}" for i in range(N)]
    hpv = [schema.GSE65858_HPV_POS_LABEL] * N_POS + [
        schema.GSE65858_HPV_NEG_LABEL
    ] * (N - N_POS)

    all_genes = [signal_gene, *extra_genes]
    mat = rng.normal(loc=0.0, scale=1.0, size=(len(all_genes), N))
    # Boost the signal gene on HPV+ samples.
    signal_ix = all_genes.index(signal_gene)
    mat[signal_ix, :N_POS] += signal_strength
    expr = pd.DataFrame(mat, index=all_genes, columns=sample_ids)
    expr.index.name = "symbol"

    clin = pd.DataFrame({
        "sample_id":  sample_ids,
        "hpv_status": hpv,
        "has_expression": True,
    })

    tmp_path.mkdir(parents=True, exist_ok=True)
    expr.to_parquet(tmp_path / "expression.parquet")
    clin.to_parquet(tmp_path / "clinical.parquet", index=False)


def test_signal_gene_produces_high_auroc_and_small_p(tmp_path):
    """A gene that carries the HPV+ signal must yield an AUROC well
    above 0.5 and a small permutation p."""
    _write_fixture(
        tmp_path,
        signal_gene="SIG1",
        extra_genes=["NOISE1", "NOISE2"],
        signal_strength=3.0,
    )
    out = transfer_score(
        ["SIG1"], n_permutations=200, seed=0, processed_dir=tmp_path,
    )
    assert out["auroc"] is not None
    assert out["auroc"] > 0.9
    assert out["p"] is not None
    assert out["p"] < 0.05
    assert out["n"] == N
    assert out["n_pos"] == N_POS
    assert out["n_neg"] == N - N_POS
    assert out["n_found"] == 1
    assert out["n_missing"] == 0
    assert out["found_symbols"] == ["SIG1"]
    assert out["missing_symbols"] == []


def test_noise_only_symbols_are_around_chance(tmp_path):
    """A set of noise-only symbols must sit near AUROC 0.5 with a
    non-significant p — the transfer test should not fool itself."""
    _write_fixture(
        tmp_path,
        signal_gene="SIG1",
        extra_genes=["NOISE1", "NOISE2", "NOISE3"],
        signal_strength=3.0,
    )
    out = transfer_score(
        ["NOISE1", "NOISE2", "NOISE3"],
        n_permutations=200, seed=0, processed_dir=tmp_path,
    )
    assert out["auroc"] is not None
    # Orientation-agnostic AUROC is bounded [0.5, 1.0]. Noise should
    # land near 0.5 (+ a small stochastic bump from the ceiling).
    assert 0.5 <= out["auroc"] < 0.70
    # A non-significant result (p should generally be large; give a
    # forgiving bound since 200 permutations is small).
    assert out["p"] is not None
    assert out["p"] > 0.10
    assert out["n_found"] == 3
    assert out["missing_symbols"] == []


def test_missing_symbols_are_reported_without_crashing(tmp_path):
    """Symbols not present in the cohort get reported cleanly."""
    _write_fixture(
        tmp_path,
        signal_gene="SIG1",
        extra_genes=["NOISE1"],
    )
    out = transfer_score(
        ["SIG1", "NOTPRESENT_A", "NOTPRESENT_B"],
        n_permutations=100, seed=0, processed_dir=tmp_path,
    )
    assert out["found_symbols"] == ["SIG1"]
    assert set(out["missing_symbols"]) == {"NOTPRESENT_A", "NOTPRESENT_B"}
    assert out["n_found"] == 1
    assert out["n_missing"] == 2
    assert out["auroc"] is not None


def test_no_found_symbols_returns_graceful_none(tmp_path):
    """When none of the winner's genes appear in the cohort, the
    function returns a graceful payload (auroc/p None) instead of
    crashing or inventing a number."""
    _write_fixture(
        tmp_path,
        signal_gene="SIG1",
        extra_genes=["NOISE1"],
    )
    out = transfer_score(
        ["FOO_MISSING", "BAR_MISSING"],
        n_permutations=50, seed=0, processed_dir=tmp_path,
    )
    assert out["auroc"] is None
    assert out["p"] is None
    assert out["n_found"] == 0
    assert out["n_missing"] == 2
    assert out["found_symbols"] == []


def test_returned_payload_contains_only_supplied_symbols(tmp_path):
    """AIRGAP invariant: the payload's gene NAMES are exactly the ones
    the caller passed in. No full gene list from the cohort leaks."""
    _write_fixture(
        tmp_path,
        signal_gene="SIG1",
        extra_genes=["NOISE_SECRET_A", "NOISE_SECRET_B"],
    )
    out = transfer_score(
        ["SIG1"], n_permutations=100, seed=0, processed_dir=tmp_path,
    )
    all_names = list(out["found_symbols"]) + list(out["missing_symbols"])
    assert all_names == ["SIG1"]
    # NOISE_SECRET_A / _B never appear in the payload — they were in
    # the cohort but never requested by the caller.
    payload_text = repr(out)
    assert "NOISE_SECRET_A" not in payload_text
    assert "NOISE_SECRET_B" not in payload_text