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)