107 lines
3.9 KiB
Python
107 lines
3.9 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from gfa_core.gwas_upstream_manifest import GwasUpstreamError, resolve_upstream_artifact
|
|
|
|
|
|
def _write_manifest(workspace: Path, *, scope: str = "genotype:scope", declared_sha: str = "") -> Path:
|
|
final = workspace / "final"
|
|
final.mkdir(parents=True)
|
|
artifact = final / "cohort.gwas_fmt.sorted.vcf.gz"
|
|
artifact.write_bytes(b"sorted-vcf")
|
|
sha = declared_sha or hashlib.sha256(artifact.read_bytes()).hexdigest()
|
|
manifest = {
|
|
"manifest_version": "1.0",
|
|
"stage": "gwas_fmt",
|
|
"status": "success",
|
|
"prefix": "cohort",
|
|
"genotype_scope_id": scope,
|
|
"artifacts": {
|
|
"sorted_vcf": {
|
|
"role": "sorted_vcf",
|
|
"path": str(artifact.relative_to(workspace)),
|
|
"full_path": str(artifact.resolve()),
|
|
"size": artifact.stat().st_size,
|
|
"sha256": sha,
|
|
}
|
|
},
|
|
}
|
|
path = final / "cohort.gwas_fmt.manifest.json"
|
|
path.write_text(json.dumps(manifest), encoding="utf-8")
|
|
return path
|
|
|
|
|
|
def test_resolver_returns_exact_manifest_artifact_and_scope(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "workspace" / "job"
|
|
_write_manifest(workspace)
|
|
decision = resolve_upstream_artifact(
|
|
workspace=workspace,
|
|
prefix="cohort",
|
|
producer_stage="gwas_fmt",
|
|
artifact_role="sorted_vcf",
|
|
)
|
|
assert decision["resolved"] is True
|
|
assert decision["genotype_scope_id"] == "genotype:scope"
|
|
assert Path(decision["path"]).read_bytes() == b"sorted-vcf"
|
|
|
|
|
|
def test_relevant_manifest_hash_mismatch_fails_closed(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "workspace" / "job"
|
|
_write_manifest(workspace, declared_sha="0" * 64)
|
|
with pytest.raises(GwasUpstreamError, match="sha256 mismatch"):
|
|
resolve_upstream_artifact(
|
|
workspace=workspace,
|
|
prefix="cohort",
|
|
producer_stage="gwas_fmt",
|
|
artifact_role="sorted_vcf",
|
|
)
|
|
|
|
|
|
def test_expected_scope_mismatch_does_not_fall_back(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "workspace" / "job"
|
|
_write_manifest(workspace, scope="genotype:first")
|
|
with pytest.raises(GwasUpstreamError, match="scope mismatch"):
|
|
resolve_upstream_artifact(
|
|
workspace=workspace,
|
|
prefix="cohort",
|
|
producer_stage="gwas_fmt",
|
|
artifact_role="sorted_vcf",
|
|
expected_scopes={"genotype_scope_id": "genotype:second"},
|
|
)
|
|
|
|
|
|
def test_missing_relevant_manifest_is_reported_without_directory_scan(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "workspace" / "job"
|
|
(workspace / "final").mkdir(parents=True)
|
|
(workspace / "final" / "looks.sorted.vcf.gz").write_bytes(b"not-authoritative")
|
|
with pytest.raises(GwasUpstreamError, match="manifest is missing"):
|
|
resolve_upstream_artifact(
|
|
workspace=workspace,
|
|
prefix="cohort",
|
|
producer_stage="gwas_fmt",
|
|
artifact_role="sorted_vcf",
|
|
)
|
|
|
|
|
|
def test_resolver_selects_exact_scope_when_sibling_identity_exists(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "workspace" / "job"
|
|
first_path = _write_manifest(workspace, scope="genotype:" + "a" * 64)
|
|
first_manifest = json.loads(first_path.read_text(encoding="utf-8"))
|
|
first_path.rename(workspace / "final" / "cohort.gwas_fmt.aaaaaaaaaaaa.manifest.json")
|
|
second_manifest = {**first_manifest, "genotype_scope_id": "genotype:" + "b" * 64}
|
|
(workspace / "final" / "cohort.gwas_fmt.bbbbbbbbbbbb.manifest.json").write_text(
|
|
json.dumps(second_manifest), encoding="utf-8",
|
|
)
|
|
|
|
resolved = resolve_upstream_artifact(
|
|
workspace=workspace, prefix="cohort", producer_stage="gwas_fmt",
|
|
artifact_role="sorted_vcf",
|
|
expected_scopes={"genotype_scope_id": "genotype:" + "a" * 64},
|
|
)
|
|
assert resolved["genotype_scope_id"] == "genotype:" + "a" * 64
|