76 lines
3.3 KiB
Python
76 lines
3.3 KiB
Python
"""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"]
|