SimurgArray/tests/unit/test_qc_report.py

"""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("&quot;", '"'))
    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