tests/test_structural_variants.py

"""The structural-variant modality — classification, and the three ways it must refuse.

SV exists because the KRAS-wild-type actionable genes (NRG1, NTRK1/2/3, ALK, ROS1, RET) are
FUSION-driven in PDAC. Carried on a mutation+CNV panel they would each have reported ~0%: a
confident false negative on the most clinically actionable question this agent can be asked.
That framing is why most of this file is about what the modality *declines* to say.
"""

from __future__ import annotations

import pytest

from src.tools.query_variant_status import query_variant_status
from src.workflows import curated_store
from src.workflows.variant_status import (
    classify_sv,
    sv_annotation_depth,
    sv_gene_symbols,
)

# --------------------------------------------------------------------------- #
# classification
# --------------------------------------------------------------------------- #


@pytest.mark.parametrize(
    "event_info, expected",
    [
        ("Protein Fusion: in frame  {ETV6:NTRK3}", "fusion_in_frame"),
        ("Protein fusion: in frame (SND1-BRAF)", "fusion_in_frame"),
        ("Protein Fusion: out of frame  {ARID1B:CDKN2A}", "fusion_out_of_frame"),
        ("CDKN2A-intragenic", "intragenic"),
        ("Deletion within transcript : mid-exon", "intragenic"),
        ("Duplication within transcript : mid-exon", "intragenic"),
        # An ANTISENSE fusion is not a protein fusion and must never read as one.
        ("Antisense Fusion", "rearrangement"),
        # A bare partner label carries no frame claim — this is CCLE's entire vocabulary.
        ("EML4-ALK fusion", "rearrangement"),
        ("ATP1B1-NRG1 Fusion - Archer", "rearrangement"),
    ],
)
def test_classify_sv_reads_frame_from_event_info(event_info, expected):
    assert classify_sv(event_info) == expected


# --------------------------------------------------------------------------- #
# the frame claim that lives in `annotation` — real `pdac_msk_2024` rows, verbatim
# --------------------------------------------------------------------------- #
# Found in the 2026-08-05 signed-in click-through. `pdac_msk_2024`'s Archer-panel rows put a
# BARE LABEL in `eventInfo` and the actual finding in `annotation`, so reading `eventInfo` alone
# reported them as `rearrangement` — 10 events over NRG1 (6), NTRK3 (2), NTRK1 (1), RET (1):
# precisely the KRAS-wild-type actionable tier the SV modality was added to answer for.
#
# Nothing flagged it. The cohort's OTHER rows do state the frame in `eventInfo`, so
# `sv_annotation_depth` said "characterized" and every gate stayed quiet.
#
# `site2EffectOnFrame` is the field that ought to carry this and cannot: it is the string 'NA'
# on all 110 somatic panel rows in the cohort — the same trap `variantClass` set.
MSK_ARCHER_ROWS = [
    {
        "eventInfo": "ATP1B1-NRG1 Fusion - Archer",
        "annotation": (
            "POSITIVE FOR GENE FUSIONS IN THE INVESTIGATIONAL PANEL: ATP1B1-NRG1 fusion. Note: "
            "The rearrangement is an in-frame fusion between genes ATP1B1 Exon2 (NM_001677) and "
            "NRG1 Exon2 (NM_004495)."
        ),
        "site2EffectOnFrame": "NA",
        "variantClass": "NA",
    },
    {
        "eventInfo": "NOTCH2-NRG1 Fusion - Archer",
        "annotation": (
            "POSITIVE FOR GENE FUSIONS IN THE INVESTIGATIONAL PANEL: NOTCH2-NRG1 fusion. Note: "
            "The rearrangement is an in-frame fusion between genes NOTCH2 Exon4 (NM_024408) and "
            "NRG1 Exon6 (NM_004495). (PMID: 30988082)"
        ),
        "site2EffectOnFrame": "NA",
        "variantClass": "NA",
    },
]


@pytest.mark.parametrize("row", MSK_ARCHER_ROWS)
def test_classify_sv_reads_the_frame_claim_out_of_annotation(row):
    """`eventInfo` alone says only that something happened; `annotation` says what."""
    assert classify_sv(row["eventInfo"], row["variantClass"]) == "rearrangement"
    assert classify_sv(row["eventInfo"], row["variantClass"], row["annotation"]) == "fusion_in_frame"


def test_annotation_is_read_in_the_hyphenated_form_it_is_actually_written_in():
    """MSK writes "in-frame" there and "in frame" in eventInfo — both must land the same."""
    spaced = classify_sv("X-Y Fusion", "NA", "an in frame fusion between genes X and Y")
    hyphenated = classify_sv("X-Y Fusion", "NA", "an in-frame fusion between genes X and Y")
    assert spaced == hyphenated == "fusion_in_frame"
    assert classify_sv("X-Y Fusion", "NA", "an out-of-frame fusion between X and Y") == (
        "fusion_out_of_frame"
    )


