135 lines
4.4 KiB
Python
135 lines
4.4 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from gfa_core.population_manifest import (
|
|
PopulationManifestError,
|
|
resolve_population_manifest_identity,
|
|
)
|
|
|
|
|
|
GENOTYPE = "genotype:" + "a" * 64
|
|
ANALYSIS = "population-analysis:" + "b" * 64
|
|
|
|
|
|
def _write_manifest(
|
|
workspace: Path,
|
|
*,
|
|
analysis_scope_id: str = ANALYSIS,
|
|
status: str = "success",
|
|
start_time: str = "2026-07-24T00:00:00+00:00",
|
|
) -> Path:
|
|
final = workspace / "final"
|
|
final.mkdir(parents=True, exist_ok=True)
|
|
artifact = final / f"cohort.pop_pca.{analysis_scope_id[-12:]}.scores.tsv"
|
|
artifact.write_text("sample\tPC1\nS1\t0.0\n", encoding="utf-8")
|
|
manifest = {
|
|
"manifest_version": "1.0",
|
|
"stage": "pop_pca",
|
|
"prefix": "cohort",
|
|
"status": status,
|
|
"start_time": start_time,
|
|
"run_id": f"run-{analysis_scope_id[-12:]}",
|
|
"genotype_scope_id": GENOTYPE,
|
|
"analysis_scope_id": analysis_scope_id,
|
|
"inputs": [{"role": "population_vcf", "sha256": "a" * 64}],
|
|
"artifacts": {} if status != "success" else {
|
|
"pca_scores": {
|
|
"artifact": "pca_scores",
|
|
"role": "pca_scores",
|
|
"path": str(artifact.relative_to(workspace)),
|
|
"full_path": str(artifact.resolve()),
|
|
"type": "text/tab-separated-values",
|
|
"size": artifact.stat().st_size,
|
|
"sha256": hashlib.sha256(artifact.read_bytes()).hexdigest(),
|
|
}
|
|
},
|
|
}
|
|
path = final / f"cohort.pop_pca.{analysis_scope_id[-12:]}.{status}.manifest.json"
|
|
path.write_text(json.dumps(manifest), encoding="utf-8")
|
|
return path
|
|
|
|
|
|
def test_empty_population_scope_materializes_from_one_authoritative_manifest(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "workspace" / "job"
|
|
manifest_path = _write_manifest(workspace)
|
|
|
|
resolved = resolve_population_manifest_identity(
|
|
workspace=workspace,
|
|
stage="pop_pca",
|
|
prefix="cohort",
|
|
genotype_scope_id="",
|
|
analysis_scope_id="",
|
|
expected_vcf_sha256="a" * 64,
|
|
)
|
|
|
|
assert resolved == {
|
|
"genotypeScopeId": GENOTYPE,
|
|
"analysisScopeId": ANALYSIS,
|
|
"runId": "run-" + ANALYSIS[-12:],
|
|
"sourceManifest": str(manifest_path),
|
|
}
|
|
|
|
|
|
def test_partial_population_scope_is_rejected(tmp_path: Path) -> None:
|
|
with pytest.raises(PopulationManifestError, match="both empty or both complete"):
|
|
resolve_population_manifest_identity(
|
|
workspace=tmp_path,
|
|
stage="pop_pca",
|
|
prefix="cohort",
|
|
genotype_scope_id=GENOTYPE,
|
|
analysis_scope_id="",
|
|
)
|
|
|
|
|
|
def test_multiple_population_analysis_identities_are_rejected(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "workspace" / "job"
|
|
_write_manifest(workspace)
|
|
_write_manifest(workspace, analysis_scope_id="population-analysis:" + "c" * 64)
|
|
|
|
with pytest.raises(PopulationManifestError, match="multiple population manifests"):
|
|
resolve_population_manifest_identity(
|
|
workspace=workspace,
|
|
stage="pop_pca",
|
|
prefix="cohort",
|
|
expected_vcf_sha256="a" * 64,
|
|
)
|
|
|
|
|
|
def test_tampered_population_artifact_is_rejected(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "workspace" / "job"
|
|
manifest_path = _write_manifest(workspace)
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
artifact = workspace / manifest["artifacts"]["pca_scores"]["path"]
|
|
artifact.write_text("sample\tPC1\nS1\t9.9\n", encoding="utf-8")
|
|
|
|
with pytest.raises(PopulationManifestError, match="sha256 mismatch"):
|
|
resolve_population_manifest_identity(
|
|
workspace=workspace,
|
|
stage="pop_pca",
|
|
prefix="cohort",
|
|
expected_vcf_sha256="a" * 64,
|
|
)
|
|
|
|
|
|
def test_newer_failed_attempt_blocks_older_success_for_same_identity(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "workspace" / "job"
|
|
_write_manifest(workspace)
|
|
_write_manifest(
|
|
workspace,
|
|
status="failed",
|
|
start_time="2026-07-24T01:00:00+00:00",
|
|
)
|
|
|
|
with pytest.raises(PopulationManifestError, match="latest population manifest is not successful"):
|
|
resolve_population_manifest_identity(
|
|
workspace=workspace,
|
|
stage="pop_pca",
|
|
prefix="cohort",
|
|
expected_vcf_sha256="a" * 64,
|
|
)
|