"""QC report (spec QC-040 … QC-063): dataset discovery, labels, types, table, HTML determinism and error codes.""" from __future__ import annotations import datetime as dt import json import shutil import subprocess from pathlib import Path import pytest from typer.testing import CliRunner from simurg.cli.main import app from simurg.core.errors import SimurgError from simurg.core.log import Reporter from simurg.pipeline.qc_report import QcReportOptions, assign_labels, resolve_datasets, run_qc_report RAW_HDR = "sample_id\tcategory\tbead_address\tcontrol_name\tcolor_label\tgreen_intensity\tred_intensity\n" PROBES = [("Staining", "1", "Biotin (High)", "Green"), ("Staining", "2", "Biotin (Bgnd)", "Blue"), ("Non-Polymorphic", "3", "NP (A)", "Red"), ("NEGATIVE", "4", "neg", "Red")] # fmt: skip def make_dataset(folder: Path, n: int = 4, array_type: str = "genotyping", summary: bool = True) -> Path: folder.mkdir(parents=True, exist_ok=True) ids = [f"2000000000{k}_R0{i + 1}C01" for k in (1,) for i in range(n)] raw = RAW_HDR + "".join( f"{s}\t{c}\t{a}\t{nm}\t{col}\t{1000 * (j + 1) + i}\t{500 + i}\n" for i, s in enumerate(ids) for j, (c, a, nm, col) in enumerate(PROBES) ) (folder / "controls_raw.tsv").write_text(raw) qc = "sample_id\tarray_type\timaging_date\tsample_plate\tsample_well\tscanner_id\tscanner_version\tbg_offset\tstaining_green\tnp_red\tna_metrics\n" qc += "".join( f"{s}\t{array_type}\t2026-01-0{i + 1}T10:00:00\tP1\tA0{i + 1}\tN1\t1.0\t3000\t{6 - i}\t{10 + i}\t\n" for i, s in enumerate(ids) ) (folder / "controls_qc.tsv").write_text(qc) if summary: cols = ["sample_id", "sample_name", "sample_folder", "status", "fail_reason", "autosomal_call_rate", "call_rate", "logr_dev", "sex_est", "sex_reported", "sex_check", "het_rate", "contamination", "functional_status", "control_status", "control_flag_codes", "flags", "Sample_Group"] # fmt: skip lines = ["\t".join(cols)] for i, s in enumerate(ids): vals = {"sample_id": s, "sample_name": f"S{i}", "sample_folder": "x", "status": "OK", "fail_reason": "", "autosomal_call_rate": f"{0.99 - 0.01 * i}", "call_rate": "0.99", "logr_dev": f"{0.1 + 0.05 * i}", "sex_est": "MF"[i % 2], "sex_reported": "U", "sex_check": "unknown", "het_rate": "0.17", "contamination": "0.001", "functional_status": "", "control_status": "", "control_flag_codes": "", "flags": "", "Sample_Group": "g" + str(i % 2)} # fmt: skip lines.append("\t".join(vals[c] for c in cols)) (folder / "sample_summary.tsv").write_text("\n".join(lines) + "\n") return folder def run(tmp_path: Path, data: list[str], **kw: object) -> Path: out = tmp_path / "out" opts = QcReportOptions(data=data, out=out, report_name="r", now=dt.datetime(2026, 1, 2, tzinfo=dt.UTC), **kw) # type: ignore[arg-type] assert run_qc_report(opts, Reporter(command="t", quiet=True)) == 0 return out def test_discovery_forms(tmp_path: Path) -> None: # QC-041 a = make_dataset(tmp_path / "p" / "b") b = make_dataset(tmp_path / "p" / "a") assert resolve_datasets([str(tmp_path / "p")]) == [b, a] # sorted by path assert resolve_datasets([f"{a},{b}"]) == [a, b] with pytest.raises(SimurgError) as e: resolve_datasets([str(tmp_path / "nope")]) assert e.value.code == "E_QC_NO_DATASETS" (tmp_path / "empty").mkdir() with pytest.raises(SimurgError) as e: resolve_datasets([str(tmp_path / "empty")]) assert e.value.code == "E_QC_NO_DATASETS" (b / "controls_raw.tsv").unlink() with pytest.raises(SimurgError) as e: resolve_datasets([f"{a},{b}"]) assert e.value.code == "E_QC_DATASET_INCOMPLETE" def test_labels(tmp_path: Path) -> None: # QC-042 rep = Reporter(command="t", quiet=True) f = [tmp_path / "x" / "run", tmp_path / "y" / "run"] assert assign_labels(f, None, rep) == ["run", "run_2"] assert assign_labels(f, ["My Label<1>", "b"], rep) == ["My_Label_1_", "b"] assert rep.warning_counts().get("W_QC_LABEL_SANITIZED") == 1 for labels, code in ((["a"], "E_QC_LABEL_COUNT"), (["a b", "a_b"], "E_QC_LABEL_DUPLICATE")): with pytest.raises(SimurgError) as e: assign_labels(f, labels, rep) assert e.value.code == code def test_report_outputs_and_determinism(tmp_path: Path) -> None: # QC-044, QC-045, QC-047 d = make_dataset(tmp_path / "ds") out = run(tmp_path, [str(d)], output_format="both") html = (out / "r.html").read_text() assert "SimurgArray QC Report" in html and "https://" not in html.split("plotly.js v")[0] first = html out2 = run(tmp_path, [str(d)], output_format="both") assert (out2 / "r.html").read_text() == first # byte-deterministic with a fixed timestamp rows = (out / "r_table.csv").read_text().splitlines() head = rows[0].split(",") assert head[:7] == ["dataset", "sample_id", "sample_name", "functional_status", "control_status", "control_flag_codes", "na_metrics"] # fmt: skip assert "neg_red" not in head and "Biotin (High)_green" in head # QC-010 exclusion assert head[-1] == "sample_well" or "Sample_Group" in head from openpyxl import load_workbook wb = load_workbook(out / "r_table.xlsx") assert wb.sheetnames == ["Samples", "Thresholds", "Datasets", "About"] ws = wb["Samples"] status = [c.value for c in ws["D"]][1:] assert status == ["PASS", "PASS", "FAIL", "FAIL"] # call rate 0.97 < 0.98; LRR SD 0.25 > 0.20 def test_privacy_and_errors(tmp_path: Path) -> None: # QC-043, QC-046, QC-063 d = make_dataset(tmp_path / "ds") out = run(tmp_path, [str(d)], output_format="csv", no_metadata=True) head = (out / "r_table.csv").read_text().splitlines()[0] assert "Sample_Group" not in head and "sample_plate" not in head runner = CliRunner() res = runner.invoke(app, ["qc", "report", "--data", str(d), "--out", str(tmp_path / "o"), "--output-format", "pdf"]) assert res.exit_code == 2 and "E_QC_FORMAT" in res.output m = make_dataset(tmp_path / "meth", array_type="methylation") res = runner.invoke(app, ["qc", "report", "--data", f"{d},{m}", "--out", str(tmp_path / "o")]) assert res.exit_code == 2 and "E_QC_MIXED_TYPES" in res.output bad = tmp_path / "bad.yaml" bad.write_text("qc: {report: {callRate: 2}}") res = runner.invoke(app, ["qc", "report", "--data", str(d), "--out", str(tmp_path / "o"), "--config", str(bad)]) assert res.exit_code == 2 and "E_QC_THRESHOLD_RANGE" in res.output def test_unmatched_samples_warn(tmp_path: Path) -> None: # QC-040 d = make_dataset(tmp_path / "ds") lines = (d / "sample_summary.tsv").read_text().splitlines() (d / "sample_summary.tsv").write_text("\n".join(lines[:-1]) + "\n") rep = Reporter(command="t", quiet=True) run_qc_report(QcReportOptions(data=[str(d)], out=tmp_path / "o", output_format="csv"), rep) assert rep.warning_counts().get("W_QC_UNMATCHED_SAMPLES") == 1 @pytest.mark.slow def test_report_selftest_in_browser(tmp_path: Path) -> None: # QC-051, QC-055 (client-side behaviour) browser = shutil.which("chromium") or shutil.which("chromium-browser") or shutil.which("google-chrome") if not browser: pytest.skip("no Chromium for the headless report test") d = make_dataset(tmp_path / "ds", n=6) out = run(tmp_path, [str(d)], output_format="csv") dom = subprocess.run( [ browser, "--headless=new", "--no-sandbox", "--disable-gpu", "--virtual-time-budget=20000", "--dump-dom", f"file://{out / 'r.html'}#selftest", ], capture_output=True, text=True, timeout=180, ).stdout marker = 'data-selftest="' assert marker in dom, dom[-2000:] res = json.loads(dom.split(marker, 1)[1].split('"', 1)[0].replace(""", '"')) assert res["yamlUnknown"] == "rejected" and res["yaml"] is True assert res["failAfterApply"] == 6 # callRate 0.9985 fails every synthetic sample (max 0.99) assert res["rowsWithSelection"] == 3 and res["linkedPlots"] > 0