105 lines
4.0 KiB
Python
105 lines
4.0 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from gfa_core.gwas_legacy_lineage import LegacyLineageError, validate_legacy_lineage
|
|
|
|
|
|
def _sha(path: Path) -> str:
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
|
|
|
def _write_fixture(tmp_path: Path) -> Path:
|
|
parent = tmp_path / "parent.vcf"
|
|
parent.write_text(
|
|
"##fileformat=VCFv4.2\n"
|
|
"#CHROM\tPOS\tID\tREF\tALT\tQUAL\tFILTER\tINFO\tFORMAT\tS1\tS2\n"
|
|
"1\t10\t.\tA\tG\t.\tPASS\t.\tGT\t0/0\t0/1\n",
|
|
encoding="utf-8",
|
|
)
|
|
fam = tmp_path / "cohort.fam"
|
|
fam.write_text("F1 S1 0 0 0 -9\nF2 S2 0 0 0 -9\n", encoding="utf-8")
|
|
q = tmp_path / "cohort.2.Q"
|
|
q.write_text("0.6 0.4\n0.2 0.8\n", encoding="utf-8")
|
|
script = tmp_path / "pop_structure.sh"
|
|
script.write_text("plink --indep-pairwise 50 10 0.2\n", encoding="utf-8")
|
|
log = tmp_path / "structure.log"
|
|
log.write_text("PLINK v1.90b6.21\nADMIXTURE Version 1.3.0\n", encoding="utf-8")
|
|
ordered_ids = "S1\nS2\n".encode()
|
|
manifest = {
|
|
"schema_version": "legacy_lineage.v1",
|
|
"stage": "pop_struct",
|
|
"status": "success",
|
|
"audited_at": "2026-07-21T00:00:00Z",
|
|
"source_genotype_sha256": _sha(parent),
|
|
"parent_genotype": {
|
|
"path": str(parent.resolve()),
|
|
"sha256": _sha(parent),
|
|
"sample_count": 2,
|
|
"variant_count": 1,
|
|
},
|
|
"sample_identity": {
|
|
"fam_id_column": "IID",
|
|
"count": 2,
|
|
"ordered_iid_sha256": hashlib.sha256(ordered_ids).hexdigest(),
|
|
},
|
|
"method": {
|
|
"ld_pruning": {"window_variants": 50, "step_variants": 10, "r2": 0.2},
|
|
"plink_version": "1.90b6.21",
|
|
"admixture_version": "1.3.0",
|
|
"admixture_k": 2,
|
|
},
|
|
"artifacts": {
|
|
"sample_fam": {"path": str(fam.resolve()), "sha256": _sha(fam)},
|
|
"admixture_q": {"path": str(q.resolve()), "sha256": _sha(q)},
|
|
"source_script": {"path": str(script.resolve()), "sha256": _sha(script)},
|
|
"execution_log": {"path": str(log.resolve()), "sha256": _sha(log)},
|
|
},
|
|
}
|
|
path = tmp_path / "legacy_lineage.manifest.json"
|
|
path.write_text(json.dumps(manifest), encoding="utf-8")
|
|
return path
|
|
|
|
|
|
def test_valid_legacy_lineage_proves_parent_samples_and_q(tmp_path: Path) -> None:
|
|
result = validate_legacy_lineage(_write_fixture(tmp_path), expected_genotype_sha256=None)
|
|
assert result["source_genotype_sha256"] == result["parent_genotype"]["sha256"]
|
|
assert result["sample_identity"]["count"] == 2
|
|
|
|
|
|
def test_legacy_lineage_allows_additional_audit_evidence(tmp_path: Path) -> None:
|
|
path = _write_fixture(tmp_path)
|
|
data = json.loads(path.read_text())
|
|
data["method"]["ld_pruning"]["retained_variants"] = 1
|
|
path.write_text(json.dumps(data), encoding="utf-8")
|
|
assert validate_legacy_lineage(path)["method"]["ld_pruning"]["retained_variants"] == 1
|
|
|
|
|
|
def test_legacy_lineage_rejects_artifact_hash_drift(tmp_path: Path) -> None:
|
|
path = _write_fixture(tmp_path)
|
|
data = json.loads(path.read_text())
|
|
Path(data["artifacts"]["admixture_q"]["path"]).write_text("0.5 0.5\n", encoding="utf-8")
|
|
with pytest.raises(LegacyLineageError, match="sha256 mismatch"):
|
|
validate_legacy_lineage(path)
|
|
|
|
|
|
def test_legacy_lineage_rejects_q_fam_row_mismatch(tmp_path: Path) -> None:
|
|
path = _write_fixture(tmp_path)
|
|
data = json.loads(path.read_text())
|
|
q_path = Path(data["artifacts"]["admixture_q"]["path"])
|
|
q_path.write_text("0.6 0.4\n", encoding="utf-8")
|
|
data["artifacts"]["admixture_q"]["sha256"] = _sha(q_path)
|
|
path.write_text(json.dumps(data), encoding="utf-8")
|
|
with pytest.raises(LegacyLineageError, match="Q/FAM row count mismatch"):
|
|
validate_legacy_lineage(path)
|
|
|
|
|
|
def test_legacy_lineage_rejects_wrong_expected_parent(tmp_path: Path) -> None:
|
|
path = _write_fixture(tmp_path)
|
|
with pytest.raises(LegacyLineageError, match="expected genotype"):
|
|
validate_legacy_lineage(path, expected_genotype_sha256="0" * 64)
|