326 lines
14 KiB
Python
326 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from gfa_core.gwas_root_inputs import resolve_root_input, resolve_root_pair
|
|
|
|
|
|
def _record(path: Path, **extra: str) -> dict:
|
|
return {
|
|
"path": str(path.resolve()),
|
|
"size": path.stat().st_size,
|
|
"sha256": hashlib.sha256(path.read_bytes()).hexdigest(),
|
|
**extra,
|
|
}
|
|
|
|
|
|
def test_external_binding_allows_large_data_by_reference_and_revalidates_hash(tmp_path: Path) -> None:
|
|
project = tmp_path / "project" / "sandbox"
|
|
workspace = project / "workspace" / "job"
|
|
workspace.mkdir(parents=True)
|
|
phenotype = project.parent / "data" / "phenotype.tsv"
|
|
phenotype.parent.mkdir()
|
|
phenotype.write_text("ID\tvalue\nS1\t1\n", encoding="utf-8")
|
|
binding = {
|
|
"manifest_version": "1.0", "job_id": "job",
|
|
"inputs": {"phenotype_table": _record(phenotype)},
|
|
}
|
|
(workspace / "external_inputs.manifest.json").write_text(json.dumps(binding), encoding="utf-8")
|
|
|
|
resolved = resolve_root_input(project_root=project, job_id="job", role="phenotype_table")
|
|
assert resolved["valid"] is True
|
|
assert resolved["decision"]["path"] == str(phenotype.resolve())
|
|
assert resolved["decision"]["source"] == "external_binding"
|
|
|
|
phenotype.write_text("tampered\n", encoding="utf-8")
|
|
invalid = resolve_root_input(project_root=project, job_id="job", role="phenotype_table")
|
|
assert invalid["valid"] is False
|
|
assert invalid["issues"][0]["code"] == "invalid_external_binding"
|
|
|
|
|
|
@pytest.mark.parametrize("malformed", ["phenotype.tsv", ["phenotype.tsv", None]])
|
|
def test_present_malformed_binding_role_fails_closed(tmp_path: Path, malformed) -> None:
|
|
project = tmp_path / "project" / "sandbox"
|
|
workspace = project / "workspace" / "job"
|
|
fallback = project / "input" / "job" / "fallback.tsv"
|
|
workspace.mkdir(parents=True)
|
|
fallback.parent.mkdir(parents=True)
|
|
fallback.write_text("ID\ttrait\nS1\t1\n", encoding="utf-8")
|
|
binding = {
|
|
"manifest_version": "1.0", "job_id": "job",
|
|
"inputs": {"phenotype_table": malformed},
|
|
}
|
|
(workspace / "external_inputs.manifest.json").write_text(json.dumps(binding), encoding="utf-8")
|
|
|
|
resolved = resolve_root_input(project_root=project, job_id="job", role="phenotype_table")
|
|
|
|
assert resolved["valid"] is False
|
|
assert resolved["needs_confirmation"] is False
|
|
assert resolved["issues"][0]["code"] == "invalid_external_binding"
|
|
|
|
|
|
def test_malformed_binding_inputs_container_fails_closed(tmp_path: Path) -> None:
|
|
project = tmp_path / "project" / "sandbox"
|
|
workspace = project / "workspace" / "job"
|
|
fallback = project / "input" / "job" / "fallback.tsv"
|
|
workspace.mkdir(parents=True)
|
|
fallback.parent.mkdir(parents=True)
|
|
fallback.write_text("ID\ttrait\nS1\t1\n", encoding="utf-8")
|
|
binding = {"manifest_version": "1.0", "job_id": "job", "inputs": "malformed"}
|
|
(workspace / "external_inputs.manifest.json").write_text(json.dumps(binding), encoding="utf-8")
|
|
|
|
resolved = resolve_root_input(project_root=project, job_id="job", role="phenotype_table")
|
|
|
|
assert resolved["valid"] is False
|
|
assert resolved["issues"][0]["code"] == "invalid_external_binding"
|
|
|
|
|
|
def test_mixed_authoritative_role_list_fails_closed(tmp_path: Path) -> None:
|
|
project = tmp_path / "project" / "sandbox"
|
|
workspace = project / "workspace" / "job"
|
|
phenotype = project.parent / "data" / "phenotype.tsv"
|
|
workspace.mkdir(parents=True)
|
|
phenotype.parent.mkdir()
|
|
phenotype.write_text("ID\ttrait\nS1\t1\n", encoding="utf-8")
|
|
binding = {
|
|
"manifest_version": "1.0", "job_id": "job",
|
|
"inputs": {"phenotype_table": [_record(phenotype), "malformed sibling"]},
|
|
}
|
|
(workspace / "external_inputs.manifest.json").write_text(json.dumps(binding), encoding="utf-8")
|
|
|
|
resolved = resolve_root_input(project_root=project, job_id="job", role="phenotype_table")
|
|
|
|
assert resolved["valid"] is False
|
|
assert resolved["issues"][0]["code"] == "invalid_external_binding"
|
|
|
|
|
|
def test_absent_binding_role_still_allows_job_local_scan(tmp_path: Path) -> None:
|
|
project = tmp_path / "project" / "sandbox"
|
|
workspace = project / "workspace" / "job"
|
|
phenotype = project / "input" / "job" / "phenotype.tsv"
|
|
workspace.mkdir(parents=True)
|
|
phenotype.parent.mkdir(parents=True)
|
|
phenotype.write_text("ID\ttrait\nS1\t1\n", encoding="utf-8")
|
|
binding = {"manifest_version": "1.0", "job_id": "job", "inputs": {"sample_fam": {}}}
|
|
(workspace / "external_inputs.manifest.json").write_text(json.dumps(binding), encoding="utf-8")
|
|
|
|
resolved = resolve_root_input(project_root=project, job_id="job", role="phenotype_table")
|
|
|
|
assert resolved["valid"] is True
|
|
assert resolved["decision"]["path"] == str(phenotype.resolve())
|
|
|
|
|
|
def test_mixed_top_level_binding_list_fails_closed(tmp_path: Path) -> None:
|
|
project = tmp_path / "project" / "sandbox"
|
|
workspace = project / "workspace" / "job"
|
|
phenotype = project.parent / "data" / "phenotype.tsv"
|
|
workspace.mkdir(parents=True)
|
|
phenotype.parent.mkdir()
|
|
phenotype.write_text("ID\ttrait\nS1\t1\n", encoding="utf-8")
|
|
binding = {
|
|
"manifest_version": "1.0", "job_id": "job",
|
|
"inputs": [{"role": "phenotype_table", **_record(phenotype)}, "malformed sibling"],
|
|
}
|
|
(workspace / "external_inputs.manifest.json").write_text(json.dumps(binding), encoding="utf-8")
|
|
|
|
resolved = resolve_root_input(project_root=project, job_id="job", role="phenotype_table")
|
|
|
|
assert resolved["valid"] is False
|
|
assert resolved["issues"][0]["code"] == "invalid_external_binding"
|
|
|
|
|
|
def test_valid_top_level_binding_list_selects_requested_role(tmp_path: Path) -> None:
|
|
project = tmp_path / "project" / "sandbox"
|
|
workspace = project / "workspace" / "job"
|
|
phenotype = project.parent / "data" / "phenotype.tsv"
|
|
workspace.mkdir(parents=True)
|
|
phenotype.parent.mkdir()
|
|
phenotype.write_text("ID\ttrait\nS1\t1\n", encoding="utf-8")
|
|
binding = {
|
|
"manifest_version": "1.0", "job_id": "job",
|
|
"inputs": [{"role": "phenotype_table", **_record(phenotype)}, {"role": "sample_fam"}],
|
|
}
|
|
(workspace / "external_inputs.manifest.json").write_text(json.dumps(binding), encoding="utf-8")
|
|
|
|
resolved = resolve_root_input(project_root=project, job_id="job", role="phenotype_table")
|
|
|
|
assert resolved["valid"] is True
|
|
assert resolved["decision"]["path"] == str(phenotype.resolve())
|
|
|
|
|
|
def test_absent_role_in_valid_top_level_list_allows_scan(tmp_path: Path) -> None:
|
|
project = tmp_path / "project" / "sandbox"
|
|
workspace = project / "workspace" / "job"
|
|
phenotype = project / "input" / "job" / "phenotype.tsv"
|
|
workspace.mkdir(parents=True)
|
|
phenotype.parent.mkdir(parents=True)
|
|
phenotype.write_text("ID\ttrait\nS1\t1\n", encoding="utf-8")
|
|
binding = {
|
|
"manifest_version": "1.0", "job_id": "job",
|
|
"inputs": [{"role": "sample_fam"}],
|
|
}
|
|
(workspace / "external_inputs.manifest.json").write_text(json.dumps(binding), encoding="utf-8")
|
|
|
|
resolved = resolve_root_input(project_root=project, job_id="job", role="phenotype_table")
|
|
|
|
assert resolved["valid"] is True
|
|
assert resolved["decision"]["source"] == "input"
|
|
|
|
|
|
def test_csv_is_not_a_valid_phenotype_root_candidate(tmp_path: Path) -> None:
|
|
project = tmp_path / "project" / "sandbox"
|
|
csv_file = project / "input" / "job" / "phenotype.csv"
|
|
csv_file.parent.mkdir(parents=True)
|
|
csv_file.write_text("ID,trait\nS1,1\n", encoding="utf-8")
|
|
|
|
resolved = resolve_root_input(project_root=project, job_id="job", role="phenotype_table")
|
|
|
|
assert resolved["valid"] is False
|
|
assert resolved["issues"][0]["code"] == "missing_root_input"
|
|
|
|
|
|
def test_comma_delimited_txt_is_not_a_valid_phenotype_root_candidate(tmp_path: Path) -> None:
|
|
project = tmp_path / "project" / "sandbox"
|
|
text_file = project / "input" / "job" / "phenotype.txt"
|
|
text_file.parent.mkdir(parents=True)
|
|
text_file.write_text("ID,trait\nS1,1\n", encoding="utf-8")
|
|
|
|
resolved = resolve_root_input(project_root=project, job_id="job", role="phenotype_table")
|
|
|
|
assert resolved["valid"] is False
|
|
assert resolved["issues"][0]["code"] == "missing_root_input"
|
|
|
|
|
|
def test_leading_blank_line_rejects_phenotype_root_candidate(tmp_path: Path) -> None:
|
|
project = tmp_path / "project" / "sandbox"
|
|
phenotype = project / "input" / "job" / "phenotype.tsv"
|
|
phenotype.parent.mkdir(parents=True)
|
|
phenotype.write_text("\nID\ttrait\nS1\t1\n", encoding="utf-8")
|
|
|
|
resolved = resolve_root_input(project_root=project, job_id="job", role="phenotype_table")
|
|
|
|
assert resolved["valid"] is False
|
|
assert resolved["issues"][0]["code"] == "missing_root_input"
|
|
|
|
|
|
def test_scan_deduplicates_content_but_requires_selection_for_distinct_candidates(tmp_path: Path) -> None:
|
|
project = tmp_path / "project" / "sandbox"
|
|
input_dir = project / "input" / "job"
|
|
final_dir = project / "workspace" / "job" / "final"
|
|
input_dir.mkdir(parents=True)
|
|
final_dir.mkdir(parents=True)
|
|
first = input_dir / "a.tsv"
|
|
duplicate = final_dir / "same.tsv"
|
|
second = input_dir / "b.tsv"
|
|
first.write_text("ID\ttrait\nS1\t1\n", encoding="utf-8")
|
|
duplicate.write_bytes(first.read_bytes())
|
|
second.write_text("ID\ttrait\nS1\t2\n", encoding="utf-8")
|
|
|
|
ambiguous = resolve_root_input(project_root=project, job_id="job", role="phenotype_table")
|
|
assert ambiguous["valid"] is False
|
|
assert ambiguous["needs_confirmation"] is True
|
|
assert len(ambiguous["candidates"]) == 2
|
|
|
|
selected = resolve_root_input(
|
|
project_root=project, job_id="job", role="phenotype_table", selected=str(second),
|
|
)
|
|
assert selected["valid"] is True
|
|
assert selected["decision"]["path"] == str(second.resolve())
|
|
|
|
|
|
def _write_legacy_lineage(project: Path, q_file: Path, fam_file: Path) -> str:
|
|
parent = project.parent / "data" / "parent.vcf"
|
|
script = project.parent / "data" / "pop_structure.sh"
|
|
log = project.parent / "data" / "structure.log"
|
|
parent.write_text(
|
|
"##fileformat=VCFv4.2\n"
|
|
"#CHROM\tPOS\tID\tREF\tALT\tQUAL\tFILTER\tINFO\tFORMAT\tI1\n"
|
|
"1\t10\t.\tA\tG\t.\tPASS\t.\tGT\t0/1\n",
|
|
encoding="utf-8",
|
|
)
|
|
script.write_text("plink --indep-pairwise 50 10 0.2\n", encoding="utf-8")
|
|
log.write_text("PLINK v1.90b6.21\nADMIXTURE Version 1.3.0\n", encoding="utf-8")
|
|
parent_sha = hashlib.sha256(parent.read_bytes()).hexdigest()
|
|
lineage = {
|
|
"schema_version": "legacy_lineage.v1", "stage": "pop_struct", "status": "success",
|
|
"source_genotype_sha256": parent_sha,
|
|
"parent_genotype": {
|
|
"path": str(parent.resolve()), "sha256": parent_sha,
|
|
"sample_count": 1, "variant_count": 1,
|
|
},
|
|
"sample_identity": {
|
|
"fam_id_column": "IID", "count": 1,
|
|
"ordered_iid_sha256": hashlib.sha256(b"I1\n").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": _record(fam_file), "admixture_q": _record(q_file),
|
|
"source_script": _record(script), "execution_log": _record(log),
|
|
},
|
|
}
|
|
lineage_path = project / "workspace" / "job" / "legacy_lineage.manifest.json"
|
|
lineage_path.write_text(json.dumps(lineage), encoding="utf-8")
|
|
return "genotype:" + parent_sha
|
|
|
|
|
|
def test_q_fam_pair_rejects_unverified_asserted_source_scope(tmp_path: Path) -> None:
|
|
project = tmp_path / "project" / "sandbox"
|
|
workspace = project / "workspace" / "job"
|
|
workspace.mkdir(parents=True)
|
|
q_file = project.parent / "data" / "cohort.2.Q"
|
|
fam_file = project.parent / "data" / "cohort.fam"
|
|
q_file.parent.mkdir()
|
|
q_file.write_text("0.4 0.6\n", encoding="utf-8")
|
|
fam_file.write_text("F1 I1 0 0 0 -9\n", encoding="utf-8")
|
|
scope = "genotype:" + "a" * 64
|
|
binding = {
|
|
"manifest_version": "1.0", "job_id": "job",
|
|
"inputs": {
|
|
"admixture_q": _record(q_file, pair_id="qk2", source_genotype_scope_id=scope),
|
|
"sample_fam": _record(fam_file, pair_id="qk2", source_genotype_scope_id=scope),
|
|
},
|
|
}
|
|
(workspace / "external_inputs.manifest.json").write_text(json.dumps(binding), encoding="utf-8")
|
|
|
|
resolved = resolve_root_pair(
|
|
project_root=project, job_id="job", first_role="admixture_q", second_role="sample_fam",
|
|
)
|
|
assert resolved["valid"] is False
|
|
assert resolved["needs_confirmation"] is False
|
|
assert resolved["issues"][0]["code"] == "unverified_covariate_lineage"
|
|
|
|
|
|
def test_q_fam_pair_accepts_matching_audited_legacy_lineage(tmp_path: Path) -> None:
|
|
project = tmp_path / "project" / "sandbox"
|
|
workspace = project / "workspace" / "job"
|
|
workspace.mkdir(parents=True)
|
|
q_file = project.parent / "data" / "cohort.2.Q"
|
|
fam_file = project.parent / "data" / "cohort.fam"
|
|
q_file.parent.mkdir()
|
|
q_file.write_text("0.4 0.6\n", encoding="utf-8")
|
|
fam_file.write_text("F1 I1 0 0 0 -9\n", encoding="utf-8")
|
|
binding = {
|
|
"manifest_version": "1.0", "job_id": "job",
|
|
"inputs": {
|
|
"admixture_q": _record(q_file, pair_id="qk2"),
|
|
"sample_fam": _record(fam_file, pair_id="qk2"),
|
|
},
|
|
}
|
|
(workspace / "external_inputs.manifest.json").write_text(json.dumps(binding), encoding="utf-8")
|
|
expected_scope = _write_legacy_lineage(project, q_file, fam_file)
|
|
|
|
resolved = resolve_root_pair(
|
|
project_root=project, job_id="job", first_role="admixture_q", second_role="sample_fam",
|
|
)
|
|
|
|
assert resolved["valid"] is True
|
|
assert resolved["source_genotype_scope_id"] == expected_scope
|
|
assert resolved["first"]["pair_id"] == resolved["second"]["pair_id"] == "qk2"
|