433 lines
22 KiB
Python
433 lines
22 KiB
Python
import asyncio
|
|
import hashlib
|
|
import json
|
|
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 import gwas_upstream_manifest as upstream
|
|
from gfa_core.gwas_upstream_manifest import GwasUpstreamError
|
|
from gfa_core.mas.stage_capabilities import get_stage_capability
|
|
from gfa_core.mas.tools.stage_identity import StageIdentity, get_stage_status_for_identity
|
|
|
|
|
|
GENOTYPE = "genotype:" + "a" * 64
|
|
PHENOTYPE = "phenotype:" + "b" * 64
|
|
COVARIATE = "covariate:" + "c" * 64
|
|
MODEL = "gwas-model:" + "d" * 64
|
|
ANALYSIS = "gwas-analysis:" + "e" * 64
|
|
TASSEL = "/opt/tassel/run_pipeline.pl"
|
|
|
|
|
|
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(workspace: Path, stage: str, role: str, scopes: dict, tag: str = "source") -> tuple[Path, Path]:
|
|
final = workspace / "final"
|
|
final.mkdir(parents=True, exist_ok=True)
|
|
suffix = ".vcf.gz" if role == "sorted_vcf" else ".txt"
|
|
artifact = final / f"cohort.{stage}.{tag}.{role}{suffix}"
|
|
artifact.write_bytes(b"authoritative\n")
|
|
manifest = {
|
|
"manifest_version": "1.0", "run_id": f"{stage}-run", "stage": stage,
|
|
"prefix": "cohort", "status": "success", **scopes,
|
|
"artifacts": {role: _artifact(workspace, artifact, role)},
|
|
}
|
|
if stage != "gwas_fmt":
|
|
manifest["analysis_scope_id"] = "gwas-analysis:" + "f" * 64
|
|
path = final / f"cohort.{stage}.{tag}.manifest.json"
|
|
path.write_text(json.dumps(manifest), encoding="utf-8")
|
|
return path, artifact
|
|
|
|
|
|
def _upstream(workspace: Path) -> dict:
|
|
fmt_m, vcf = _producer(workspace, "gwas_fmt", "sorted_vcf", {"genotype_scope_id": GENOTYPE})
|
|
trait_m, trait = _producer(workspace, "gwas_trait", "trait_table", {"phenotype_scope_id": PHENOTYPE})
|
|
cov_m, cov = _producer(workspace, "gwas_cov", "covariate_table", {
|
|
"genotype_scope_id": GENOTYPE, "covariate_scope_id": COVARIATE,
|
|
})
|
|
return {"vcf_manifest": fmt_m, "sorted_vcf": vcf, "trait_manifest": trait_m,
|
|
"trait_table": trait, "covariate_manifest": cov_m, "covariate_table": cov}
|
|
|
|
|
|
def _scopes() -> dict:
|
|
return {"genotypeScopeId": GENOTYPE, "phenotypeScopeId": PHENOTYPE,
|
|
"covariateScopeId": COVARIATE}
|
|
|
|
|
|
def _glm_manifest(workspace: Path, roles=None, *, model=MODEL, analysis=ANALYSIS, tag="identity") -> Path:
|
|
final = workspace / "final"
|
|
final.mkdir(parents=True, exist_ok=True)
|
|
roles = roles or (
|
|
"glm_full", "glm_structure_full", "glm_pvalues", "glm_structure_pvalues",
|
|
"log", "report", "environment",
|
|
)
|
|
artifacts = {}
|
|
for role in roles:
|
|
suffix = ".md" if role == "report" else ".txt" if role in {"log", "environment", "glm_full", "glm_structure_full"} else ".tsv"
|
|
path = final / f"cohort.gwas_glm.{tag}.{role}{suffix}"
|
|
path.write_text(role + "\n", encoding="utf-8")
|
|
artifacts[role] = _artifact(workspace, path, role)
|
|
manifest = {
|
|
"manifest_version": "1.0", "stage": "gwas_glm", "prefix": "cohort",
|
|
"status": "success", "genotype_scope_id": GENOTYPE,
|
|
"phenotype_scope_id": PHENOTYPE, "covariate_scope_id": COVARIATE,
|
|
"model_scope_id": model, "analysis_scope_id": analysis,
|
|
"elapsed_sec": 4.26, "artifacts": artifacts,
|
|
}
|
|
path = final / f"cohort.gwas_glm.{tag}.manifest.json"
|
|
path.write_text(json.dumps(manifest), encoding="utf-8")
|
|
return path
|
|
|
|
|
|
def test_glm_resolver_returns_three_exact_manifest_artifacts(tmp_path):
|
|
workspace = tmp_path / "workspace"
|
|
expected = _upstream(workspace)
|
|
resolved = upstream.resolve_gwas_glm_upstream(
|
|
workspace=workspace, prefix="cohort", genotype_scope_id=GENOTYPE,
|
|
phenotype_scope_id=PHENOTYPE, covariate_scope_id=COVARIATE,
|
|
)
|
|
assert {key: resolved[key]["path"] for key in ("sorted_vcf", "trait_table", "covariate_table")} == {
|
|
"sorted_vcf": str(expected["sorted_vcf"]), "trait_table": str(expected["trait_table"]),
|
|
"covariate_table": str(expected["covariate_table"]),
|
|
}
|
|
assert resolved["sorted_vcf"]["source_manifest"] == str(expected["vcf_manifest"])
|
|
|
|
|
|
@pytest.mark.parametrize("failure", ["scope", "hash", "status", "multiple"])
|
|
def test_glm_resolver_fails_closed_without_scan_latest_or_override(tmp_path, failure):
|
|
workspace = tmp_path / "workspace"
|
|
paths = _upstream(workspace)
|
|
if failure == "scope":
|
|
payload = json.loads(paths["covariate_manifest"].read_text())
|
|
payload["genotype_scope_id"] = "genotype:" + "9" * 64
|
|
paths["covariate_manifest"].write_text(json.dumps(payload))
|
|
elif failure == "hash":
|
|
paths["trait_table"].write_text("changed\n")
|
|
elif failure == "status":
|
|
payload = json.loads(paths["trait_manifest"].read_text()); payload["status"] = "failed"
|
|
paths["trait_manifest"].write_text(json.dumps(payload))
|
|
else:
|
|
_producer(workspace, "gwas_trait", "trait_table", {"phenotype_scope_id": PHENOTYPE}, tag="copy")
|
|
with pytest.raises(GwasUpstreamError):
|
|
upstream.resolve_gwas_glm_upstream(
|
|
workspace=workspace, prefix="cohort", genotype_scope_id=GENOTYPE,
|
|
phenotype_scope_id=PHENOTYPE, covariate_scope_id=COVARIATE,
|
|
)
|
|
|
|
|
|
def test_glm_preflight_materializes_exact_paths_and_manifests(monkeypatch, tmp_path):
|
|
workspace = tmp_path / "workspace/job"
|
|
paths = _upstream(workspace)
|
|
monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace")
|
|
result = api._api_semantic_preflight("job", "gwas_glm", "cohort", _scopes())
|
|
assert result["valid"] is True
|
|
assert result["materialized_params"] == {
|
|
"gwasSortedVcf": str(paths["sorted_vcf"]), "gwasVcfManifest": str(paths["vcf_manifest"]),
|
|
"gwasTraitFile": str(paths["trait_table"]), "gwasTraitManifest": str(paths["trait_manifest"]),
|
|
"gwasCovariateFile": str(paths["covariate_table"]), "gwasCovariateManifest": str(paths["covariate_manifest"]),
|
|
"tasselPipeline": "run_pipeline.pl", "tasselXmx": "10g", **_scopes(),
|
|
}
|
|
|
|
|
|
def test_glm_preflight_materializes_scopes_from_unique_producer_manifests(
|
|
monkeypatch, tmp_path
|
|
):
|
|
workspace = tmp_path / "workspace/job"
|
|
paths = _upstream(workspace)
|
|
monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace")
|
|
|
|
result = api._api_semantic_preflight("job", "gwas_glm", "cohort", {})
|
|
|
|
assert result["valid"] is True
|
|
assert result["materialized_params"] == {
|
|
"gwasSortedVcf": str(paths["sorted_vcf"]),
|
|
"gwasVcfManifest": str(paths["vcf_manifest"]),
|
|
"gwasTraitFile": str(paths["trait_table"]),
|
|
"gwasTraitManifest": str(paths["trait_manifest"]),
|
|
"gwasCovariateFile": str(paths["covariate_table"]),
|
|
"gwasCovariateManifest": str(paths["covariate_manifest"]),
|
|
"tasselPipeline": "run_pipeline.pl",
|
|
"tasselXmx": "10g",
|
|
**_scopes(),
|
|
}
|
|
|
|
|
|
def test_glm_command_is_only_canonical_runner(tmp_path):
|
|
command = backend._gwas_glm_cli_command(
|
|
workspace=tmp_path, job_id="job", prefix="cohort", sorted_vcf=tmp_path / "x.vcf.gz",
|
|
trait=tmp_path / "trait.txt", covariate=tmp_path / "cov.txt",
|
|
vcf_manifest=tmp_path / "fmt.json", trait_manifest=tmp_path / "trait.json",
|
|
covariate_manifest=tmp_path / "cov.json", genotype_scope_id=GENOTYPE,
|
|
phenotype_scope_id=PHENOTYPE, covariate_scope_id=COVARIATE,
|
|
tassel_pipeline=TASSEL, xmx="20g", log_file=tmp_path / "logs/gwas_glm.log",
|
|
)
|
|
assert command[:3] == [backend.sys.executable, "-m", "apps.cli.run_gwas_tassel_glm"]
|
|
for flag in ("--vcf", "--trait", "--covariate", "--vcf-manifest", "--trait-manifest",
|
|
"--covariate-manifest", "--genotype-scope-id", "--phenotype-scope-id",
|
|
"--covariate-scope-id", "--tassel-pipeline", "--xmx", "--log-file"):
|
|
assert command.count(flag) == 1
|
|
assert not any(token.endswith(".sh") for token in command)
|
|
|
|
|
|
def test_backend_glm_accepts_only_fresh_exact_success_manifest(monkeypatch, tmp_path):
|
|
workspace = tmp_path / "workspace"
|
|
decisions = {
|
|
role: {"path": str(path), "source_manifest": str(manifest)}
|
|
for role, manifest, path in (
|
|
("sorted_vcf", *_producer(workspace, "gwas_fmt", "sorted_vcf", {"genotype_scope_id": GENOTYPE})),
|
|
("trait_table", *_producer(workspace, "gwas_trait", "trait_table", {"phenotype_scope_id": PHENOTYPE})),
|
|
("covariate_table", *_producer(workspace, "gwas_cov", "covariate_table", {
|
|
"genotype_scope_id": GENOTYPE, "covariate_scope_id": COVARIATE,
|
|
})),
|
|
)
|
|
}
|
|
monkeypatch.setattr(backend, "_resolve_gwas_glm_inputs", lambda **_kwargs: decisions)
|
|
|
|
def fake_run(command, cwd=None):
|
|
assert command[:3] == [backend.sys.executable, "-m", "apps.cli.run_gwas_tassel_glm"]
|
|
assert command[command.index("--log-file") + 1].endswith("_gwas_glm.log")
|
|
_glm_manifest(workspace)
|
|
return 0
|
|
|
|
monkeypatch.setattr(backend, "_run_child_owned_log", fake_run)
|
|
message, manifest = backend.run_stage(
|
|
stage="gwas_glm",
|
|
job=backend.JobInfo("job", str(tmp_path / "input"), str(workspace), str(workspace / "logs"), "now"),
|
|
prefix="cohort", genotype_scope_id=GENOTYPE, phenotype_scope_id=PHENOTYPE,
|
|
covariate_scope_id=COVARIATE, tassel_pipeline=TASSEL, tassel_xmx="20g",
|
|
)
|
|
assert message.startswith("✅")
|
|
assert manifest == workspace / "final/cohort.gwas_glm.identity.manifest.json"
|
|
|
|
|
|
def test_api_task_rejects_partial_glm_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", _glm_manifest(workspace_root / "job", roles=(
|
|
"glm_full", "glm_structure_full", "glm_pvalues", "log", "report", "environment",
|
|
)),
|
|
))
|
|
api.tasks["partial-glm"] = {
|
|
"id": "partial-glm", "stage": "gwas_glm", "job_id": "job",
|
|
"status": "pending", "message": "", "created_at": "2026-07-21T00:00:00",
|
|
}
|
|
api.run_stage_sync("partial-glm", "gwas_glm", "job", {"prefix": "cohort", **_scopes()})
|
|
task = api.tasks["partial-glm"]
|
|
assert task["status"] == "failed"
|
|
assert task["result"]["error"] == "invalid_gwas_glm_outputs"
|
|
|
|
|
|
def test_execute_request_capability_tool_and_skill_are_closed():
|
|
from gfa_core.mas.tools.stage_tools import StageToolRegistry
|
|
request = api.ExecuteRequest(stage="gwas_glm", gwasVcfManifest="v", gwasTraitManifest="t",
|
|
gwasCovariateManifest="c", **_scopes())
|
|
assert request.gwasCovariateManifest == "c"
|
|
capability = get_stage_capability("gwas_glm")
|
|
assert [item.role for item in capability.consumes] == ["sorted_vcf", "trait_table", "covariate_table"]
|
|
expected = ["glm_full", "glm_structure_full", "glm_pvalues", "glm_structure_pvalues",
|
|
"manifest", "report", "environment"]
|
|
assert [item.role for item in capability.provides] == expected
|
|
tool = StageToolRegistry().get("gwas_glm")
|
|
assert tool.required_inputs == ["sorted_vcf", "trait_table", "covariate_table"]
|
|
assert tool.outputs == expected
|
|
config = (Path(api.PROJECT_ROOT) / "skills/gwas-tassel-glm/config.yaml").read_text()
|
|
skill = (Path(api.PROJECT_ROOT) / "skills/gwas-tassel-glm/SKILL.md").read_text()
|
|
for token in ("apps.cli.run_gwas_tassel_glm", *expected, "sorted_vcf", "trait_table", "covariate_table"):
|
|
assert token in config and token in skill
|
|
|
|
|
|
def test_glm_status_and_results_require_exact_five_scopes(monkeypatch, tmp_path):
|
|
workspace = tmp_path / "workspace/job"
|
|
_glm_manifest(workspace)
|
|
monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace")
|
|
identity = StageIdentity(stage="gwas_glm", prefix="cohort", genotype_scope_id=GENOTYPE,
|
|
phenotype_scope_id=PHENOTYPE, covariate_scope_id=COVARIATE,
|
|
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
|
|
exact = asyncio.run(api.get_results(job_id="job", stage="gwas_glm", genotype_scope_id=GENOTYPE,
|
|
phenotype_scope_id=PHENOTYPE, covariate_scope_id=COVARIATE,
|
|
model_scope_id=MODEL, analysis_scope_id=ANALYSIS))
|
|
assert exact["gwas_glm"]["images"] == []
|
|
assert {item["role"] for item in exact["gwas_glm"]["files"]} == {
|
|
"glm_full", "glm_structure_full", "glm_pvalues", "glm_structure_pvalues",
|
|
"log", "report", "environment", "manifest",
|
|
}
|
|
assert exact["gwas_glm"]["elapsed_time"] == 4.3
|
|
wrong = asyncio.run(api.get_results(job_id="job", stage="gwas_glm", genotype_scope_id=GENOTYPE,
|
|
phenotype_scope_id=PHENOTYPE, covariate_scope_id="covariate:" + "9" * 64,
|
|
model_scope_id=MODEL, analysis_scope_id=ANALYSIS))
|
|
assert wrong["gwas_glm"]["files"] == []
|
|
|
|
|
|
def test_glm_partial_identity_and_partial_outputs_fail_closed(monkeypatch, tmp_path):
|
|
workspace = tmp_path / "workspace/job"
|
|
_glm_manifest(workspace, roles=("glm_full", "glm_pvalues", "log", "report", "environment"))
|
|
partial = StageIdentity(stage="gwas_glm", prefix="cohort", genotype_scope_id=GENOTYPE)
|
|
report = get_stage_status_for_identity("job", partial, workspace_root=tmp_path / "workspace",
|
|
strict_identity=True)
|
|
assert report.completed is False and "all five" in report.errors[0]
|
|
|
|
|
|
def test_glm_frontend_declares_typed_handoff_five_scope_queries_and_no_dialog():
|
|
frontend = (Path(api.PROJECT_ROOT) / "apps/web/vue3/index.html").read_text(encoding="utf-8")
|
|
assert "handoffGwasGlmProducerIdentity" in frontend
|
|
assert "stageExecutionStatus.value.gwas_glm !== 'pending'" in frontend
|
|
for producer in ("gwas_fmt", "gwas_trait", "gwas_cov"):
|
|
assert producer in frontend
|
|
assert "gwasGlmExactIdentity" in frontend and "rememberGwasGlmIdentity" in frontend
|
|
assert "sourceModelScopeId" not in frontend[frontend.index("const gwasGlmExactIdentity ="):frontend.index("const checkManifestExists =", frontend.index("const gwasGlmExactIdentity ="))]
|
|
for role in ("glm_full", "glm_structure_full", "glm_pvalues", "glm_structure_pvalues", "manifest", "report", "environment"):
|
|
assert role in frontend
|
|
assert "showGwasGlm" not in frontend and "gwasGlmDialog" not in frontend
|
|
|
|
|
|
def test_glm_natural_language_is_explicit_and_bare_gwas_is_not_alias():
|
|
from gfa_core.mas.utils.intent_parser import parse_intent_from_query
|
|
explicit = parse_intent_from_query("使用 TASSEL 执行固定效应 GLM 分析")
|
|
bare = parse_intent_from_query("执行 GWAS 分析")
|
|
assert explicit["primary_stage"] == "gwas_glm"
|
|
assert bare["primary_stage"] == "gwas_glm"
|
|
assert len(bare["stages"]) > 1 and "gwas_glm" in bare["stages"]
|
|
|
|
|
|
def test_execute_glm_preserves_nondefault_tassel_heap_to_worker_submission(monkeypatch, tmp_path):
|
|
workspace = tmp_path / "workspace/job"
|
|
_upstream(workspace)
|
|
|
|
class DummyExecutor:
|
|
def __init__(self): self.calls = []
|
|
def submit(self, *args): self.calls.append(args)
|
|
|
|
dummy = DummyExecutor()
|
|
monkeypatch.setattr(api, "PROJECT_ROOT", tmp_path)
|
|
monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace")
|
|
monkeypatch.setattr(api, "INPUT_ROOT", tmp_path / "input")
|
|
monkeypatch.setattr(api, "CELERY_AVAILABLE", False)
|
|
monkeypatch.setattr(api, "executor", dummy)
|
|
result = asyncio.run(api.execute_stage(api.ExecuteRequest(
|
|
stage="gwas_glm", job_id="job", prefix="cohort", tasselXmx="20g", **_scopes(),
|
|
)))
|
|
assert result["async"] is True and len(dummy.calls) == 1
|
|
submitted_params = dummy.calls[0][4]
|
|
assert submitted_params["tasselXmx"] == "20g"
|
|
|
|
|
|
def test_background_completion_materializes_all_three_glm_scopes(tmp_path):
|
|
frontend = (Path(api.PROJECT_ROOT) / "apps/web/vue3/index.html").read_text(encoding="utf-8")
|
|
handoff_start = frontend.index("const handoffGwasGlmProducerIdentity =")
|
|
apply_start = frontend.index("const applyGwasExecutionParams =", handoff_start)
|
|
handler_start = frontend.index("const handleStageCompleted =")
|
|
handler_end = frontend.index("const resolveSubfamilyReview =", handler_start)
|
|
stages = [{"id": "gwas_glm", "params": {}}]
|
|
results = [
|
|
{"stage": "gwas_fmt", "scope": {"genotype_scope_id": GENOTYPE}},
|
|
{"stage": "gwas_trait", "scope": {"phenotype_scope_id": PHENOTYPE}},
|
|
{"stage": "gwas_cov", "scope": {"genotype_scope_id": GENOTYPE, "covariate_scope_id": COVARIATE}},
|
|
]
|
|
script = tmp_path / "background-glm-handoff.cjs"
|
|
script.write_text(
|
|
"const batchExecuteStages={value:" + json.dumps(stages) + "};"
|
|
"const stageExecutionStatus={value:{gwas_glm:'pending'}};"
|
|
"const getStageId=s=>s.id;const executionPrefix=()=> 'cohort';"
|
|
"const setResultStageFilter=()=>{};const refreshResults=async()=>{};"
|
|
"const openCompletedSubfamilyReview=async()=>{};"
|
|
"const methodologyIdentifyMemberReviewRequired=()=>false;"
|
|
"const subfamilyPerSpeciesMode=()=>false;const maybeShowIdentifyDomainReview=async()=>false;"
|
|
"const rememberGwasGlmIdentity=()=>false;"
|
|
+ frontend[handoff_start:apply_start]
|
|
+ frontend[handler_start:handler_end]
|
|
+ "const rows=" + json.dumps(results) + ";"
|
|
+ "Promise.all(rows.map(row=>handleStageCompleted(row.stage,{result:{stage_status:{completed:true,stage:row.stage,prefix:'cohort',...row.scope},stage_identity:{stage:row.stage,prefix:'cohort',...row.scope}}},{})))"
|
|
+ ".then(()=>process.stdout.write(JSON.stringify(batchExecuteStages.value[0].params)))"
|
|
+ ".catch(e=>{process.stderr.write(String(e));process.exit(1)});",
|
|
encoding="utf-8",
|
|
)
|
|
completed = __import__("subprocess").run(["node", str(script)], text=True, capture_output=True, check=False)
|
|
assert completed.returncode == 0, completed.stderr
|
|
assert json.loads(completed.stdout) == _scopes()
|
|
|
|
|
|
def test_frontend_limits_each_scheduler_wave_to_one_gwas_model_heavy(tmp_path):
|
|
frontend = (Path(api.PROJECT_ROOT) / "apps/web/vue3/index.html").read_text(encoding="utf-8")
|
|
assert "const limitGwasModelHeavyWave =" in frontend
|
|
assert "limitGwasModelHeavyWave(executableCandidates, stageExecutionStatus.value)" in frontend
|
|
assert "limitGwasModelHeavyWave(executableCandidates, localStatus)" in frontend
|
|
start = frontend.index("const gwasModelHeavyStages =")
|
|
end = frontend.index("const confirmParams =", start)
|
|
script = tmp_path / "gwas-heavy-wave.cjs"
|
|
script.write_text(
|
|
frontend[start:end]
|
|
+ "const candidates=['gwas_mlm','gwas_emmax','gwas_glm','gwas_plot'];"
|
|
+ "const pending=limitGwasModelHeavyWave(candidates,{});"
|
|
+ "const running=limitGwasModelHeavyWave(candidates,{gwas_emmax:'running'});"
|
|
+ "process.stdout.write(JSON.stringify({pending,running}));",
|
|
encoding="utf-8",
|
|
)
|
|
completed = __import__("subprocess").run(["node", str(script)], text=True, capture_output=True, check=False)
|
|
assert completed.returncode == 0, completed.stderr
|
|
assert json.loads(completed.stdout) == {
|
|
"pending": ["gwas_mlm", "gwas_plot"],
|
|
"running": ["gwas_plot"],
|
|
}
|
|
|
|
|
|
def test_gwas_model_heavy_lease_is_cross_job_exclusive_and_released(monkeypatch, tmp_path):
|
|
from gfa_core.mas import resource_governor
|
|
from gfa_core.mas.stage_catalog import get_stage_resource_profile
|
|
|
|
monkeypatch.setenv("BBAV5_RESOURCE_STATE_DIR", str(tmp_path / "governor"))
|
|
for stage in ("gwas_mlm", "gwas_emmax", "gwas_glm"):
|
|
assert get_stage_resource_profile(stage)["concurrency_group"] == "gwas_model_heavy"
|
|
first = resource_governor.try_acquire(
|
|
task_id="one", stage="gwas_mlm", job_id="job-a", prefix="cohort", cpu=1,
|
|
concurrency_group="gwas_model_heavy",
|
|
)
|
|
blocked = resource_governor.try_acquire(
|
|
task_id="two", stage="gwas_glm", job_id="job-b", prefix="cohort", cpu=1,
|
|
concurrency_group="gwas_model_heavy",
|
|
)
|
|
assert first["granted"] is True
|
|
assert blocked["granted"] is False and "gwas_model_heavy" in blocked["reason"]
|
|
resource_governor.release("one", status="failed")
|
|
acquired = resource_governor.try_acquire(
|
|
task_id="two", stage="gwas_glm", job_id="job-b", prefix="cohort", cpu=1,
|
|
concurrency_group="gwas_model_heavy",
|
|
)
|
|
assert acquired["granted"] is True
|
|
|
|
|
|
def test_both_threadpool_and_celery_pass_the_catalog_concurrency_group():
|
|
main_source = (Path(api.PROJECT_ROOT) / "apps/web/vue3/main.py").read_text(encoding="utf-8")
|
|
celery_source = (Path(api.PROJECT_ROOT) / "tasks.py").read_text(encoding="utf-8")
|
|
assert 'concurrency_group=str(profile.get("concurrency_group") or "")' in main_source
|
|
assert 'concurrency_group=str(resource_profile.get("concurrency_group") or "")' in celery_source
|