156 lines
6.5 KiB
Python
156 lines
6.5 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import json
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from gfa_core.genotype_context import build_population_analysis_scope_id
|
||
|
|
from gfa_core.mas.stage_capabilities import CAPABILITIES
|
||
|
|
from gfa_core.mas.stage_catalog import PIPELINE_STAGES, STAGE_DEPENDENCIES, STAGE_RESOURCE_PROFILES
|
||
|
|
from gfa_core.mas.tools.stage_identity import StageIdentity, get_stage_status_for_identity, resolve_stage_identity
|
||
|
|
from apps.web.pipeline_backend import _resolve_population_vcf
|
||
|
|
|
||
|
|
|
||
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[2]
|
||
|
|
POPULATION_STAGES = {
|
||
|
|
"pop_pca": ("medium", 4),
|
||
|
|
"pop_phylogeny": ("heavy", 6),
|
||
|
|
"pop_structure": ("heavy", 6),
|
||
|
|
"pop_ld_decay": ("heavy", 6),
|
||
|
|
"pop_snp_density": ("light", 4),
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def test_population_stages_have_root_catalog_capability_and_contract() -> None:
|
||
|
|
for stage, (resource_class, cpu) in POPULATION_STAGES.items():
|
||
|
|
assert stage in PIPELINE_STAGES
|
||
|
|
assert STAGE_DEPENDENCIES[stage] == []
|
||
|
|
assert STAGE_RESOURCE_PROFILES[stage]["resource_class"] == resource_class
|
||
|
|
assert STAGE_RESOURCE_PROFILES[stage]["default_cpu"] == cpu
|
||
|
|
assert CAPABILITIES[stage].category == "population_genomics"
|
||
|
|
assert [need.role for need in CAPABILITIES[stage].consumes][0] == "population_vcf"
|
||
|
|
assert (PROJECT_ROOT / "gfa_core" / "schemas" / "stages" / f"{stage}_contract_v1.md").is_file()
|
||
|
|
|
||
|
|
|
||
|
|
def test_population_vcf_resolves_unique_input_file(tmp_path: Path) -> None:
|
||
|
|
input_dir = tmp_path / "input" / "job"
|
||
|
|
final_dir = tmp_path / "workspace" / "job" / "final"
|
||
|
|
input_dir.mkdir(parents=True)
|
||
|
|
final_dir.mkdir(parents=True)
|
||
|
|
vcf = input_dir / "population_11samples.snp.vcf.gz"
|
||
|
|
vcf.write_bytes(b"test-vcf")
|
||
|
|
assert _resolve_population_vcf(explicit="", input_dir=input_dir, final_dir=final_dir) == vcf.resolve()
|
||
|
|
|
||
|
|
|
||
|
|
def test_population_vcf_does_not_guess_between_inputs(tmp_path: Path) -> None:
|
||
|
|
input_dir = tmp_path / "input"
|
||
|
|
final_dir = tmp_path / "final"
|
||
|
|
input_dir.mkdir()
|
||
|
|
final_dir.mkdir()
|
||
|
|
(input_dir / "a.vcf").write_bytes(b"a")
|
||
|
|
(input_dir / "b.vcf.gz").write_bytes(b"b")
|
||
|
|
with pytest.raises(ValueError, match="Multiple population VCFs"):
|
||
|
|
_resolve_population_vcf(explicit="", input_dir=input_dir, final_dir=final_dir)
|
||
|
|
|
||
|
|
|
||
|
|
def test_population_vcf_resolves_successful_final_manifest(tmp_path: Path) -> None:
|
||
|
|
input_dir = tmp_path / "input"
|
||
|
|
final_dir = tmp_path / "workspace" / "job" / "final"
|
||
|
|
input_dir.mkdir(parents=True)
|
||
|
|
final_dir.mkdir(parents=True)
|
||
|
|
vcf = final_dir / "upstream.vcf.gz"
|
||
|
|
vcf.write_bytes(b"upstream-vcf")
|
||
|
|
manifest = {
|
||
|
|
"status": "success",
|
||
|
|
"stage": "pop_pca",
|
||
|
|
"artifacts": {"population_vcf": {"path": "final/upstream.vcf.gz"}},
|
||
|
|
}
|
||
|
|
(final_dir / "upstream.manifest.json").write_text(json.dumps(manifest), encoding="utf-8")
|
||
|
|
assert _resolve_population_vcf(explicit="", input_dir=input_dir, final_dir=final_dir) == vcf.resolve()
|
||
|
|
|
||
|
|
|
||
|
|
def test_population_analysis_scope_is_canonical_and_sensitive() -> None:
|
||
|
|
kwargs = {
|
||
|
|
"stage": "pop_pca",
|
||
|
|
"genotype_scope_id": "genotype:" + "a" * 64,
|
||
|
|
"input_artifact_hashes": {"population_vcf": "b" * 64},
|
||
|
|
"parameters": {"ld_r2": 0.2, "ld_window_variants": 50},
|
||
|
|
"method_version": "plink_pca.v1",
|
||
|
|
"contract_version": "v1",
|
||
|
|
"tool_versions": {"plink": "1.90b6.21"},
|
||
|
|
}
|
||
|
|
first = build_population_analysis_scope_id(**kwargs)
|
||
|
|
same = build_population_analysis_scope_id(**dict(reversed(list(kwargs.items()))))
|
||
|
|
changed_params = build_population_analysis_scope_id(**{**kwargs, "parameters": {"ld_r2": 0.1, "ld_window_variants": 50}})
|
||
|
|
changed_hash = build_population_analysis_scope_id(**{**kwargs, "input_artifact_hashes": {"population_vcf": "c" * 64}})
|
||
|
|
assert first == same
|
||
|
|
assert first.startswith("population-analysis:")
|
||
|
|
assert changed_params != first
|
||
|
|
assert changed_hash != first
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize("missing", ["genotype_scope_id", "analysis_scope_id"])
|
||
|
|
def test_population_status_requires_complete_scientific_identity(tmp_path: Path, missing: str) -> None:
|
||
|
|
identity_values = {
|
||
|
|
"genotype_scope_id": "genotype:" + "a" * 64,
|
||
|
|
"analysis_scope_id": "population-analysis:" + "b" * 64,
|
||
|
|
}
|
||
|
|
identity_values[missing] = ""
|
||
|
|
report = get_stage_status_for_identity(
|
||
|
|
"job",
|
||
|
|
StageIdentity(stage="pop_pca", prefix="cohort", **identity_values),
|
||
|
|
workspace_root=tmp_path / "workspace",
|
||
|
|
require_artifacts=False,
|
||
|
|
)
|
||
|
|
assert report.completed is False
|
||
|
|
assert "both genotype_scope_id and analysis_scope_id" in " ".join(report.errors)
|
||
|
|
|
||
|
|
|
||
|
|
def test_population_status_selects_exact_analysis_scope(tmp_path: Path) -> None:
|
||
|
|
final = tmp_path / "workspace" / "job" / "final"
|
||
|
|
final.mkdir(parents=True)
|
||
|
|
genotype = "genotype:" + "a" * 64
|
||
|
|
wanted = "population-analysis:" + "b" * 64
|
||
|
|
other = "population-analysis:" + "c" * 64
|
||
|
|
for digest, scope in (("bbbbbbbbbbbb", wanted), ("cccccccccccc", other)):
|
||
|
|
manifest = {
|
||
|
|
"manifest_version": "1.0",
|
||
|
|
"stage": "pop_pca",
|
||
|
|
"prefix": "cohort",
|
||
|
|
"status": "success",
|
||
|
|
"genotype_scope_id": genotype,
|
||
|
|
"analysis_scope_id": scope,
|
||
|
|
"artifacts": {},
|
||
|
|
}
|
||
|
|
(final / f"cohort.pop_pca.{digest}.manifest.json").write_text(json.dumps(manifest), encoding="utf-8")
|
||
|
|
|
||
|
|
identity = resolve_stage_identity(
|
||
|
|
"pop_pca",
|
||
|
|
"cohort",
|
||
|
|
{"genotype_scope_id": genotype, "analysis_scope_id": wanted},
|
||
|
|
)
|
||
|
|
report = get_stage_status_for_identity(
|
||
|
|
"job", identity, workspace_root=tmp_path / "workspace", require_artifacts=False
|
||
|
|
)
|
||
|
|
assert report.completed is True
|
||
|
|
assert report.manifest is not None
|
||
|
|
assert report.manifest["analysis_scope_id"] == wanted
|
||
|
|
|
||
|
|
failed_manifest = {
|
||
|
|
"manifest_version": "1.0", "stage": "pop_pca", "prefix": "cohort",
|
||
|
|
"status": "failed", "genotype_scope_id": genotype,
|
||
|
|
"analysis_scope_id": wanted, "artifacts": {}, "error": "controlled retry failure",
|
||
|
|
}
|
||
|
|
failed_path = final / "cohort.pop_pca.failed.manifest.json"
|
||
|
|
failed_path.write_text(json.dumps(failed_manifest), encoding="utf-8")
|
||
|
|
newest = max(path.stat().st_mtime for path in final.glob("*.manifest.json")) + 1
|
||
|
|
failed_path.touch()
|
||
|
|
import os
|
||
|
|
os.utime(failed_path, (newest, newest))
|
||
|
|
retry_report = get_stage_status_for_identity(
|
||
|
|
"job", identity, workspace_root=tmp_path / "workspace", require_artifacts=False
|
||
|
|
)
|
||
|
|
assert retry_report.completed is False
|
||
|
|
assert retry_report.manifest_status == "failed"
|