Spaces:
Sleeping
Sleeping
File size: 3,005 Bytes
0fff343 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 | """Smoke tests for data_pipeline.build. No network."""
from __future__ import annotations
import math
from pathlib import Path
import pytest
from data_pipeline import build, schema
def test_label_msi_thresholds():
# >= 10 -> MSI-H
assert build.label_msi(10.0) == "MSI-H"
assert build.label_msi(42.3) == "MSI-H"
# < 4 -> MSS
assert build.label_msi(0.0) == "MSS"
assert build.label_msi(3.99) == "MSS"
# in between -> Indeterminate
assert build.label_msi(4.0) == "MSI-Indeterminate"
assert build.label_msi(9.99) == "MSI-Indeterminate"
# missing
assert build.label_msi(float("nan")) == "NA"
assert build.label_msi(None) == "NA" # type: ignore[arg-type]
def test_normalise_stage():
assert build.normalise_stage("STAGE IIIB") == "III"
assert build.normalise_stage("Stage iv") == "IV"
assert build.normalise_stage("STAGE I") == "I"
assert build.normalise_stage("STAGE IIA") == "II"
assert build.normalise_stage("[Not Available]") == "NA"
assert build.normalise_stage(None) == "NA"
assert build.normalise_stage(float("nan")) == "NA"
def test_parse_os_status():
assert build.parse_os_status("0:LIVING") == 0.0
assert build.parse_os_status("1:DECEASED") == 1.0
assert math.isnan(build.parse_os_status(""))
assert math.isnan(build.parse_os_status(None))
def _write_fake_clinical(tmp_path: Path, columns: list[str], rows: list[list[str]]) -> Path:
"""Write a cBioPortal-shaped TSV: 4 leading '#'-prefixed metadata lines, header, rows."""
p = tmp_path / "data_clinical_sample.txt"
lines = [
"#" + "\t".join(columns), # display names
"#" + "\t".join("desc" for _ in columns), # descriptions
"#" + "\t".join("STRING" for _ in columns), # types
"#" + "\t".join("1" for _ in columns), # priorities
"\t".join(columns),
*["\t".join(r) for r in rows],
]
p.write_text("\n".join(lines) + "\n")
return p
def test_read_clinical_missing_column_raises_loudly(tmp_path):
cols = ["PATIENT_ID", "SAMPLE_ID", "ONCOTREE_CODE"] # missing MSI_SENSOR_SCORE, TMB_NONSYNONYMOUS
path = _write_fake_clinical(tmp_path, cols, [["P1", "P1-01", "COAD"]])
with pytest.raises(KeyError) as exc:
build.read_clinical(path, schema.REQUIRED_SAMPLE_COLS)
msg = str(exc.value)
assert "MSI_SENSOR_SCORE" in msg
assert "TMB_NONSYNONYMOUS" in msg
# Must surface what *is* present so a human can debug.
assert "ONCOTREE_CODE" in msg
def test_read_clinical_happy_path(tmp_path):
cols = schema.REQUIRED_SAMPLE_COLS
path = _write_fake_clinical(
tmp_path,
cols,
[
["P1", "P1-01", "COAD", "12.3", "5.4"],
["P2", "P2-01", "READ", "0.5", "1.2"],
],
)
df = build.read_clinical(path, schema.REQUIRED_SAMPLE_COLS)
assert len(df) == 2
assert list(df.columns)[: len(cols)] == cols
assert df.iloc[0]["PATIENT_ID"] == "P1"
|