tests/test_airgap.py

"""Tests for the airgap (gene-name anonymisation).

Covers both the anonymise/reveal round-trip and the structural invariant that
the engine doesn't depend on the sealed map or on `reveal`.
"""

from __future__ import annotations

import json
from pathlib import Path

import pandas as pd
import pytest

from airgap import anonymise, reveal
from airgap.seal import OPAQUE_ID_RE
from dsl import Search

REPO_ROOT = Path(__file__).resolve().parent.parent


@pytest.fixture(autouse=True)
def _isolated_sealed_map(tmp_path, monkeypatch):
    """Each test gets its own sealed-map path so we don't touch the real one."""
    import airgap.seal as seal
    fake_path = tmp_path / "_sealed_gene_map.json"
    monkeypatch.setattr(seal, "SEALED_PATH", fake_path)
    yield fake_path


def _toy_named_matrix() -> pd.DataFrame:
    return pd.DataFrame(
        {
            "MLH1":  [1.0, 2.0, 3.0],
            "TP53":  [4.0, 5.0, 6.0],
            "CD8A":  [7.0, 8.0, 9.0],
            "BRCA1": [0.5, 0.6, 0.7],
        },
        index=[f"s{i}" for i in range(3)],
    )


def test_anonymise_renames_to_opaque_ids_and_writes_sealed_map(_isolated_sealed_map):
    m = _toy_named_matrix()
    anon = anonymise(m)
    assert all(OPAQUE_ID_RE.match(c) for c in anon.columns)
    assert anon.shape == m.shape
    assert _isolated_sealed_map.exists()
    sealed = json.loads(_isolated_sealed_map.read_text())
    assert sealed["n_genes"] == m.shape[1]
    assert set(sealed["id_to_symbol"].values()) == set(m.columns)


def test_reveal_round_trips(_isolated_sealed_map):
    m = _toy_named_matrix()
    anon = anonymise(m)
    revealed = reveal(list(anon.columns))
    # values per opaque ID column should equal values of its revealed name column.
    for opaque, real in zip(anon.columns, revealed):
        assert (anon[opaque].values == m[real].values).all()


def test_anonymise_idempotent_with_existing_sealed_map(_isolated_sealed_map):
    m = _toy_named_matrix()
    a1 = anonymise(m)
    a2 = anonymise(m)
    assert list(a1.columns) == list(a2.columns)


def test_anonymise_fails_loudly_on_unknown_gene(_isolated_sealed_map):
    m = _toy_named_matrix()
    anonymise(m)
    extended = m.copy()
    extended["NOVEL_GENE"] = [0.1, 0.2, 0.3]
    with pytest.raises(ValueError, match="not in the sealed map"):
        anonymise(extended)


def test_reveal_rejects_unknown_id(_isolated_sealed_map):
    anonymise(_toy_named_matrix())
    with pytest.raises(KeyError, match="unknown opaque IDs"):
        reveal(["g99999"])


# --- Enforcement ------------------------------------------------------------


def test_search_only_accepts_anonymised_matrix(_isolated_sealed_map):
    m = _toy_named_matrix()
    with pytest.raises(ValueError, match="opaque IDs"):
        Search(m, lambda s: float(s.sum()), 2)
    # The anonymised view passes — Search runs on opaque IDs.
    anon = anonymise(m)
    top = Search(anon, lambda s: float(s.sum()), 2)
    assert all(OPAQUE_ID_RE.match(c) for c in top)


def test_engine_does_not_import_sealed_map_or_reveal():
    """Structural invariant: the GP engine must never see the sealed map.

    We scan every file under engine/ for references to `reveal` or to the
    sealed map filename, and forbid either. This is the airgap's load-bearing
    structural check — the runtime opaque-ID regex in Search complements it.
    """
    engine_dir = REPO_ROOT / "engine"
    forbidden = ["reveal", "_sealed_gene_map.json", "airgap.seal"]
    offenders: list[tuple[Path, str]] = []
    for path in engine_dir.rglob("*.py"):
        text = path.read_text()
        for token in forbidden:
            if token in text:
                offenders.append((path, token))
    assert not offenders, (
        "Engine has a forbidden airgap-breaking reference: "
        + ", ".join(f"{p.relative_to(REPO_ROOT)} -> {t}" for p, t in offenders)
    )