101 lines
3.5 KiB
Python
101 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
|
|
from apps.web import pipeline_backend as backend
|
|
from apps.web.vue3 import main as api
|
|
from gfa_core.mas.tools.stage_status import get_stage_status
|
|
|
|
|
|
def _upstream(workspace: Path) -> tuple[Path, str]:
|
|
final = workspace / "final"
|
|
final.mkdir(parents=True)
|
|
vcf = final / "cohort.gwas_fmt.sorted.vcf.gz"
|
|
vcf.write_bytes(b"sorted")
|
|
sha = hashlib.sha256(vcf.read_bytes()).hexdigest()
|
|
scope = "genotype:" + "a" * 64
|
|
manifest = {
|
|
"manifest_version": "1.0",
|
|
"stage": "gwas_fmt",
|
|
"status": "success",
|
|
"prefix": "cohort",
|
|
"genotype_scope_id": scope,
|
|
"artifacts": {
|
|
"sorted_vcf": {
|
|
"role": "sorted_vcf",
|
|
"path": str(vcf.relative_to(workspace)),
|
|
"full_path": str(vcf.resolve()),
|
|
"size": vcf.stat().st_size,
|
|
"sha256": sha,
|
|
}
|
|
},
|
|
}
|
|
(final / "cohort.gwas_fmt.manifest.json").write_text(json.dumps(manifest), encoding="utf-8")
|
|
return vcf, scope
|
|
|
|
|
|
def test_backend_resolves_stage19_only_from_gwas_fmt_manifest(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "workspace" / "job"
|
|
vcf, scope = _upstream(workspace)
|
|
decision = backend._resolve_gwas_kinship_input(workspace=workspace, prefix="cohort")
|
|
assert decision["path"] == str(vcf.resolve())
|
|
assert decision["genotype_scope_id"] == scope
|
|
|
|
|
|
def test_backend_builds_only_canonical_stage19_runner_command(tmp_path: Path) -> None:
|
|
command = backend._gwas_kinship_cli_command(
|
|
workspace=tmp_path,
|
|
job_id="job",
|
|
prefix="cohort",
|
|
sorted_vcf=tmp_path / "sorted.vcf.gz",
|
|
genotype_scope_id="genotype:" + "a" * 64,
|
|
tassel_pipeline="/opt/tassel/run_pipeline.pl",
|
|
xmx="10g",
|
|
log_file=tmp_path / "logs" / "gwas_kinship.log",
|
|
)
|
|
joined = " ".join(command)
|
|
assert "-m apps.cli.run_gwas_kinship" in joined
|
|
assert "--genotype-scope-id genotype:" in joined
|
|
assert "-ck" not in command
|
|
assert command.count("apps.cli.run_gwas_kinship") == 1
|
|
|
|
|
|
def test_api_preflight_short_circuits_gene_family_resolution(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
workspace_root = tmp_path / "workspace"
|
|
_upstream(workspace_root / "job")
|
|
monkeypatch.setattr(api, "WORKSPACE_ROOT", workspace_root)
|
|
result = api._api_semantic_preflight("job", "gwas_kinship", "cohort", {})
|
|
assert result["valid"] is True
|
|
assert result["issues"] == []
|
|
assert result["decisions"]["sorted_vcf"]["producer_stage"] == "gwas_fmt"
|
|
|
|
|
|
def test_status_finds_stage_then_identity_manifest_filename(tmp_path: Path) -> None:
|
|
workspace_root = tmp_path / "workspace"
|
|
final = workspace_root / "job" / "final"
|
|
final.mkdir(parents=True)
|
|
manifest_path = final / "cohort.gwas_kinship.123456789abc.manifest.json"
|
|
manifest_path.write_text(json.dumps({
|
|
"stage": "gwas_kinship",
|
|
"prefix": "cohort",
|
|
"status": "success",
|
|
"genotype_scope_id": "genotype:" + "a" * 64,
|
|
"analysis_scope_id": "gwas-analysis:" + "1" * 64,
|
|
"artifacts": {},
|
|
}), encoding="utf-8")
|
|
report = get_stage_status(
|
|
job_id="job",
|
|
prefix="cohort",
|
|
stage="gwas_kinship",
|
|
workspace_root=workspace_root,
|
|
require_artifacts=False,
|
|
strict_identity=True,
|
|
genotype_scope_id="genotype:" + "a" * 64,
|
|
)
|
|
assert report.manifest_path == manifest_path
|
|
assert report.completed is True
|