test_p6_vcf: stream the gtc2vcf oracle output (OUT-012)

The INFO/extra-FORMAT comparison held ~700 000 gtc2vcf records as text plus two
dicts each (> 4.8 GB, stopped by the Pi memory guard). Now written to a temp
file and kept as compact tuples: peak 3.0 GB; all 5 tests pass.
This commit is contained in:
kemal committed 2026-09-30 01:25:32 +01:00
1 parent 08e67cff29
commit 12ca5da62d
1 file changed
+17 -10
+17 -10
View File
@@ -178,18 +178,24 @@ def test_info_and_extra_format_match_gtc2vcf(tmp_path: Path) -> None: # OUT-012
) # fmt: skip
assert res.exit_code == 0, res.output
env = dict(os.environ, BCFTOOLS_PLUGINS=str(plugin.parent))
g2v = subprocess.run(
g2v = tmp_path / "g2v.vcf"
with g2v.open("w") as fh: # streamed: the text of ~700 000 records would cost GBs held as dicts
subprocess.run(
[str(BCFTOOLS), "+gtc2vcf", "--no-version", "-t", "GT,X,Y,NORMX,NORMY,THETA,R", "-b", str(GSA_BPM), "-c",
str(GSA_CSV), "-e", str(GSA_EGT), "-f", str(FASTA), "-g", str(gtcs), "-O", "v"],
capture_output=True, text=True, env=env, check=True,
).stdout # fmt: skip
ref = {}
for line in g2v.splitlines():
stdout=fh, env=env, check=True,
) # fmt: skip
keys = ("X", "Y", "NORMX", "NORMY", "THETA", "R")
ref: dict[str, tuple[str, ...]] = {} # id → (ASSAY_TYPE, GenTrain_Score, Cluster_Sep, *keys)
with g2v.open() as fh:
for line in fh:
if line.startswith("#"):
continue
c = line.split("\t")
c = line.rstrip("\n").split("\t")
info = dict(kv.split("=", 1) for kv in c[7].split(";") if "=" in kv)
ref[c[2]] = (info, dict(zip(c[8].split(":"), c[9].split(":"), strict=True)))
fmt = dict(zip(c[8].split(":"), c[9].split(":"), strict=True))
ref[c[2]] = (info["ASSAY_TYPE"], info["GenTrain_Score"], info["Cluster_Sep"], *(fmt[k] for k in keys))
g2v.unlink()
def close(a: str, b: str) -> bool:
na, nb = a in (".", "") or "nan" in a.lower(), b in (".", "") or "nan" in b.lower()
@@ -207,9 +213,10 @@ def test_info_and_extra_format_match_gtc2vcf(tmp_path: Path) -> None: # OUT-012
continue
info = dict(kv.split("=", 1) for kv in c[7].split(";"))
fmt = dict(zip(c[8].split(":"), c[9].split(":"), strict=True))
ri, rf = ref[c[2]]
assert info["ASSAY_TYPE"] == ri["ASSAY_TYPE"]
assert close(info["GENTRAIN"], ri["GenTrain_Score"]) and close(info["CLUSTER_SEP"], ri["Cluster_Sep"])
at, gt_score, sep, *vals = ref[c[2]]
rf = dict(zip(keys, vals, strict=True))
assert info["ASSAY_TYPE"] == at
assert close(info["GENTRAIN"], gt_score) and close(info["CLUSTER_SEP"], sep)
for k in ("X", "Y", "NORMX", "NORMY"):
assert close(fmt[k], rf[k]), (c[2], k, fmt[k], rf[k])
if fmt["NORMX"] == "0" and fmt["NORMY"] == "0":