Spaces:
Sleeping
Sleeping
| """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" | |