512 lines
24 KiB
Python
512 lines
24 KiB
Python
|
import asyncio
|
||
|
|
import hashlib
|
||
|
|
import inspect
|
||
|
|
import json
|
||
|
|
import subprocess
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from apps.web import pipeline_backend as backend
|
||
|
|
from apps.web.vue3 import main as api
|
||
|
|
from gfa_core.mas.stage_capabilities import get_stage_capability
|
||
|
|
from gfa_core.mas.tools.stage_identity import StageIdentity, get_stage_status_for_identity
|
||
|
|
from gfa_core.project_paths import get_conda_env_bin, get_conda_env_rscript, get_configured_path
|
||
|
|
|
||
|
|
|
||
|
|
GENOTYPE = "genotype:" + "a" * 64
|
||
|
|
PHENOTYPE = "phenotype:" + "b" * 64
|
||
|
|
MODEL = "gwas-model:" + "c" * 64
|
||
|
|
ANALYSIS = "gwas-analysis:" + "d" * 64
|
||
|
|
PLINK = str(get_conda_env_bin("ASAW") / "plink")
|
||
|
|
EMMAX = "/home/xzq/Auto_GWAS/gwas_work/emmax-intel64"
|
||
|
|
OFFICIAL_EMMAX = get_configured_path("GWAS_EMMAX_BIN", "emmax")
|
||
|
|
RSCRIPT = str(get_conda_env_rscript("ASAW"))
|
||
|
|
|
||
|
|
|
||
|
|
def _sha(path: Path) -> str:
|
||
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
||
|
|
|
||
|
|
|
||
|
|
def _artifact(workspace: Path, path: Path, role: str) -> dict:
|
||
|
|
return {
|
||
|
|
"role": role, "artifact": role,
|
||
|
|
"path": str(path.relative_to(workspace)), "full_path": str(path.resolve()),
|
||
|
|
"size": path.stat().st_size, "sha256": _sha(path),
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def _producer_manifests(workspace: Path, prefix: str = "cohort") -> tuple[dict, dict]:
|
||
|
|
final = workspace / "final"
|
||
|
|
final.mkdir(parents=True, exist_ok=True)
|
||
|
|
vcf = final / f"{prefix}.gwas_fmt.sorted.vcf.gz"
|
||
|
|
trait = final / f"{prefix}.gwas_trait.trait.txt"
|
||
|
|
vcf.write_bytes(b"vcf")
|
||
|
|
trait.write_text("<Trait>\theading_days\nS1\t1\nS2\t2\n", encoding="utf-8")
|
||
|
|
fmt = {
|
||
|
|
"manifest_version": "1.0", "run_id": "fmt-run", "stage": "gwas_fmt",
|
||
|
|
"prefix": prefix, "status": "success", "genotype_scope_id": GENOTYPE,
|
||
|
|
"artifacts": {"sorted_vcf": _artifact(workspace, vcf, "sorted_vcf")},
|
||
|
|
}
|
||
|
|
trait_manifest = {
|
||
|
|
"manifest_version": "1.0", "run_id": "trait-run", "stage": "gwas_trait",
|
||
|
|
"prefix": prefix, "status": "success", "phenotype_scope_id": PHENOTYPE,
|
||
|
|
"analysis_scope_id": "gwas-analysis:" + "e" * 64,
|
||
|
|
"artifacts": {"trait_table": _artifact(workspace, trait, "trait_table")},
|
||
|
|
}
|
||
|
|
fmt_path = final / f"{prefix}.gwas_fmt.manifest.json"
|
||
|
|
trait_path = final / f"{prefix}.gwas_trait.identity.manifest.json"
|
||
|
|
fmt_path.write_text(json.dumps(fmt), encoding="utf-8")
|
||
|
|
trait_path.write_text(json.dumps(trait_manifest), encoding="utf-8")
|
||
|
|
return {
|
||
|
|
"path": fmt_path, "artifact": vcf, "manifest": fmt,
|
||
|
|
}, {"path": trait_path, "artifact": trait, "manifest": trait_manifest}
|
||
|
|
|
||
|
|
|
||
|
|
def _emmax_manifest(workspace: Path, prefix: str = "cohort", roles=None) -> Path:
|
||
|
|
final = workspace / "final"
|
||
|
|
final.mkdir(parents=True, exist_ok=True)
|
||
|
|
roles = roles or ("emmax_full", "emmax_pvalues", "report", "environment")
|
||
|
|
artifacts = {}
|
||
|
|
for role in roles:
|
||
|
|
suffix = ".md" if role == "report" else ".txt" if role == "environment" else ".tsv"
|
||
|
|
path = final / f"{prefix}.gwas_emmax.{role}{suffix}"
|
||
|
|
path.write_text(f"{role}\n", encoding="utf-8")
|
||
|
|
artifacts[role] = _artifact(workspace, path, role)
|
||
|
|
manifest = {
|
||
|
|
"manifest_version": "1.0", "stage": "gwas_emmax", "prefix": prefix,
|
||
|
|
"status": "success", "genotype_scope_id": GENOTYPE,
|
||
|
|
"phenotype_scope_id": PHENOTYPE, "model_scope_id": MODEL,
|
||
|
|
"analysis_scope_id": ANALYSIS, "elapsed_sec": 12.34, "artifacts": artifacts,
|
||
|
|
}
|
||
|
|
path = final / f"{prefix}.gwas_emmax.identity.manifest.json"
|
||
|
|
path.write_text(json.dumps(manifest), encoding="utf-8")
|
||
|
|
return path
|
||
|
|
|
||
|
|
|
||
|
|
def _scopes() -> dict:
|
||
|
|
return {"genotypeScopeId": GENOTYPE, "phenotypeScopeId": PHENOTYPE}
|
||
|
|
|
||
|
|
|
||
|
|
def test_emmax_unset_default_is_resolved_server_side_across_entry_points(monkeypatch, tmp_path):
|
||
|
|
workspace = tmp_path / "workspace" / "job"
|
||
|
|
_producer_manifests(workspace)
|
||
|
|
monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace")
|
||
|
|
|
||
|
|
request = api.ExecuteRequest(stage="gwas_emmax")
|
||
|
|
preflight = api._api_semantic_preflight("job", "gwas_emmax", "cohort", _scopes())
|
||
|
|
|
||
|
|
assert request.gwasEmmax == ""
|
||
|
|
assert preflight["materialized_params"]["gwasEmmax"] == OFFICIAL_EMMAX
|
||
|
|
assert inspect.signature(backend.run_stage).parameters["gwas_emmax"].default == OFFICIAL_EMMAX
|
||
|
|
assert "default=GWAS_EMMAX_BIN" in inspect.getsource(
|
||
|
|
__import__("gfa_core.mas.tools.stage_tools", fromlist=["StageToolExecutor"]).StageToolExecutor.execute
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def test_emmax_semantic_preflight_materializes_two_exact_producer_manifests(monkeypatch, tmp_path):
|
||
|
|
workspace = tmp_path / "workspace" / "job"
|
||
|
|
fmt, trait = _producer_manifests(workspace)
|
||
|
|
monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace")
|
||
|
|
|
||
|
|
result = api._api_semantic_preflight("job", "gwas_emmax", "cohort", _scopes())
|
||
|
|
|
||
|
|
assert result["valid"] is True
|
||
|
|
assert result["materialized_params"] == {
|
||
|
|
"gwasSortedVcf": str(fmt["artifact"]), "gwasTraitFile": str(trait["artifact"]),
|
||
|
|
"genotypeScopeId": GENOTYPE, "phenotypeScopeId": PHENOTYPE,
|
||
|
|
"gwasPlink": PLINK, "gwasEmmax": OFFICIAL_EMMAX, "gwasRscript": RSCRIPT,
|
||
|
|
"gwasRequestedPcs": 3,
|
||
|
|
}
|
||
|
|
assert result["decisions"]["sorted_vcf"]["source_manifest"] == str(fmt["path"])
|
||
|
|
assert result["decisions"]["trait_table"]["source_manifest"] == str(trait["path"])
|
||
|
|
|
||
|
|
|
||
|
|
def test_emmax_semantic_preflight_materializes_scopes_from_unique_producer_manifests(
|
||
|
|
monkeypatch, tmp_path
|
||
|
|
):
|
||
|
|
workspace = tmp_path / "workspace" / "job"
|
||
|
|
fmt, trait = _producer_manifests(workspace)
|
||
|
|
monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace")
|
||
|
|
|
||
|
|
result = api._api_semantic_preflight("job", "gwas_emmax", "cohort", {})
|
||
|
|
|
||
|
|
assert result["valid"] is True
|
||
|
|
assert result["materialized_params"] == {
|
||
|
|
"gwasSortedVcf": str(fmt["artifact"]),
|
||
|
|
"gwasTraitFile": str(trait["artifact"]),
|
||
|
|
"genotypeScopeId": GENOTYPE,
|
||
|
|
"phenotypeScopeId": PHENOTYPE,
|
||
|
|
"gwasPlink": PLINK,
|
||
|
|
"gwasEmmax": OFFICIAL_EMMAX,
|
||
|
|
"gwasRscript": RSCRIPT,
|
||
|
|
"gwasRequestedPcs": 3,
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize("requested,valid", [(0, True), (-1, False), (True, False)])
|
||
|
|
def test_emmax_preflight_requested_pcs_matches_cli_nonnegative_contract(
|
||
|
|
monkeypatch, tmp_path, requested, valid,
|
||
|
|
):
|
||
|
|
_producer_manifests(tmp_path / "workspace" / "job")
|
||
|
|
monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace")
|
||
|
|
result = api._api_semantic_preflight(
|
||
|
|
"job", "gwas_emmax", "cohort", {**_scopes(), "gwasRequestedPcs": requested},
|
||
|
|
)
|
||
|
|
assert result["valid"] is valid
|
||
|
|
if valid:
|
||
|
|
assert result["materialized_params"]["gwasRequestedPcs"] == requested
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize("case", [
|
||
|
|
"missing_scope", "wrong_scope", "failed", "bad_hash", "bad_size",
|
||
|
|
"missing_manifest", "caller_vcf", "caller_trait", "bad_version",
|
||
|
|
"missing_run_id", "bad_trait_analysis",
|
||
|
|
])
|
||
|
|
def test_emmax_semantic_preflight_fails_closed(monkeypatch, tmp_path, case):
|
||
|
|
workspace = tmp_path / "workspace" / "job"
|
||
|
|
fmt, trait = _producer_manifests(workspace)
|
||
|
|
params = _scopes()
|
||
|
|
target = fmt if case in {"failed", "bad_hash", "bad_size", "bad_version", "missing_run_id"} else trait
|
||
|
|
manifest = target["manifest"]
|
||
|
|
if case == "missing_scope": params["phenotypeScopeId"] = ""
|
||
|
|
elif case == "wrong_scope": params["phenotypeScopeId"] = "phenotype:" + "9" * 64
|
||
|
|
elif case == "failed": manifest["status"] = "failed"
|
||
|
|
elif case == "bad_hash": next(iter(manifest["artifacts"].values()))["sha256"] = "0" * 64
|
||
|
|
elif case == "bad_size": next(iter(manifest["artifacts"].values()))["size"] += 1
|
||
|
|
elif case == "missing_manifest": trait["path"].unlink()
|
||
|
|
elif case == "caller_vcf": params["gwasSortedVcf"] = str(trait["artifact"])
|
||
|
|
elif case == "caller_trait": params["gwasTraitFile"] = str(fmt["artifact"])
|
||
|
|
elif case == "bad_version": manifest["manifest_version"] = "0.9"
|
||
|
|
elif case == "missing_run_id": manifest.pop("run_id")
|
||
|
|
elif case == "bad_trait_analysis": manifest["analysis_scope_id"] = ""
|
||
|
|
if case in {"failed", "bad_hash", "bad_size", "bad_version", "missing_run_id", "bad_trait_analysis"}:
|
||
|
|
target["path"].write_text(json.dumps(manifest), encoding="utf-8")
|
||
|
|
monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace")
|
||
|
|
|
||
|
|
result = api._api_semantic_preflight("job", "gwas_emmax", "cohort", params)
|
||
|
|
|
||
|
|
assert result["valid"] is False
|
||
|
|
assert result["issues"][0]["code"] == "invalid_gwas_emmax_upstream"
|
||
|
|
|
||
|
|
|
||
|
|
def test_emmax_command_is_only_canonical_runner_and_preserves_explicit_values(tmp_path):
|
||
|
|
command = backend._gwas_emmax_cli_command(
|
||
|
|
workspace=tmp_path, job_id="job", prefix="cohort",
|
||
|
|
sorted_vcf=tmp_path / "in.vcf.gz", trait=tmp_path / "trait.txt",
|
||
|
|
vcf_manifest=tmp_path / "fmt.json", trait_manifest=tmp_path / "trait.json",
|
||
|
|
genotype_scope_id=GENOTYPE, phenotype_scope_id=PHENOTYPE,
|
||
|
|
plink=PLINK, emmax=EMMAX, rscript=RSCRIPT, requested_pcs=3,
|
||
|
|
threads=6, log_file=tmp_path / "logs/emmax.log",
|
||
|
|
)
|
||
|
|
assert command[:3] == [backend.sys.executable, "-m", "apps.cli.run_gwas_emmax"]
|
||
|
|
expected = {
|
||
|
|
"--vcf": str(tmp_path / "in.vcf.gz"), "--trait": str(tmp_path / "trait.txt"),
|
||
|
|
"--vcf-manifest": str(tmp_path / "fmt.json"),
|
||
|
|
"--trait-manifest": str(tmp_path / "trait.json"),
|
||
|
|
"--genotype-scope-id": GENOTYPE, "--phenotype-scope-id": PHENOTYPE,
|
||
|
|
"--plink": PLINK, "--emmax": EMMAX, "--rscript": RSCRIPT,
|
||
|
|
"--requested-pcs": "3", "--threads": "6",
|
||
|
|
"--log-file": str(tmp_path / "logs/emmax.log"),
|
||
|
|
}
|
||
|
|
for flag, value in expected.items():
|
||
|
|
assert command.count(flag) == 1
|
||
|
|
assert command[command.index(flag) + 1] == value
|
||
|
|
assert not any(token.endswith(".sh") or token.endswith(".R") for token in command)
|
||
|
|
|
||
|
|
|
||
|
|
def test_execute_request_and_frontend_dag_forward_emmax_contract(tmp_path):
|
||
|
|
request = api.ExecuteRequest(
|
||
|
|
stage="gwas_emmax", gwasPlink=PLINK, gwasEmmax=EMMAX,
|
||
|
|
gwasRscript=RSCRIPT, gwasRequestedPcs=2, **_scopes(),
|
||
|
|
)
|
||
|
|
assert (request.gwasPlink, request.gwasEmmax, request.gwasRequestedPcs) == (PLINK, EMMAX, 2)
|
||
|
|
frontend = (Path(__file__).resolve().parents[2] / "apps/web/vue3/index.html").read_text(encoding="utf-8")
|
||
|
|
start = frontend.index("const applyGwasExecutionParams =")
|
||
|
|
execute = frontend.index("const executeSingleStageWithParams =", start)
|
||
|
|
end = frontend.index("// 执行单个阶段（内部函数）", execute)
|
||
|
|
stage_params = {**_scopes(), "gwasPlink": PLINK, "gwasEmmax": EMMAX,
|
||
|
|
"gwasRscript": RSCRIPT, "gwasRequestedPcs": 2}
|
||
|
|
script = tmp_path / "emmax-request.cjs"
|
||
|
|
script.write_text(
|
||
|
|
"let captured=null;const batchExecuteParams={value:{plan:{params:{}}}};"
|
||
|
|
"const batchExecuteStages={value:[{id:'gwas_emmax',params:" + json.dumps(stage_params) + "}]};"
|
||
|
|
"const confirmParams=new Proxy({cpu:6,comboPanels:[]},{get:(o,k)=>o[k]});"
|
||
|
|
"const sessionId={value:'s'};const jobInfo={job_id:'job'};const getStageId=s=>s.id;"
|
||
|
|
"const getStageAction=s=>s.action;const usesRnaseqSampleParallel=()=>false;"
|
||
|
|
"const stageCpuForExecution=()=>6;const confirmCpuForStage=()=>6;"
|
||
|
|
"const executionPrefix=()=> 'cohort';const applyGeneFamilyContextToRequest=()=>{};"
|
||
|
|
"const applyRnaseqExecutionParams=()=>{};const applyWgsExecutionParams=()=>{};const submitExecuteRequestWithAutoAnnotation=async b=>{captured=b;return {success:true,async:false};};"
|
||
|
|
"const createStructuredExecuteError=x=>new Error(JSON.stringify(x));const runInteractiveSubfamilyConfirm=async()=>({});"
|
||
|
|
"const syncIdentifyExecutionSpecies=()=>{};const addAgentMessage=()=>{};const getStageLabel=x=>x;"
|
||
|
|
"const activePipelineTaskIds={value:[]};const pollTaskUntilComplete=async()=>({});const console={log:()=>{}};"
|
||
|
|
+ frontend[start:execute] + frontend[execute:end]
|
||
|
|
+ "executeSingleStageWithParams('gwas_emmax').then(()=>process.stdout.write(JSON.stringify(captured))).catch(e=>{process.stderr.write(String(e));process.exit(1);});",
|
||
|
|
encoding="utf-8",
|
||
|
|
)
|
||
|
|
completed = subprocess.run(["node", str(script)], text=True, capture_output=True, check=False)
|
||
|
|
assert completed.returncode == 0, completed.stderr
|
||
|
|
body = json.loads(completed.stdout)
|
||
|
|
for key, value in stage_params.items():
|
||
|
|
assert body[key] == value
|
||
|
|
|
||
|
|
|
||
|
|
def _run_frontend_emmax_handoff(tmp_path: Path, producer_results: list[tuple[str, dict]]) -> dict:
|
||
|
|
frontend = (Path(__file__).resolve().parents[2] / "apps/web/vue3/index.html").read_text(encoding="utf-8")
|
||
|
|
handoff_start = frontend.index("const handoffGwasEmmaxProducerIdentity =")
|
||
|
|
apply_start = frontend.index("const applyGwasExecutionParams =", handoff_start)
|
||
|
|
execute_start = frontend.index("const executeSingleStageWithParams =", apply_start)
|
||
|
|
stages = [
|
||
|
|
{"id": "gwas_fmt", "params": {}},
|
||
|
|
{"id": "gwas_trait", "params": {}},
|
||
|
|
{"id": "gwas_emmax", "params": {}},
|
||
|
|
]
|
||
|
|
script = tmp_path / "emmax-handoff.cjs"
|
||
|
|
script.write_text(
|
||
|
|
"const batchExecuteStages={value:" + json.dumps(stages) + "};"
|
||
|
|
"const batchExecuteParams={value:{plan:{params:{}}}};"
|
||
|
|
"const stageExecutionStatus={value:{gwas_fmt:'running',gwas_trait:'pending',gwas_emmax:'pending'}};"
|
||
|
|
"const confirmParams={};const getStageId=s=>s.id;const executionPrefix=()=> 'cohort';"
|
||
|
|
+ frontend[handoff_start:apply_start]
|
||
|
|
+ frontend[apply_start:execute_start]
|
||
|
|
+ "const producerResults=" + json.dumps(producer_results) + ";"
|
||
|
|
+ "for(const [stage,result] of producerResults) handoffGwasEmmaxProducerIdentity(stage,result);"
|
||
|
|
+ "const body={};const emmax=batchExecuteStages.value.find(s=>s.id==='gwas_emmax');"
|
||
|
|
+ "applyGwasExecutionParams(body,'gwas_emmax',emmax.params);"
|
||
|
|
+ "process.stdout.write(JSON.stringify({body,params:emmax.params}));",
|
||
|
|
encoding="utf-8",
|
||
|
|
)
|
||
|
|
completed = subprocess.run(["node", str(script)], text=True, capture_output=True, check=False)
|
||
|
|
assert completed.returncode == 0, completed.stderr
|
||
|
|
return json.loads(completed.stdout)
|
||
|
|
|
||
|
|
|
||
|
|
def test_frontend_runtime_handoff_feeds_exact_producer_scopes_to_emmax_preflight(
|
||
|
|
monkeypatch, tmp_path,
|
||
|
|
):
|
||
|
|
_producer_manifests(tmp_path / "workspace/job")
|
||
|
|
monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace")
|
||
|
|
producer_results = [
|
||
|
|
["gwas_fmt", {"status": "success", "result": {
|
||
|
|
"stage_identity": {"stage": "gwas_fmt", "prefix": "cohort",
|
||
|
|
"genotype_scope_id": GENOTYPE},
|
||
|
|
"stage_status": {"stage": "gwas_fmt", "completed": True,
|
||
|
|
"prefix": "cohort", "genotype_scope_id": GENOTYPE},
|
||
|
|
}}],
|
||
|
|
["gwas_trait", {"status": "success", "result": {
|
||
|
|
"stage_identity": {"stage": "gwas_trait", "prefix": "cohort",
|
||
|
|
"phenotype_scope_id": PHENOTYPE},
|
||
|
|
"stage_status": {"stage": "gwas_trait", "completed": True,
|
||
|
|
"prefix": "cohort", "phenotype_scope_id": PHENOTYPE},
|
||
|
|
}}],
|
||
|
|
]
|
||
|
|
captured = _run_frontend_emmax_handoff(tmp_path, producer_results)
|
||
|
|
assert captured["params"] == {
|
||
|
|
"genotypeScopeId": GENOTYPE, "phenotypeScopeId": PHENOTYPE,
|
||
|
|
}
|
||
|
|
preflight = api._api_semantic_preflight(
|
||
|
|
"job", "gwas_emmax", "cohort", captured["body"],
|
||
|
|
)
|
||
|
|
assert preflight["valid"] is True
|
||
|
|
assert preflight["materialized_params"]["genotypeScopeId"] == GENOTYPE
|
||
|
|
assert preflight["materialized_params"]["phenotypeScopeId"] == PHENOTYPE
|
||
|
|
frontend = (Path(__file__).resolve().parents[2] / "apps/web/vue3/index.html").read_text(encoding="utf-8")
|
||
|
|
verified = frontend.index("if (!manifestCompleted)")
|
||
|
|
handoff_call = frontend.index("handoffGwasEmmaxProducerIdentity(stageAction, stageResult);", verified)
|
||
|
|
completed_mark = frontend.index("stageExecutionStatus.value[stageId] = 'completed';", handoff_call)
|
||
|
|
assert verified < handoff_call < completed_mark
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize("bad_result", [
|
||
|
|
{"status": "success", "result": {"stage_identity": {
|
||
|
|
"stage": "gwas_trait", "prefix": "cohort", "genotype_scope_id": GENOTYPE,
|
||
|
|
}, "stage_status": {"stage": "gwas_trait", "prefix": "cohort", "completed": True}}},
|
||
|
|
{"status": "success", "result": {"stage_identity": {
|
||
|
|
"stage": "gwas_fmt", "prefix": "cohort", "genotype_scope_id": "genotype:bad",
|
||
|
|
}, "stage_status": {"stage": "gwas_fmt", "prefix": "cohort", "completed": True}}},
|
||
|
|
{"status": "success", "result": {"stage_status": {
|
||
|
|
"stage": "gwas_fmt", "prefix": "cohort", "completed": False,
|
||
|
|
"genotype_scope_id": GENOTYPE,
|
||
|
|
}}},
|
||
|
|
{"status": "success", "result": {
|
||
|
|
"stage_identity": {"stage": "gwas_fmt", "prefix": "other",
|
||
|
|
"genotype_scope_id": GENOTYPE},
|
||
|
|
"stage_status": {"stage": "gwas_fmt", "prefix": "other", "completed": True},
|
||
|
|
}},
|
||
|
|
])
|
||
|
|
def test_frontend_handoff_rejects_bad_identity_but_backend_uses_authoritative_manifests(
|
||
|
|
monkeypatch, tmp_path, bad_result,
|
||
|
|
):
|
||
|
|
_producer_manifests(tmp_path / "workspace/job")
|
||
|
|
monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace")
|
||
|
|
captured = _run_frontend_emmax_handoff(tmp_path, [["gwas_fmt", bad_result]])
|
||
|
|
assert "genotypeScopeId" not in captured["params"]
|
||
|
|
result = api._api_semantic_preflight("job", "gwas_emmax", "cohort", captured["body"])
|
||
|
|
assert result["valid"] is True
|
||
|
|
assert result["materialized_params"]["genotypeScopeId"] == GENOTYPE
|
||
|
|
assert result["materialized_params"]["phenotypeScopeId"] == PHENOTYPE
|
||
|
|
|
||
|
|
|
||
|
|
def test_emmax_declarations_and_exact_four_scope_status(tmp_path):
|
||
|
|
from gfa_core.mas.tools.stage_tools import StageToolRegistry
|
||
|
|
|
||
|
|
capability = get_stage_capability("gwas_emmax")
|
||
|
|
assert [item.role for item in capability.consumes] == ["sorted_vcf", "trait_table"]
|
||
|
|
assert [item.role for item in capability.provides] == [
|
||
|
|
"emmax_full", "emmax_pvalues", "manifest", "report", "environment",
|
||
|
|
]
|
||
|
|
tool = StageToolRegistry().get("gwas_emmax")
|
||
|
|
assert tool.outputs == ["emmax_full", "emmax_pvalues", "manifest", "report", "environment"]
|
||
|
|
workspace = tmp_path / "workspace" / "job"
|
||
|
|
_emmax_manifest(workspace)
|
||
|
|
identity = StageIdentity(
|
||
|
|
stage="gwas_emmax", prefix="cohort", genotype_scope_id=GENOTYPE,
|
||
|
|
phenotype_scope_id=PHENOTYPE, model_scope_id=MODEL, analysis_scope_id=ANALYSIS,
|
||
|
|
)
|
||
|
|
assert get_stage_status_for_identity(
|
||
|
|
"job", identity, workspace_root=tmp_path / "workspace", strict_identity=True,
|
||
|
|
).completed is True
|
||
|
|
identity.model_scope_id = ""
|
||
|
|
partial = get_stage_status_for_identity(
|
||
|
|
"job", identity, workspace_root=tmp_path / "workspace", strict_identity=True,
|
||
|
|
)
|
||
|
|
assert partial.completed is False
|
||
|
|
assert "all four" in partial.errors[0]
|
||
|
|
|
||
|
|
|
||
|
|
def test_emmax_results_require_exact_scopes_and_elapsed(monkeypatch, tmp_path):
|
||
|
|
workspace = tmp_path / "workspace" / "job"
|
||
|
|
_emmax_manifest(workspace)
|
||
|
|
monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace")
|
||
|
|
exact = asyncio.run(api.get_results(
|
||
|
|
job_id="job", stage="gwas_emmax", genotype_scope_id=GENOTYPE,
|
||
|
|
phenotype_scope_id=PHENOTYPE, model_scope_id=MODEL, analysis_scope_id=ANALYSIS,
|
||
|
|
))
|
||
|
|
assert exact["gwas_emmax"]["images"] == []
|
||
|
|
assert {item["role"] for item in exact["gwas_emmax"]["files"]} == {
|
||
|
|
"emmax_full", "emmax_pvalues", "report", "environment", "manifest",
|
||
|
|
}
|
||
|
|
assert exact["gwas_emmax"]["elapsed_time"] == 12.3
|
||
|
|
wrong = asyncio.run(api.get_results(
|
||
|
|
job_id="job", stage="gwas_emmax", genotype_scope_id=GENOTYPE,
|
||
|
|
phenotype_scope_id=PHENOTYPE, model_scope_id="gwas-model:" + "9" * 64,
|
||
|
|
analysis_scope_id=ANALYSIS,
|
||
|
|
))
|
||
|
|
assert wrong["gwas_emmax"]["files"] == []
|
||
|
|
|
||
|
|
|
||
|
|
def test_backend_emmax_accepts_only_fresh_exact_success_manifest(monkeypatch, tmp_path):
|
||
|
|
workspace = tmp_path / "workspace"
|
||
|
|
fmt, trait = _producer_manifests(workspace)
|
||
|
|
decisions = {
|
||
|
|
"sorted_vcf": {"path": str(fmt["artifact"]), "source_manifest": str(fmt["path"])},
|
||
|
|
"trait_table": {"path": str(trait["artifact"]), "source_manifest": str(trait["path"])},
|
||
|
|
}
|
||
|
|
monkeypatch.setattr(backend, "_resolve_gwas_emmax_inputs", lambda **_kwargs: decisions)
|
||
|
|
|
||
|
|
def fake_run(command, cwd=None):
|
||
|
|
assert command[command.index("--log-file") + 1].endswith("_gwas_emmax.log")
|
||
|
|
_emmax_manifest(workspace)
|
||
|
|
return 0
|
||
|
|
|
||
|
|
monkeypatch.setattr(backend, "_run_child_owned_log", fake_run)
|
||
|
|
message, manifest = backend.run_stage(
|
||
|
|
stage="gwas_emmax",
|
||
|
|
job=backend.JobInfo("job", str(tmp_path / "input"), str(workspace), str(workspace / "logs"), "now"),
|
||
|
|
prefix="cohort", cpu=6, genotype_scope_id=GENOTYPE,
|
||
|
|
phenotype_scope_id=PHENOTYPE, gwas_plink=PLINK, gwas_emmax=EMMAX,
|
||
|
|
gwas_rscript=RSCRIPT, gwas_requested_pcs=3,
|
||
|
|
)
|
||
|
|
assert message.startswith("✅")
|
||
|
|
assert manifest == workspace / "final/cohort.gwas_emmax.identity.manifest.json"
|
||
|
|
|
||
|
|
|
||
|
|
def test_backend_emmax_failure_never_replays_prior_success(monkeypatch, tmp_path):
|
||
|
|
workspace = tmp_path / "workspace"
|
||
|
|
fmt, trait = _producer_manifests(workspace)
|
||
|
|
prior = _emmax_manifest(workspace)
|
||
|
|
decisions = {
|
||
|
|
"sorted_vcf": {"path": str(fmt["artifact"]), "source_manifest": str(fmt["path"])},
|
||
|
|
"trait_table": {"path": str(trait["artifact"]), "source_manifest": str(trait["path"])},
|
||
|
|
}
|
||
|
|
monkeypatch.setattr(backend, "_resolve_gwas_emmax_inputs", lambda **_kwargs: decisions)
|
||
|
|
monkeypatch.setattr(backend, "_run_child_owned_log", lambda *_args, **_kwargs: 7)
|
||
|
|
message, manifest = backend.run_stage(
|
||
|
|
stage="gwas_emmax",
|
||
|
|
job=backend.JobInfo("job", str(tmp_path / "input"), str(workspace), str(workspace / "logs"), "now"),
|
||
|
|
prefix="cohort", cpu=6, genotype_scope_id=GENOTYPE,
|
||
|
|
phenotype_scope_id=PHENOTYPE,
|
||
|
|
)
|
||
|
|
assert "exit code: 7" in message
|
||
|
|
assert manifest is None
|
||
|
|
assert prior.is_file()
|
||
|
|
|
||
|
|
|
||
|
|
def _configure_task_test(monkeypatch, tmp_path, fake_run_stage):
|
||
|
|
import gfa_core.mas.resource_governor as resource_governor
|
||
|
|
|
||
|
|
class NoopLease:
|
||
|
|
def __init__(self, **_kwargs): pass
|
||
|
|
def wait(self, update_state=None): return None
|
||
|
|
def release(self, **_kwargs): return None
|
||
|
|
|
||
|
|
monkeypatch.setattr(resource_governor, "ResourceLease", NoopLease)
|
||
|
|
monkeypatch.setattr(resource_governor, "get_task", lambda _task_id: {})
|
||
|
|
monkeypatch.setattr(resource_governor, "snapshot", lambda **_kwargs: {"usage": {}})
|
||
|
|
monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace")
|
||
|
|
monkeypatch.setattr(api, "INPUT_ROOT", tmp_path / "input")
|
||
|
|
monkeypatch.setattr(api, "_attach_single_stage_status", lambda _task: None)
|
||
|
|
monkeypatch.setattr(api, "_attach_annotation_progress", lambda _task: None)
|
||
|
|
monkeypatch.setattr(api.time, "sleep", lambda _seconds: None)
|
||
|
|
monkeypatch.setattr(backend, "run_stage", fake_run_stage)
|
||
|
|
|
||
|
|
|
||
|
|
def test_api_task_forwards_emmax_fields_and_accepts_complete_postflight(monkeypatch, tmp_path):
|
||
|
|
captured = {}
|
||
|
|
|
||
|
|
def fake_run_stage(**kwargs):
|
||
|
|
captured.update(kwargs)
|
||
|
|
return "✅ done", _emmax_manifest(tmp_path / "workspace/job")
|
||
|
|
|
||
|
|
_configure_task_test(monkeypatch, tmp_path, fake_run_stage)
|
||
|
|
api.tasks["emmax-task"] = {
|
||
|
|
"id": "emmax-task", "stage": "gwas_emmax", "job_id": "job",
|
||
|
|
"status": "pending", "message": "", "created_at": "2026-07-21T00:00:00",
|
||
|
|
}
|
||
|
|
params = {"prefix": "cohort", **_scopes(), "gwasPlink": PLINK,
|
||
|
|
"gwasEmmax": EMMAX, "gwasRscript": RSCRIPT, "gwasRequestedPcs": 2}
|
||
|
|
api.run_stage_sync("emmax-task", "gwas_emmax", "job", params)
|
||
|
|
assert {key: captured[key] for key in (
|
||
|
|
"gwas_plink", "gwas_emmax", "gwas_rscript", "gwas_requested_pcs",
|
||
|
|
)} == {
|
||
|
|
"gwas_plink": PLINK, "gwas_emmax": EMMAX,
|
||
|
|
"gwas_rscript": RSCRIPT, "gwas_requested_pcs": 2,
|
||
|
|
}
|
||
|
|
task = api.tasks["emmax-task"]
|
||
|
|
assert task["status"] == "success", task
|
||
|
|
files = task["result"]["stage_files"]["gwas_emmax"]["files"]
|
||
|
|
assert {item["role"] for item in files} == {
|
||
|
|
"emmax_full", "emmax_pvalues", "report", "environment", "manifest",
|
||
|
|
}
|
||
|
|
assert task["result"]["stage_files"]["gwas_emmax"]["elapsed_time"] == 12.3
|
||
|
|
|
||
|
|
|
||
|
|
def test_api_task_rejects_partial_emmax_manifest_without_publishing_result(monkeypatch, tmp_path):
|
||
|
|
def fake_run_stage(**_kwargs):
|
||
|
|
path = _emmax_manifest(
|
||
|
|
tmp_path / "workspace/job",
|
||
|
|
roles=("emmax_full", "report", "environment"),
|
||
|
|
)
|
||
|
|
return "✅ done", path
|
||
|
|
|
||
|
|
_configure_task_test(monkeypatch, tmp_path, fake_run_stage)
|
||
|
|
api.tasks["partial-emmax"] = {
|
||
|
|
"id": "partial-emmax", "stage": "gwas_emmax", "job_id": "job",
|
||
|
|
"status": "pending", "message": "", "created_at": "2026-07-21T00:00:00",
|
||
|
|
}
|
||
|
|
api.run_stage_sync("partial-emmax", "gwas_emmax", "job", {"prefix": "cohort", **_scopes()})
|
||
|
|
task = api.tasks["partial-emmax"]
|
||
|
|
assert task["status"] == "failed"
|
||
|
|
assert task["result"]["success"] is False
|
||
|
|
assert "files" not in task["result"] and "images" not in task["result"]
|