def test_event_info_wins_when_both_fields_make_a_claim():
    """`eventInfo` is the curated event summary; `annotation` is the fallback, not an override."""
    assert classify_sv("Protein Fusion: out of frame {A:B}", "NA", "an in-frame fusion") == (
        "fusion_out_of_frame"
    )


def test_a_cohort_annotated_only_in_annotation_is_still_characterized():
    """Otherwise the per-cohort collapse flattens the very events this fix recovers.

    The depth gate made the identical string test as the classifier, so fixing one alone would
    have left the cohort judged uncharacterized and every in-frame call reset to
    `rearrangement`. They now share `_frame_claim` + `_sv_text`.
    """
    assert sv_annotation_depth(MSK_ARCHER_ROWS) == "characterized"


def test_annotation_cannot_promote_ccle():
    """CCLE's whole caveat rests on bare partner labels never being promoted.

    Verified against the live cohort before `annotation` was trusted at all: across its 322
    somatic panel rows, no annotation contains "in frame" or "out of frame" in any spelling —
    they are partner/panel strings like "gs1 259h13.10 >chr7:...,brca1 >archerdx panel". So the
    fallback cannot fire here, and an absent in-frame call stays a fact about the annotation.
    """
    ccle_like = [
        {
            "eventInfo": "EML4-ALK fusion",
            "annotation": "ac010642.1 >chr19:58790332 58826563:+,gnas >archerdx panel,foundationone",
        },
        {"eventInfo": "PVT1-MYC fusion", "annotation": "rp13 16h11.2 >chr10:26953135 26965088: ,gnas"},
    ]
    for row in ccle_like:
        assert classify_sv(row["eventInfo"], "NA", row["annotation"]) == "rearrangement"
    assert sv_annotation_depth(ccle_like) == "uncharacterized"


def test_separator_normalisation_does_not_invent_a_frame_claim():
    """Hyphens are load-bearing in gene symbols, not only in "in-frame"."""
    for info in ("EML4-ALK fusion", "ATP1B1-NRG1 Fusion - Archer", "PVT1-MYC fusion"):
        assert classify_sv(info) == "rearrangement"
    assert sv_annotation_depth([{"eventInfo": "EML4-ALK fusion"}]) == "uncharacterized"


def test_classify_sv_never_promotes_an_uncharacterized_event_to_a_fusion():
    """The whole failure mode in one assertion.

    `variantClass` describes the GENOMIC event (TRANSLOCATION/DELETION/…), not its consequence
    for the protein, so it can never license an in-frame claim. A classifier keyed on it would
    have turned every one of CCLE's translocations into a "fusion".
    """
    for vc in ("TRANSLOCATION", "DELETION", "DUPLICATION", "INVERSION", "NA"):
        assert classify_sv("", vc) in {"rearrangement", "none"}
        assert classify_sv("", vc) != "fusion_in_frame"


def test_empty_event_with_no_variant_class_is_not_an_alteration():
    assert classify_sv("", "") == "none"


def test_sv_gene_symbols_reads_both_ends():
    """A panel gene is often the 3' partner; reading site1 alone loses most NRG1/NTRK3 events."""
    row = {"site1HugoSymbol": "ETV6", "site2HugoSymbol": "NTRK3"}
    assert sv_gene_symbols(row) == ["ETV6", "NTRK3"]
    assert sv_gene_symbols({"site1HugoSymbol": "CDKN2A", "site2HugoSymbol": ""}) == ["CDKN2A"]


# --------------------------------------------------------------------------- #
# the annotation-depth gate
# --------------------------------------------------------------------------- #


def test_annotation_depth_distinguishes_the_two_sv_cohorts():
    """`characterized` vs `uncharacterized` is the difference between the only two SV cohorts."""
    msk_like = [{"eventInfo": "Protein Fusion: in frame  {ETV6:NTRK3}"}, {"eventInfo": "CDKN2A-intragenic"}]
    ccle_like = [{"eventInfo": "EML4-ALK fusion"}, {"eventInfo": "PVT1-MYC fusion"}]

    assert sv_annotation_depth(msk_like) == "characterized"
    assert sv_annotation_depth(ccle_like) == "uncharacterized"
    assert sv_annotation_depth([]) == "uncharacterized"


def test_committed_sv_cohorts_carry_the_annotation_depth_we_measured():
    """Pinned against the real artifacts, because this drove a design decision.

    `ccle_broad_2019` reports `variantClass: NA` on every panel-gene row and labels events only
    as free text; `pdac_msk_2024` states frame. If a re-curation ever flips one of these, the
    payload's caveat changes meaning and someone should look.
    """
    assert curated_store.read("ccle_broad_2019")["sv_annotation"] == "uncharacterized"
    assert curated_store.read("pdac_msk_2024")["sv_annotation"] == "characterized"


