"""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"