tests: PGx demo oracle tests — definition sanity (PGX-023), limitations (PGX-009), determinism (PGX-029), JSON/TSV (PGX-028)

tests/oracle/test_pgx_demo.py

@@ -0,0 +1,75 @@
+"""PGx on the GSA v4 PGx demo (30 CEPH individuals) — definition sanity (PGX-023), array limitations (PGX-009),
+determinism (PGX-029) and JSON ↔ TSV equivalence (PGX-028).
+
+Needs the local PGx DB (``simurg pgx db build`` with the GSA v4 PGx manifest, see docs/validation/V2_pgx_star.md) and
+the demo SNV + PGx CN VCFs under ``data/pgx/demo``; skipped when absent.
+"""
+
+from __future__ import annotations
+
+import csv
+import json
+import sqlite3
+from pathlib import Path
+
+import pytest
+from tests.conftest import ROOT
+from typer.testing import CliRunner
+
+from simurg.cli.main import app
+
+pytestmark = pytest.mark.data
+DEMO = ROOT / "data" / "pgx" / "demo"
+DBDIR = ROOT / "data" / "pgx" / "dbm"
+DBS = sorted(DBDIR.glob("pgx_db-*.sqlite")) if DBDIR.exists() else []
+MANIFEST = "GSA-PGx-48v4-0_20079540_E2.bpm"
+
+
+def _need() -> Path:
+    if not DBS or not (DEMO / "vcf").exists() or not (DEMO / "call").exists():
+        pytest.skip("PGx demo DB/VCFs not present")
+    return DBS[-1]
+
+
+def test_definition_sanity() -> None:  # PGX-023
+    db = _need()
+    con = sqlite3.connect(db)
+    assert con.execute("SELECT COUNT(*) FROM allele_variants WHERE gene = 'CYP2A6' AND allele = '*1'").fetchone() == (
+        0,
+    )
+    from simurg.pgx.phenotype import build_table, phenotype
+
+    rows = con.execute("SELECT lookup_key, activity_score, phenotype FROM phenotypes WHERE gene = 'ABCG2'").fetchall()
+    alle = con.execute("SELECT allele, function, activity FROM alleles WHERE gene = 'ABCG2'").fetchall()
+    ref = next(a for a, *_ in alle if "reference" in a)
+    assert phenotype(build_table("ABCG2", rows, alle), (ref, ref)).phenotype.startswith("Normal")
+
+
+def test_known_limitations() -> None:  # PGX-009 regression oracles that hold for our definitions
+    from simurg.pipeline.pgx_db import coverage_rows
+
+    st = {(r["gene"], r["allele"]): r["status"] for r in coverage_rows(_need(), MANIFEST)}
+    assert st[("CYP4F2", "*2")] == "not_covered"  # reference: CYP4F2 *2 reported as *1
+    assert st[("CYP4F2", "*4")] == "equivalent_to:*3"  # reference: *4 reported as *3
+
+
+def test_determinism_and_json_tsv(tmp_path: Path) -> None:  # PGX-029, PGX-028
+    db = _need()
+    run = CliRunner()
+    outs = []
+    for k in range(2):
+        out = tmp_path / f"s{k}"
+        r = run.invoke(app, ["pgx", "star-allele", "call", "--vcfs", str(DEMO / "vcf"), "--vcfs", str(DEMO / "call"),
+                             "--db", str(db), "--out", str(out), "--gene", "CYP2D6", "--gene", "CYP2C19", "--quiet"])  # fmt: skip
+        assert r.exit_code == 0, r.output
+        r = run.invoke(app, ["pgx", "phenotype", "--calls", str(out), "--db", str(db), "--quiet"])
+        assert r.exit_code == 0, r.output
+        outs.append(out)
+    for name in ("diplotype_summary.tsv", "supporting_variants.tsv", "diplotype_annotations.tsv", "pgx_calls.json"):
+        assert (outs[0] / name).read_bytes() == (outs[1] / name).read_bytes(), name
+    with (outs[0] / "diplotype_annotations.tsv").open() as fh:
+        rows = list(csv.DictReader((ln for ln in fh if not ln.startswith("#")), delimiter="\t"))
+    for r in rows:
+        doc = json.loads((outs[0] / f"{r['sample_id']}.pgx.json").read_text())
+        g = next(x for x in doc["genes"] if x["gene"] == r["gene"])
+        assert g["genotype"] == r["genotype"] and (g["phenotype"] or "") == r["phenotype"]