def test_uncharacterized_cohort_never_reports_an_in_frame_fusion():
    """An absence of in-frame fusions in CCLE is a fact about the annotation, not the biology."""
    res = query_variant_status(["ALK", "MET", "NRG1"], "cbioportal:ccle_broad_2019")

    assert res["sv_annotation"] == "uncharacterized"
    assert res["caveats"], "an uncharacterized SV cohort must say so in the answer itself"
    assert "NOT about the biology" in res["caveats"][0]
    for gene, entry in res["genes"].items():
        for status in (entry.get("sv") or {}).get("per_sample", {}).values():
            assert status == "rearrangement", f"{gene} claimed {status} in an uncharacterized cohort"


def test_characterized_cohort_may_report_frame():
    res = query_variant_status(["NTRK1", "NTRK3", "ROS1"], "cbioportal:pdac_msk_2024")

    assert res["sv_annotation"] == "characterized"
    assert res["caveats"] == []
    seen = {s for e in res["genes"].values() for s in (e.get("sv") or {}).get("per_sample", {}).values()}
    assert "fusion_in_frame" in seen


# --------------------------------------------------------------------------- #
# the per-study modality gate — five of seven cohorts have no SV at all
# --------------------------------------------------------------------------- #


@pytest.mark.parametrize(
    "study", ["paad_tcga", "pancreas_cptac_gdc", "paad_utsw_2015", "paad_qcmg_uq_2016", "paad_cptac_2021"]
)
def test_cohort_without_sv_yields_no_fusion_answer(study):
    """No SV profile ⇒ no `sv` key anywhere, and the absence is NAMED.

    A null "no fusions found" column would be the exact confident false negative this modality
    was built to remove — worse than the mutation+CNV panel it replaced, because it would look
    like a measurement.
    """
    res = query_variant_status(["NRG1", "NTRK3", "ROS1"], f"cbioportal:{study}")

    assert "sv" in res["unavailable_modalities"]
    assert res["sv_annotation"] is None
    for gene, entry in res["genes"].items():
        assert "sv" not in entry, f"{gene} got an sv block from a cohort with no SV profile"


# --------------------------------------------------------------------------- #
# the per-gene gate is MODALITY-SCOPED
# --------------------------------------------------------------------------- #


def test_gene_off_the_dna_panel_still_reports_its_fusions():
    """NRG1 on MSK: absent from the DNA panel, yet the cohort holds 6 somatic rearrangements.

    `genes_assayed` comes from the DNA gene panel (IMPACT341/468/505…), so it answers "was this
    sequenced for mutations and copy number?" and nothing else. Applying it to SV suppressed
    data we actually hold — the mirror image of the false negative the gate exists to prevent:
    refusing an answer we have, rather than inventing one we don't.
    """
    res = query_variant_status(["NRG1"], "cbioportal:pdac_msk_2024")
    entry = res["genes"]["NRG1"]

    assert entry["assayed"] is False  # honest about the DNA panel
    assert "mutation" not in entry  # …and still refuses the modality it cannot speak to
    assert "cnv" not in entry
    assert entry["sv"]["n_altered"] == 6  # but the fusions are reported
    assert "Do not read the SV frequency as a mutation frequency" in entry["note"]


def test_gene_off_the_panel_with_no_events_stays_fully_refused():
    """GATA6 on MSK: off the DNA panel AND no observed SV — no evidence of interrogation at all."""
    res = query_variant_status(["GATA6"], "cbioportal:pdac_msk_2024")
    entry = res["genes"]["GATA6"]

    assert entry["assayed"] is False
    assert "sv" not in entry
    assert "mutation" not in entry
    assert "not absence of alteration" in entry["note"]


def test_a_panel_gene_with_no_fusions_reports_a_real_zero():
    """KRAS is on the DNA panel and fusion-called, so 0% here is a measurement, not a gap."""
    res = query_variant_status(["KRAS"], "cbioportal:pdac_msk_2024")
    sv = res["genes"]["KRAS"]["sv"]

    assert sv["n_altered"] == 0
    assert sv["n_profiled"] == 2336
    assert res["genes"]["KRAS"]["mutation"]["n_altered"] > 2000  # the gene is plainly assayed


# --------------------------------------------------------------------------- #
# provenance must not collide with the mutation half
# --------------------------------------------------------------------------- #


def test_sv_provenance_does_not_overwrite_mutation_provenance():
    """Both dicts are keyed (gene, sample) and one sample can carry a mutation AND a fusion.

    Sharing one dict would let whichever curation step ran last win, so a KRAS G12D could be
    reported as the provenance of a rearrangement.
    """
    res = query_variant_status(["TP53", "CDKN2A"], "cbioportal:pdac_msk_2024")

    overlapping = 0
    for gene in ("TP53", "CDKN2A"):
        entry = res["genes"][gene]
        mut_prov = entry["mutation"]["provenance"]
        sv_prov = entry["sv"]["provenance"]
        for sample in set(mut_prov) & set(sv_prov):
            overlapping += 1
            # the mutation string names the gene's protein change; the SV string is an event
            assert "rearrangement" in sv_prov[sample] or "usion" in sv_prov[sample]
            assert sv_prov[sample] != mut_prov[sample]
    assert overlapping, "expected at least one sample carrying both a mutation and an SV"