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"