493 lines
23 KiB
Python
493 lines
23 KiB
Python
|
import asyncio
|
||
|
|
import hashlib
|
||
|
|
import json
|
||
|
|
import subprocess
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
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_rscript
|
||
|
|
|
||
|
|
|
||
|
|
GENOTYPE = "genotype:" + "a" * 64
|
||
|
|
PHENOTYPE = "phenotype:" + "b" * 64
|
||
|
|
SOURCE_MODEL = "gwas-model:" + "c" * 64
|
||
|
|
ANALYSIS = "gwas-analysis:" + "d" * 64
|
||
|
|
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 _emmax_producer(workspace: Path, prefix: str = "cohort") -> tuple[Path, Path]:
|
||
|
|
final = workspace / "final"
|
||
|
|
final.mkdir(parents=True, exist_ok=True)
|
||
|
|
pvalues = final / f"{prefix}.gwas_emmax.identity.emmax_pvalues.tsv"
|
||
|
|
pvalues.write_text("chr\tpos\tp\n1\t10\t0.01\n", encoding="utf-8")
|
||
|
|
manifest = {
|
||
|
|
"manifest_version": "1.0", "run_id": "emmax-run", "stage": "gwas_emmax",
|
||
|
|
"prefix": prefix, "status": "success", "genotype_scope_id": GENOTYPE,
|
||
|
|
"phenotype_scope_id": PHENOTYPE, "model_scope_id": SOURCE_MODEL,
|
||
|
|
"analysis_scope_id": "gwas-analysis:" + "e" * 64,
|
||
|
|
"metrics": {"emmax_tested_count": 87466},
|
||
|
|
"artifacts": {"emmax_pvalues": _artifact(workspace, pvalues, "emmax_pvalues")},
|
||
|
|
}
|
||
|
|
path = final / f"{prefix}.gwas_emmax.identity.manifest.json"
|
||
|
|
path.write_text(json.dumps(manifest), encoding="utf-8")
|
||
|
|
return path, pvalues
|
||
|
|
|
||
|
|
|
||
|
|
def _emplot_manifest(
|
||
|
|
workspace: Path, prefix: str = "cohort", roles=None, *,
|
||
|
|
source_model: str = SOURCE_MODEL, analysis: str = ANALYSIS, tag: str = "identity",
|
||
|
|
) -> Path:
|
||
|
|
final = workspace / "final"
|
||
|
|
final.mkdir(parents=True, exist_ok=True)
|
||
|
|
roles = roles or ("emmax_manhattan", "emmax_qq", "report", "environment")
|
||
|
|
artifacts = {}
|
||
|
|
for role in roles:
|
||
|
|
suffix = ".png" if role.startswith("emmax_") else ".md" if role == "report" else ".txt"
|
||
|
|
path = final / f"{prefix}.gwas_emplot.{tag}.{role}{suffix}"
|
||
|
|
path.write_bytes(b"png" if suffix == ".png" else f"{role}\n".encode())
|
||
|
|
artifacts[role] = _artifact(workspace, path, role)
|
||
|
|
manifest = {
|
||
|
|
"manifest_version": "1.0", "stage": "gwas_emplot", "prefix": prefix,
|
||
|
|
"status": "success", "genotype_scope_id": GENOTYPE,
|
||
|
|
"phenotype_scope_id": PHENOTYPE, "source_model_scope_id": source_model,
|
||
|
|
"analysis_scope_id": analysis, "elapsed_sec": 3.26, "artifacts": artifacts,
|
||
|
|
}
|
||
|
|
path = final / f"{prefix}.gwas_emplot.{tag}.manifest.json"
|
||
|
|
path.write_text(json.dumps(manifest), encoding="utf-8")
|
||
|
|
return path
|
||
|
|
|
||
|
|
|
||
|
|
def _scopes() -> dict:
|
||
|
|
return {
|
||
|
|
"genotypeScopeId": GENOTYPE, "phenotypeScopeId": PHENOTYPE,
|
||
|
|
"sourceModelScopeId": SOURCE_MODEL,
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def test_emplot_preflight_materializes_exact_emmax_manifest_and_count(monkeypatch, tmp_path):
|
||
|
|
workspace = tmp_path / "workspace/job"
|
||
|
|
manifest, pvalues = _emmax_producer(workspace)
|
||
|
|
monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace")
|
||
|
|
|
||
|
|
result = api._api_semantic_preflight("job", "gwas_emplot", "cohort", _scopes())
|
||
|
|
|
||
|
|
assert result["valid"] is True
|
||
|
|
assert result["materialized_params"] == {
|
||
|
|
"gwasEmmaxPvalues": str(pvalues), "gwasEmmaxManifest": str(manifest),
|
||
|
|
"gwasEmmaxTestedHypotheses": 87466, "gwasRscript": RSCRIPT,
|
||
|
|
**_scopes(),
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def test_emplot_preflight_materializes_scopes_from_unique_emmax_manifest(
|
||
|
|
monkeypatch, tmp_path
|
||
|
|
):
|
||
|
|
workspace = tmp_path / "workspace/job"
|
||
|
|
manifest, pvalues = _emmax_producer(workspace)
|
||
|
|
monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace")
|
||
|
|
|
||
|
|
result = api._api_semantic_preflight("job", "gwas_emplot", "cohort", {})
|
||
|
|
|
||
|
|
assert result["valid"] is True
|
||
|
|
assert result["materialized_params"] == {
|
||
|
|
"gwasEmmaxPvalues": str(pvalues),
|
||
|
|
"gwasEmmaxManifest": str(manifest),
|
||
|
|
"gwasEmmaxTestedHypotheses": 87466,
|
||
|
|
"gwasRscript": RSCRIPT,
|
||
|
|
**_scopes(),
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def test_emplot_preflight_rejects_wrong_scope_and_caller_override(monkeypatch, tmp_path):
|
||
|
|
_emmax_producer(tmp_path / "workspace/job")
|
||
|
|
monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace")
|
||
|
|
for params in (
|
||
|
|
{**_scopes(), "sourceModelScopeId": "gwas-model:" + "9" * 64},
|
||
|
|
{**_scopes(), "gwasEmmaxPvalues": str(tmp_path / "other.tsv")},
|
||
|
|
{**_scopes(), "gwasEmmaxTestedHypotheses": 1},
|
||
|
|
):
|
||
|
|
result = api._api_semantic_preflight("job", "gwas_emplot", "cohort", params)
|
||
|
|
assert result["valid"] is False
|
||
|
|
assert result["issues"][0]["code"] == "invalid_gwas_emplot_upstream"
|
||
|
|
|
||
|
|
|
||
|
|
def test_emplot_command_is_only_canonical_runner(tmp_path):
|
||
|
|
command = backend._gwas_emplot_cli_command(
|
||
|
|
workspace=tmp_path, job_id="job", prefix="cohort",
|
||
|
|
emmax_pvalues=tmp_path / "p.tsv", emmax_manifest=tmp_path / "producer.json",
|
||
|
|
genotype_scope_id=GENOTYPE, phenotype_scope_id=PHENOTYPE,
|
||
|
|
source_model_scope_id=SOURCE_MODEL, emmax_tested_hypotheses=87466,
|
||
|
|
rscript=RSCRIPT, log_file=tmp_path / "logs/emplot.log",
|
||
|
|
)
|
||
|
|
assert command[:3] == [backend.sys.executable, "-m", "apps.cli.run_gwas_emmax_plot"]
|
||
|
|
for flag, value in {
|
||
|
|
"--emmax-pvalues": str(tmp_path / "p.tsv"),
|
||
|
|
"--emmax-manifest": str(tmp_path / "producer.json"),
|
||
|
|
"--genotype-scope-id": GENOTYPE, "--phenotype-scope-id": PHENOTYPE,
|
||
|
|
"--source-model-scope-id": SOURCE_MODEL,
|
||
|
|
"--emmax-tested-hypotheses": "87466", "--rscript": RSCRIPT,
|
||
|
|
}.items():
|
||
|
|
assert command.count(flag) == 1 and 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_declarations_match_emplot_contract():
|
||
|
|
from gfa_core.mas.tools.stage_tools import StageToolRegistry
|
||
|
|
|
||
|
|
request = api.ExecuteRequest(
|
||
|
|
stage="gwas_emplot", gwasEmmaxPvalues="p", gwasEmmaxManifest="m",
|
||
|
|
gwasEmmaxTestedHypotheses=87466, gwasRscript=RSCRIPT, **_scopes(),
|
||
|
|
)
|
||
|
|
assert request.gwasEmmaxTestedHypotheses == 87466
|
||
|
|
capability = get_stage_capability("gwas_emplot")
|
||
|
|
assert [item.role for item in capability.consumes] == ["emmax_pvalues"]
|
||
|
|
assert [item.role for item in capability.provides] == [
|
||
|
|
"emmax_manhattan", "emmax_qq", "manifest", "report", "environment",
|
||
|
|
]
|
||
|
|
tool = StageToolRegistry().get("gwas_emplot")
|
||
|
|
assert tool.required_inputs == ["emmax_pvalues"]
|
||
|
|
assert tool.outputs == ["emmax_manhattan", "emmax_qq", "manifest", "report", "environment"]
|
||
|
|
|
||
|
|
|
||
|
|
def test_backend_emplot_accepts_only_fresh_exact_success_manifest(monkeypatch, tmp_path):
|
||
|
|
workspace = tmp_path / "workspace"
|
||
|
|
producer, pvalues = _emmax_producer(workspace)
|
||
|
|
decisions = {
|
||
|
|
"emmax_pvalues": {"path": str(pvalues), "source_manifest": str(producer)},
|
||
|
|
"emmax_tested_count": 87466, "source_manifest": str(producer),
|
||
|
|
}
|
||
|
|
monkeypatch.setattr(backend, "_resolve_gwas_emplot_inputs", lambda **_kwargs: decisions)
|
||
|
|
|
||
|
|
def fake_run(command, cwd=None):
|
||
|
|
assert command[command.index("--log-file") + 1].endswith("_gwas_emplot.log")
|
||
|
|
_emplot_manifest(workspace)
|
||
|
|
return 0
|
||
|
|
|
||
|
|
monkeypatch.setattr(backend, "_run_child_owned_log", fake_run)
|
||
|
|
message, manifest = backend.run_stage(
|
||
|
|
stage="gwas_emplot",
|
||
|
|
job=backend.JobInfo("job", str(tmp_path / "input"), str(workspace), str(workspace / "logs"), "now"),
|
||
|
|
prefix="cohort", genotype_scope_id=GENOTYPE, phenotype_scope_id=PHENOTYPE,
|
||
|
|
source_model_scope_id=SOURCE_MODEL, gwas_rscript=RSCRIPT,
|
||
|
|
)
|
||
|
|
assert message.startswith("✅")
|
||
|
|
assert manifest == workspace / "final/cohort.gwas_emplot.identity.manifest.json"
|
||
|
|
|
||
|
|
|
||
|
|
def test_backend_emplot_failure_never_replays_prior_success(monkeypatch, tmp_path):
|
||
|
|
workspace = tmp_path / "workspace"
|
||
|
|
producer, pvalues = _emmax_producer(workspace)
|
||
|
|
_emplot_manifest(workspace)
|
||
|
|
monkeypatch.setattr(backend, "_resolve_gwas_emplot_inputs", lambda **_kwargs: {
|
||
|
|
"emmax_pvalues": {"path": str(pvalues), "source_manifest": str(producer)},
|
||
|
|
"emmax_tested_count": 87466, "source_manifest": str(producer),
|
||
|
|
})
|
||
|
|
monkeypatch.setattr(backend, "_run_child_owned_log", lambda *_args, **_kwargs: 7)
|
||
|
|
message, manifest = backend.run_stage(
|
||
|
|
stage="gwas_emplot",
|
||
|
|
job=backend.JobInfo("job", str(tmp_path / "input"), str(workspace), str(workspace / "logs"), "now"),
|
||
|
|
prefix="cohort", genotype_scope_id=GENOTYPE, phenotype_scope_id=PHENOTYPE,
|
||
|
|
source_model_scope_id=SOURCE_MODEL, gwas_rscript=RSCRIPT,
|
||
|
|
)
|
||
|
|
assert "exit code: 7" in message and manifest is None
|
||
|
|
|
||
|
|
|
||
|
|
def test_emplot_status_and_results_require_exact_four_scopes(monkeypatch, tmp_path):
|
||
|
|
workspace = tmp_path / "workspace/job"
|
||
|
|
_emplot_manifest(workspace)
|
||
|
|
monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace")
|
||
|
|
identity = StageIdentity(
|
||
|
|
stage="gwas_emplot", prefix="cohort", genotype_scope_id=GENOTYPE,
|
||
|
|
phenotype_scope_id=PHENOTYPE, source_model_scope_id=SOURCE_MODEL,
|
||
|
|
analysis_scope_id=ANALYSIS,
|
||
|
|
)
|
||
|
|
assert get_stage_status_for_identity(
|
||
|
|
"job", identity, workspace_root=tmp_path / "workspace", strict_identity=True,
|
||
|
|
).completed is True
|
||
|
|
exact = asyncio.run(api.get_results(
|
||
|
|
job_id="job", stage="gwas_emplot", genotype_scope_id=GENOTYPE,
|
||
|
|
phenotype_scope_id=PHENOTYPE, source_model_scope_id=SOURCE_MODEL,
|
||
|
|
analysis_scope_id=ANALYSIS,
|
||
|
|
))
|
||
|
|
assert len(exact["gwas_emplot"]["images"]) == 2
|
||
|
|
assert {item["role"] for item in exact["gwas_emplot"]["files"]} == {
|
||
|
|
"report", "environment", "manifest",
|
||
|
|
}
|
||
|
|
assert exact["gwas_emplot"]["elapsed_time"] == 3.3
|
||
|
|
wrong = asyncio.run(api.get_results(
|
||
|
|
job_id="job", stage="gwas_emplot", genotype_scope_id=GENOTYPE,
|
||
|
|
phenotype_scope_id=PHENOTYPE, source_model_scope_id="gwas-model:" + "9" * 64,
|
||
|
|
analysis_scope_id=ANALYSIS,
|
||
|
|
))
|
||
|
|
assert wrong["gwas_emplot"]["images"] == [] and wrong["gwas_emplot"]["files"] == []
|
||
|
|
|
||
|
|
|
||
|
|
def test_emplot_same_prefix_siblings_require_exact_source_model_and_analysis(monkeypatch, tmp_path):
|
||
|
|
workspace = tmp_path / "workspace/job"
|
||
|
|
other_model = "gwas-model:" + "8" * 64
|
||
|
|
other_analysis = "gwas-analysis:" + "7" * 64
|
||
|
|
_emplot_manifest(workspace, tag="first")
|
||
|
|
_emplot_manifest(
|
||
|
|
workspace, source_model=other_model, analysis=other_analysis, tag="second",
|
||
|
|
)
|
||
|
|
monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace")
|
||
|
|
|
||
|
|
exact = asyncio.run(api.get_results(
|
||
|
|
job_id="job", stage="gwas_emplot", genotype_scope_id=GENOTYPE,
|
||
|
|
phenotype_scope_id=PHENOTYPE, source_model_scope_id=SOURCE_MODEL,
|
||
|
|
analysis_scope_id=ANALYSIS,
|
||
|
|
))
|
||
|
|
sibling = asyncio.run(api.get_results(
|
||
|
|
job_id="job", stage="gwas_emplot", genotype_scope_id=GENOTYPE,
|
||
|
|
phenotype_scope_id=PHENOTYPE, source_model_scope_id=other_model,
|
||
|
|
analysis_scope_id=other_analysis,
|
||
|
|
))
|
||
|
|
mixed = asyncio.run(api.get_results(
|
||
|
|
job_id="job", stage="gwas_emplot", genotype_scope_id=GENOTYPE,
|
||
|
|
phenotype_scope_id=PHENOTYPE, source_model_scope_id=SOURCE_MODEL,
|
||
|
|
analysis_scope_id=other_analysis,
|
||
|
|
))
|
||
|
|
assert len(exact["gwas_emplot"]["images"]) == 2
|
||
|
|
assert len(sibling["gwas_emplot"]["images"]) == 2
|
||
|
|
assert mixed["gwas_emplot"]["images"] == [] and mixed["gwas_emplot"]["files"] == []
|
||
|
|
|
||
|
|
|
||
|
|
def test_api_task_accepts_only_complete_emplot_postflight(monkeypatch, tmp_path):
|
||
|
|
import gfa_core.mas.resource_governor as resource_governor
|
||
|
|
|
||
|
|
workspace_root = tmp_path / "workspace"
|
||
|
|
captured = {}
|
||
|
|
|
||
|
|
class NoopLease:
|
||
|
|
def __init__(self, **_kwargs): pass
|
||
|
|
def wait(self, update_state=None): return None
|
||
|
|
def release(self, **_kwargs): return None
|
||
|
|
|
||
|
|
def fake_run_stage(**kwargs):
|
||
|
|
captured.update(kwargs)
|
||
|
|
return "✅ done", _emplot_manifest(workspace_root / "job")
|
||
|
|
|
||
|
|
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", workspace_root)
|
||
|
|
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)
|
||
|
|
api.tasks["emplot-task"] = {
|
||
|
|
"id": "emplot-task", "stage": "gwas_emplot", "job_id": "job",
|
||
|
|
"status": "pending", "message": "", "created_at": "2026-07-21T00:00:00",
|
||
|
|
}
|
||
|
|
params = {
|
||
|
|
"prefix": "cohort", "gwasEmmaxPvalues": "p", "gwasEmmaxManifest": "m",
|
||
|
|
"gwasEmmaxTestedHypotheses": 87466, "gwasRscript": RSCRIPT, **_scopes(),
|
||
|
|
}
|
||
|
|
api.run_stage_sync("emplot-task", "gwas_emplot", "job", params)
|
||
|
|
assert {key: captured[key] for key in (
|
||
|
|
"gwas_emmax_pvalues", "gwas_emmax_manifest",
|
||
|
|
"gwas_emmax_tested_hypotheses", "source_model_scope_id",
|
||
|
|
)} == {
|
||
|
|
"gwas_emmax_pvalues": "p", "gwas_emmax_manifest": "m",
|
||
|
|
"gwas_emmax_tested_hypotheses": 87466,
|
||
|
|
"source_model_scope_id": SOURCE_MODEL,
|
||
|
|
}
|
||
|
|
task = api.tasks["emplot-task"]
|
||
|
|
assert task["status"] == "success", task
|
||
|
|
assert len(task["result"]["stage_files"]["gwas_emplot"]["images"]) == 2
|
||
|
|
assert {item["role"] for item in task["result"]["stage_files"]["gwas_emplot"]["files"]} == {
|
||
|
|
"report", "environment", "manifest",
|
||
|
|
}
|
||
|
|
assert task["result"]["stage_files"]["gwas_emplot"]["elapsed_time"] == 3.3
|
||
|
|
|
||
|
|
|
||
|
|
def test_api_task_rejects_partial_emplot_postflight(monkeypatch, tmp_path):
|
||
|
|
import gfa_core.mas.resource_governor as resource_governor
|
||
|
|
|
||
|
|
workspace_root = tmp_path / "workspace"
|
||
|
|
|
||
|
|
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", workspace_root)
|
||
|
|
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", lambda **_kwargs: (
|
||
|
|
"✅ done", _emplot_manifest(
|
||
|
|
workspace_root / "job", roles=("emmax_manhattan", "report", "environment"),
|
||
|
|
),
|
||
|
|
),
|
||
|
|
)
|
||
|
|
api.tasks["partial-emplot"] = {
|
||
|
|
"id": "partial-emplot", "stage": "gwas_emplot", "job_id": "job",
|
||
|
|
"status": "pending", "message": "", "created_at": "2026-07-21T00:00:00",
|
||
|
|
}
|
||
|
|
api.run_stage_sync("partial-emplot", "gwas_emplot", "job", {"prefix": "cohort", **_scopes()})
|
||
|
|
task = api.tasks["partial-emplot"]
|
||
|
|
assert task["status"] == "failed"
|
||
|
|
assert task["result"]["error"] == "invalid_gwas_emplot_outputs"
|
||
|
|
assert "images" not in task["result"] and "files" not in task["result"]
|
||
|
|
|
||
|
|
|
||
|
|
def test_emplot_frontend_has_typed_pending_same_prefix_handoff_and_roles():
|
||
|
|
frontend = (Path(api.PROJECT_ROOT) / "apps/web/vue3/index.html").read_text(encoding="utf-8")
|
||
|
|
assert "handoffGwasEmmaxPlotProducerIdentity" in frontend
|
||
|
|
assert "stageExecutionStatus.value.gwas_emplot !== 'pending'" in frontend
|
||
|
|
assert "completedStage !== 'gwas_emmax'" in frontend
|
||
|
|
assert "sourceModelScopeId" in frontend
|
||
|
|
for role in ("emmax_manhattan", "emmax_qq", "manifest", "report", "environment"):
|
||
|
|
assert role in frontend
|
||
|
|
|
||
|
|
|
||
|
|
def test_emplot_frontend_handoff_materializes_only_completed_same_prefix_identity(tmp_path):
|
||
|
|
frontend = (Path(api.PROJECT_ROOT) / "apps/web/vue3/index.html").read_text(encoding="utf-8")
|
||
|
|
handoff_start = frontend.index("const handoffGwasEmmaxPlotProducerIdentity =")
|
||
|
|
apply_start = frontend.index("const applyGwasExecutionParams =", handoff_start)
|
||
|
|
stages = [{"id": "gwas_emmax", "params": {}}, {"id": "gwas_emplot", "params": {}}]
|
||
|
|
producer = {
|
||
|
|
"status": "success", "result": {
|
||
|
|
"stage_identity": {
|
||
|
|
"stage": "gwas_emmax", "prefix": "cohort",
|
||
|
|
"genotype_scope_id": GENOTYPE, "phenotype_scope_id": PHENOTYPE,
|
||
|
|
"model_scope_id": SOURCE_MODEL,
|
||
|
|
},
|
||
|
|
"stage_status": {"stage": "gwas_emmax", "prefix": "cohort", "completed": True},
|
||
|
|
},
|
||
|
|
}
|
||
|
|
script = tmp_path / "emplot-handoff.cjs"
|
||
|
|
script.write_text(
|
||
|
|
"const batchExecuteStages={value:" + json.dumps(stages) + "};"
|
||
|
|
"const stageExecutionStatus={value:{gwas_emmax:'running',gwas_emplot:'pending'}};"
|
||
|
|
"const getStageId=s=>s.id;const executionPrefix=()=> 'cohort';"
|
||
|
|
+ frontend[handoff_start:apply_start]
|
||
|
|
+ "const ok=handoffGwasEmmaxPlotProducerIdentity('gwas_emmax',"
|
||
|
|
+ json.dumps(producer) + ");"
|
||
|
|
+ "process.stdout.write(JSON.stringify({ok,params:batchExecuteStages.value[1].params}));",
|
||
|
|
encoding="utf-8",
|
||
|
|
)
|
||
|
|
completed = subprocess.run(["node", str(script)], text=True, capture_output=True, check=False)
|
||
|
|
assert completed.returncode == 0, completed.stderr
|
||
|
|
payload = json.loads(completed.stdout)
|
||
|
|
assert payload == {"ok": True, "params": _scopes()}
|
||
|
|
|
||
|
|
|
||
|
|
def _run_emplot_frontend_queries(
|
||
|
|
tmp_path: Path, stage_params: dict, stage_result: dict | None = None,
|
||
|
|
stage_filter: str = "gwas_emplot",
|
||
|
|
) -> dict:
|
||
|
|
frontend = (Path(api.PROJECT_ROOT) / "apps/web/vue3/index.html").read_text(encoding="utf-8")
|
||
|
|
helper_start = frontend.index("const gwasEmplotExactIdentity =")
|
||
|
|
check_start = frontend.index("const checkManifestExists =", helper_start)
|
||
|
|
check_end = frontend.index("const pollTaskUntilComplete =", check_start)
|
||
|
|
fetch_start = frontend.index("const fetchStageCompleted =", check_end)
|
||
|
|
fetch_end = frontend.index("const executeStageDirect =", fetch_start)
|
||
|
|
refresh_start = frontend.index("const refreshResults =", fetch_end)
|
||
|
|
refresh_end = frontend.index("// 打开图片预览", refresh_start)
|
||
|
|
script = tmp_path / "emplot-query.cjs"
|
||
|
|
script.write_text(
|
||
|
|
"const params={prefix:'cohort'};const jobInfo={job_id:'job'};"
|
||
|
|
"const console={log:()=>{},error:()=>{}};"
|
||
|
|
"const batchExecuteParams={value:{plan:{params:{}}}};"
|
||
|
|
"const batchExecuteStages={value:[{id:'gwas_emplot',params:" + json.dumps(stage_params) + "}]};"
|
||
|
|
"const activeGenotypeScopeId={value:''};const activeRnaseqScopeId={value:''};"
|
||
|
|
"const confirmParams={};const normalizedPrefix=x=>x;const executionPrefix=()=> 'cohort';"
|
||
|
|
"const memberSetForScope=()=>'';const getStageId=s=>s.id;"
|
||
|
|
"const resultStageAliases={};const stages=[{name:'gwas_emplot',images:[],files:[]}];"
|
||
|
|
"const activeGeneFamilyScopeId=()=>'';const refreshMcscanxHtml=async()=>{};"
|
||
|
|
"const urls=[];const fetch=async url=>{urls.push(url);return {json:async()=>"
|
||
|
|
"(url.startsWith('/api/stage-status')?{completed:true}:{gwas_emplot:{images:['a','b'],files:[],elapsed_time:3.3}})};};"
|
||
|
|
+ frontend[helper_start:check_start]
|
||
|
|
+ frontend[check_start:check_end]
|
||
|
|
+ frontend[fetch_start:fetch_end]
|
||
|
|
+ frontend[refresh_start:refresh_end]
|
||
|
|
+ ("rememberGwasEmplotIdentity('gwas_emplot'," + json.dumps(stage_result) + ");" if stage_result else "")
|
||
|
|
+ "Promise.all([checkManifestExists('gwas_emplot','cohort',batchExecuteStages.value[0].params),"
|
||
|
|
+ "fetchStageCompleted('gwas_emplot','cohort',batchExecuteStages.value[0].params)])"
|
||
|
|
+ ".then(async status=>{await refreshResults('job'," + json.dumps(stage_filter) + ");"
|
||
|
|
+ "process.stdout.write(JSON.stringify({status,urls,stage:stages[0]}));})"
|
||
|
|
+ ".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
|
||
|
|
return json.loads(completed.stdout)
|
||
|
|
|
||
|
|
|
||
|
|
def test_emplot_frontend_status_and_results_send_all_four_scopes(tmp_path):
|
||
|
|
payload = _run_emplot_frontend_queries(tmp_path, _scopes(), {
|
||
|
|
"result": {"stage_identity": {
|
||
|
|
"stage": "gwas_emplot", "prefix": "cohort",
|
||
|
|
"genotype_scope_id": GENOTYPE, "phenotype_scope_id": PHENOTYPE,
|
||
|
|
"source_model_scope_id": SOURCE_MODEL, "analysis_scope_id": ANALYSIS,
|
||
|
|
}},
|
||
|
|
})
|
||
|
|
assert payload["status"] == [True, True]
|
||
|
|
assert len(payload["stage"]["images"]) == 2
|
||
|
|
for url in payload["urls"]:
|
||
|
|
query = url.split("?", 1)[1]
|
||
|
|
for key, value in {
|
||
|
|
"genotype_scope_id": GENOTYPE, "phenotype_scope_id": PHENOTYPE,
|
||
|
|
"source_model_scope_id": SOURCE_MODEL, "analysis_scope_id": ANALYSIS,
|
||
|
|
}.items():
|
||
|
|
assert f"{key}={value.replace(':', '%3A')}" in query
|
||
|
|
|
||
|
|
|
||
|
|
def test_emplot_frontend_incomplete_identity_is_fail_closed_without_partial_status(tmp_path):
|
||
|
|
payload = _run_emplot_frontend_queries(tmp_path, _scopes())
|
||
|
|
assert payload["status"] == [False, False]
|
||
|
|
assert all("/api/stage-status" not in url for url in payload["urls"])
|
||
|
|
results_url = next(url for url in payload["urls"] if url.startswith("/api/results"))
|
||
|
|
assert not any(key in results_url for key in (
|
||
|
|
"genotype_scope_id", "phenotype_scope_id", "source_model_scope_id", "analysis_scope_id",
|
||
|
|
))
|
||
|
|
|
||
|
|
|
||
|
|
def test_emplot_unfiltered_refresh_overlays_exact_persisted_result(tmp_path):
|
||
|
|
stage_result = {"result": {"stage_identity": {
|
||
|
|
"stage": "gwas_emplot", "prefix": "cohort",
|
||
|
|
"genotype_scope_id": GENOTYPE, "phenotype_scope_id": PHENOTYPE,
|
||
|
|
"source_model_scope_id": SOURCE_MODEL, "analysis_scope_id": ANALYSIS,
|
||
|
|
}}}
|
||
|
|
payload = _run_emplot_frontend_queries(
|
||
|
|
tmp_path, _scopes(), stage_result, stage_filter="",
|
||
|
|
)
|
||
|
|
exact_urls = [
|
||
|
|
url for url in payload["urls"]
|
||
|
|
if url.startswith("/api/results?") and "stage=gwas_emplot" in url
|
||
|
|
]
|
||
|
|
assert len(exact_urls) == 1
|
||
|
|
assert len(payload["stage"]["images"]) == 2
|
||
|
|
assert payload["stage"]["elapsed_time"] == 3.3
|
||
|
|
|
||
|
|
|
||
|
|
def test_emplot_skill_declares_exact_runner_and_roles():
|
||
|
|
config = (Path(api.PROJECT_ROOT) / "skills/gwas-emmax-plot/config.yaml").read_text(encoding="utf-8")
|
||
|
|
skill = (Path(api.PROJECT_ROOT) / "skills/gwas-emmax-plot/SKILL.md").read_text(encoding="utf-8")
|
||
|
|
assert "apps.cli.run_gwas_emmax_plot" in config and "apps.cli.run_gwas_emmax_plot" in skill
|
||
|
|
for role in ("emmax_pvalues", "emmax_manhattan", "emmax_qq", "manifest", "report", "environment"):
|
||
|
|
assert role in config and role in skill
|