590 lines
27 KiB
Python
590 lines
27 KiB
Python
|
import hashlib
|
||
|
|
import json
|
||
|
|
import sys
|
||
|
|
import types
|
||
|
|
from dataclasses import dataclass
|
||
|
|
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
|
||
|
|
|
||
|
|
|
||
|
|
GENOTYPE = "genotype:" + "a" * 64
|
||
|
|
PHENOTYPE = "phenotype:" + "b" * 64
|
||
|
|
COVARIATE = "covariate:" + "c" * 64
|
||
|
|
MODEL = "gwas-model:" + "d" * 64
|
||
|
|
ANALYSIS = "gwas-analysis:" + "e" * 64
|
||
|
|
|
||
|
|
|
||
|
|
def _sha(path: Path) -> str:
|
||
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
||
|
|
|
||
|
|
|
||
|
|
def _upstream(workspace: Path, prefix: str, stage: str, role: str, scopes: dict) -> Path:
|
||
|
|
final = workspace / "final"
|
||
|
|
final.mkdir(parents=True, exist_ok=True)
|
||
|
|
artifact = final / f"{prefix}.{stage}.{role}.txt"
|
||
|
|
artifact.write_text(f"{stage}\n", encoding="utf-8")
|
||
|
|
manifest = {
|
||
|
|
"manifest_version": "1.0", "stage": stage, "prefix": prefix, "status": "success",
|
||
|
|
**scopes,
|
||
|
|
"artifacts": {role: {
|
||
|
|
"role": role, "path": f"final/{artifact.name}", "full_path": str(artifact.resolve()),
|
||
|
|
"size": artifact.stat().st_size, "sha256": _sha(artifact),
|
||
|
|
}},
|
||
|
|
}
|
||
|
|
path = final / f"{prefix}.{stage}.manifest.json"
|
||
|
|
path.write_text(json.dumps(manifest), encoding="utf-8")
|
||
|
|
return artifact
|
||
|
|
|
||
|
|
|
||
|
|
def _four_upstreams(workspace: Path, prefix: str = "cohort") -> dict[str, Path]:
|
||
|
|
return {
|
||
|
|
"sorted_vcf": _upstream(workspace, prefix, "gwas_fmt", "sorted_vcf", {"genotype_scope_id": GENOTYPE}),
|
||
|
|
"kinship_matrix": _upstream(workspace, prefix, "gwas_kinship", "kinship_matrix", {"genotype_scope_id": GENOTYPE}),
|
||
|
|
"trait_table": _upstream(workspace, prefix, "gwas_trait", "trait_table", {"phenotype_scope_id": PHENOTYPE}),
|
||
|
|
"covariate_table": _upstream(workspace, prefix, "gwas_cov", "covariate_table", {
|
||
|
|
"genotype_scope_id": GENOTYPE, "covariate_scope_id": COVARIATE,
|
||
|
|
}),
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def _mlm_success_manifest(workspace: Path, prefix: str = "cohort") -> Path:
|
||
|
|
final = workspace / "final"
|
||
|
|
final.mkdir(parents=True, exist_ok=True)
|
||
|
|
artifacts = {}
|
||
|
|
for role in ("mlm_full", "cmlm_full", "mlm_pvalues", "cmlm_pvalues", "report", "environment", "log"):
|
||
|
|
artifact = final / f"{prefix}.gwas_mlm.{role}.txt"
|
||
|
|
artifact.write_text(f"{role}\n", encoding="utf-8")
|
||
|
|
artifacts[role] = {
|
||
|
|
"role": role, "path": f"final/{artifact.name}", "full_path": str(artifact.resolve()),
|
||
|
|
"size": artifact.stat().st_size, "sha256": _sha(artifact),
|
||
|
|
}
|
||
|
|
manifest = {
|
||
|
|
"manifest_version": "1.0", "stage": "gwas_mlm", "prefix": prefix, "status": "success",
|
||
|
|
"genotype_scope_id": GENOTYPE, "phenotype_scope_id": PHENOTYPE,
|
||
|
|
"covariate_scope_id": COVARIATE, "model_scope_id": MODEL,
|
||
|
|
"analysis_scope_id": ANALYSIS, "elapsed_sec": 12.34, "artifacts": artifacts,
|
||
|
|
}
|
||
|
|
path = final / f"{prefix}.gwas_mlm.identity.manifest.json"
|
||
|
|
path.write_text(json.dumps(manifest), encoding="utf-8")
|
||
|
|
return path
|
||
|
|
|
||
|
|
|
||
|
|
def test_mlm_semantic_preflight_resolves_four_authoritative_manifests(monkeypatch, tmp_path):
|
||
|
|
workspace = tmp_path / "workspace" / "job"
|
||
|
|
expected = _four_upstreams(workspace)
|
||
|
|
monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace")
|
||
|
|
|
||
|
|
result = api._api_semantic_preflight("job", "gwas_mlm", "cohort", {
|
||
|
|
"genotypeScopeId": GENOTYPE, "phenotypeScopeId": PHENOTYPE,
|
||
|
|
"covariateScopeId": COVARIATE,
|
||
|
|
})
|
||
|
|
|
||
|
|
assert result["valid"] is True
|
||
|
|
assert set(result["decisions"]) == set(expected)
|
||
|
|
assert {key: Path(value["path"]) for key, value in result["decisions"].items()} == expected
|
||
|
|
assert result["materialized_params"] == {
|
||
|
|
"gwasSortedVcf": str(expected["sorted_vcf"]),
|
||
|
|
"gwasKinshipFile": str(expected["kinship_matrix"]),
|
||
|
|
"gwasTraitFile": str(expected["trait_table"]),
|
||
|
|
"gwasCovariateFile": str(expected["covariate_table"]),
|
||
|
|
"genotypeScopeId": GENOTYPE,
|
||
|
|
"phenotypeScopeId": PHENOTYPE,
|
||
|
|
"covariateScopeId": COVARIATE,
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def test_mlm_semantic_preflight_materializes_scopes_from_unique_upstream_manifests(
|
||
|
|
monkeypatch, tmp_path
|
||
|
|
):
|
||
|
|
workspace = tmp_path / "workspace" / "job"
|
||
|
|
expected = _four_upstreams(workspace)
|
||
|
|
monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace")
|
||
|
|
|
||
|
|
result = api._api_semantic_preflight("job", "gwas_mlm", "cohort", {})
|
||
|
|
|
||
|
|
assert result["valid"] is True
|
||
|
|
assert set(result["decisions"]) == set(expected)
|
||
|
|
assert result["materialized_params"] == {
|
||
|
|
"gwasSortedVcf": str(expected["sorted_vcf"]),
|
||
|
|
"gwasKinshipFile": str(expected["kinship_matrix"]),
|
||
|
|
"gwasTraitFile": str(expected["trait_table"]),
|
||
|
|
"gwasCovariateFile": str(expected["covariate_table"]),
|
||
|
|
"genotypeScopeId": GENOTYPE,
|
||
|
|
"phenotypeScopeId": PHENOTYPE,
|
||
|
|
"covariateScopeId": COVARIATE,
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize("field,bad", [
|
||
|
|
("genotypeScopeId", "genotype:" + "f" * 64),
|
||
|
|
("phenotypeScopeId", "phenotype:" + "f" * 64),
|
||
|
|
("covariateScopeId", "covariate:" + "f" * 64),
|
||
|
|
])
|
||
|
|
def test_mlm_semantic_preflight_fails_closed_on_wrong_scope(monkeypatch, tmp_path, field, bad):
|
||
|
|
workspace = tmp_path / "workspace" / "job"
|
||
|
|
_four_upstreams(workspace)
|
||
|
|
monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace")
|
||
|
|
params = {"genotypeScopeId": GENOTYPE, "phenotypeScopeId": PHENOTYPE, "covariateScopeId": COVARIATE}
|
||
|
|
params[field] = bad
|
||
|
|
result = api._api_semantic_preflight("job", "gwas_mlm", "cohort", params)
|
||
|
|
assert result["valid"] is False
|
||
|
|
assert result["issues"][0]["code"] == "invalid_gwas_mlm_upstream"
|
||
|
|
|
||
|
|
|
||
|
|
def test_mlm_semantic_preflight_rejects_ambiguous_manifest(monkeypatch, tmp_path):
|
||
|
|
workspace = tmp_path / "workspace" / "job"
|
||
|
|
_four_upstreams(workspace)
|
||
|
|
original = workspace / "final" / "cohort.gwas_trait.manifest.json"
|
||
|
|
(workspace / "final" / "cohort.gwas_trait.copy.manifest.json").write_bytes(original.read_bytes())
|
||
|
|
monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace")
|
||
|
|
result = api._api_semantic_preflight("job", "gwas_mlm", "cohort", {
|
||
|
|
"genotypeScopeId": GENOTYPE, "phenotypeScopeId": PHENOTYPE, "covariateScopeId": COVARIATE,
|
||
|
|
})
|
||
|
|
assert result["valid"] is False
|
||
|
|
assert "multiple gwas_trait manifests" in result["issues"][0]["message"]
|
||
|
|
|
||
|
|
|
||
|
|
def test_mlm_command_uses_canonical_runner_and_all_inputs_once(tmp_path):
|
||
|
|
command = backend._gwas_mlm_cli_command(
|
||
|
|
workspace=tmp_path, job_id="job", prefix="cohort",
|
||
|
|
sorted_vcf=tmp_path / "a.vcf.gz", kinship=tmp_path / "k.txt",
|
||
|
|
trait=tmp_path / "t.txt", covariate=tmp_path / "q.txt",
|
||
|
|
genotype_scope_id=GENOTYPE, phenotype_scope_id=PHENOTYPE,
|
||
|
|
covariate_scope_id=COVARIATE, tassel_pipeline="/opt/tassel/run_pipeline.pl",
|
||
|
|
xmx="20g",
|
||
|
|
)
|
||
|
|
assert command[:3] == [backend.sys.executable, "-m", "apps.cli.run_gwas_tassel_mlm"]
|
||
|
|
assert command.count("apps.cli.run_gwas_tassel_mlm") == 1
|
||
|
|
for flag in ("--vcf", "--kinship", "--trait", "--covariate", "--genotype-scope-id",
|
||
|
|
"--phenotype-scope-id", "--covariate-scope-id", "--tassel-pipeline", "--xmx"):
|
||
|
|
assert command.count(flag) == 1
|
||
|
|
assert "--log-file" not in command
|
||
|
|
|
||
|
|
|
||
|
|
def test_mlm_capability_roles_match_contract():
|
||
|
|
capability = get_stage_capability("gwas_mlm")
|
||
|
|
assert [item.role for item in capability.consumes] == [
|
||
|
|
"sorted_vcf", "kinship_matrix", "trait_table", "covariate_table",
|
||
|
|
]
|
||
|
|
assert [item.role for item in capability.provides] == [
|
||
|
|
"mlm_full", "cmlm_full", "mlm_pvalues", "cmlm_pvalues",
|
||
|
|
"manifest", "report", "environment",
|
||
|
|
]
|
||
|
|
|
||
|
|
|
||
|
|
def test_execute_request_accepts_all_mlm_paths_scopes_and_tool_params():
|
||
|
|
request = api.ExecuteRequest(
|
||
|
|
stage="gwas_mlm", gwasSortedVcf="v", gwasKinshipFile="k",
|
||
|
|
gwasTraitFile="t", gwasCovariateFile="q", genotypeScopeId=GENOTYPE,
|
||
|
|
phenotypeScopeId=PHENOTYPE, covariateScopeId=COVARIATE,
|
||
|
|
modelScopeId=MODEL, analysisScopeId=ANALYSIS,
|
||
|
|
tasselPipeline="/tassel/run_pipeline.pl", tasselXmx="20g",
|
||
|
|
)
|
||
|
|
assert request.modelScopeId == MODEL
|
||
|
|
assert request.gwasSortedVcf == "v"
|
||
|
|
assert request.gwasKinshipFile == "k"
|
||
|
|
assert request.gwasTraitFile == "t"
|
||
|
|
assert request.gwasCovariateFile == "q"
|
||
|
|
|
||
|
|
|
||
|
|
def test_stage_tool_executor_forwards_all_mlm_parameters(monkeypatch, tmp_path):
|
||
|
|
import asyncio
|
||
|
|
from gfa_core.mas.tools.stage_tools import StageToolExecutor, StageToolRegistry
|
||
|
|
|
||
|
|
captured = {}
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class FakeJobInfo:
|
||
|
|
job_id: str
|
||
|
|
input_dir: str
|
||
|
|
workspace_dir: str
|
||
|
|
logs_dir: str
|
||
|
|
created_at: str
|
||
|
|
|
||
|
|
def fake_run_stage(**kwargs):
|
||
|
|
captured.update(kwargs)
|
||
|
|
path = Path(kwargs["job"].workspace_dir) / "final" / "cohort.gwas_mlm.x.manifest.json"
|
||
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
||
|
|
path.write_text(json.dumps({"stage": "gwas_mlm", "status": "success"}), encoding="utf-8")
|
||
|
|
return "✅ done", path
|
||
|
|
|
||
|
|
fake_backend = types.ModuleType("apps.web.pipeline_backend")
|
||
|
|
fake_backend.JobInfo = FakeJobInfo
|
||
|
|
fake_backend.run_stage = fake_run_stage
|
||
|
|
monkeypatch.setitem(sys.modules, "apps.web.pipeline_backend", fake_backend)
|
||
|
|
registry = StageToolRegistry()
|
||
|
|
registry._executor = StageToolExecutor(project_root=tmp_path)
|
||
|
|
params = {
|
||
|
|
"gwasSortedVcf": "v", "gwasKinshipFile": "k", "gwasTraitFile": "t",
|
||
|
|
"gwasCovariateFile": "q", "genotypeScopeId": GENOTYPE,
|
||
|
|
"phenotypeScopeId": PHENOTYPE, "covariateScopeId": COVARIATE,
|
||
|
|
"tasselPipeline": "/tassel/run_pipeline.pl", "tasselXmx": "20g",
|
||
|
|
}
|
||
|
|
result = asyncio.run(registry.execute("gwas_mlm", "job", "cohort", params))
|
||
|
|
assert result["status"] == "success"
|
||
|
|
assert {key: captured[key] for key in (
|
||
|
|
"gwas_sorted_vcf", "gwas_kinship_file", "gwas_trait_file", "gwas_covariate_file",
|
||
|
|
"genotype_scope_id", "phenotype_scope_id", "covariate_scope_id",
|
||
|
|
"tassel_pipeline", "tassel_xmx",
|
||
|
|
)} == {
|
||
|
|
"gwas_sorted_vcf": "v", "gwas_kinship_file": "k", "gwas_trait_file": "t",
|
||
|
|
"gwas_covariate_file": "q", "genotype_scope_id": GENOTYPE,
|
||
|
|
"phenotype_scope_id": PHENOTYPE, "covariate_scope_id": COVARIATE,
|
||
|
|
"tassel_pipeline": "/tassel/run_pipeline.pl", "tassel_xmx": "20g",
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def test_api_task_forwards_all_materialized_mlm_parameters(monkeypatch, tmp_path):
|
||
|
|
import asyncio
|
||
|
|
import gfa_core.mas.resource_governor as resource_governor
|
||
|
|
|
||
|
|
captured = {}
|
||
|
|
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
|
||
|
|
|
||
|
|
def fake_run_stage(**kwargs):
|
||
|
|
captured.update(kwargs)
|
||
|
|
path = _mlm_success_manifest(workspace_root / "job")
|
||
|
|
return "✅ done", path
|
||
|
|
|
||
|
|
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["mlm-task"] = {"id": "mlm-task", "stage": "gwas_mlm", "job_id": "job",
|
||
|
|
"status": "pending", "message": "", "created_at": "2026-07-21T00:00:00"}
|
||
|
|
params = {"prefix": "cohort", "gwasSortedVcf": "v", "gwasKinshipFile": "k",
|
||
|
|
"gwasTraitFile": "t", "gwasCovariateFile": "q", "genotypeScopeId": GENOTYPE,
|
||
|
|
"phenotypeScopeId": PHENOTYPE, "covariateScopeId": COVARIATE,
|
||
|
|
"tasselPipeline": "/tassel/run_pipeline.pl", "tasselXmx": "20g"}
|
||
|
|
api.run_stage_sync("mlm-task", "gwas_mlm", "job", params)
|
||
|
|
assert {key: captured[key] for key in (
|
||
|
|
"gwas_sorted_vcf", "gwas_kinship_file", "gwas_trait_file", "gwas_covariate_file",
|
||
|
|
"genotype_scope_id", "phenotype_scope_id", "covariate_scope_id",
|
||
|
|
"tassel_pipeline", "tassel_xmx",
|
||
|
|
)} == {"gwas_sorted_vcf": "v", "gwas_kinship_file": "k", "gwas_trait_file": "t",
|
||
|
|
"gwas_covariate_file": "q", "genotype_scope_id": GENOTYPE,
|
||
|
|
"phenotype_scope_id": PHENOTYPE, "covariate_scope_id": COVARIATE,
|
||
|
|
"tassel_pipeline": "/tassel/run_pipeline.pl", "tassel_xmx": "20g"}
|
||
|
|
task = api.tasks["mlm-task"]
|
||
|
|
assert task["status"] == "success"
|
||
|
|
assert task["stage_identity"] == {
|
||
|
|
**task["stage_identity"],
|
||
|
|
"genotype_scope_id": GENOTYPE, "phenotype_scope_id": PHENOTYPE,
|
||
|
|
"covariate_scope_id": COVARIATE, "model_scope_id": MODEL,
|
||
|
|
"analysis_scope_id": ANALYSIS,
|
||
|
|
}
|
||
|
|
assert task["result"]["stage_identity"] == task["stage_identity"]
|
||
|
|
assert task["result"]["stage_status"]["model_scope_id"] == MODEL
|
||
|
|
assert task["result"]["stage_files"]["gwas_mlm"]["elapsed_time"] == 12.3
|
||
|
|
assert {item["role"] for item in task["result"]["stage_files"]["gwas_mlm"]["files"]} == {
|
||
|
|
"mlm_full", "cmlm_full", "mlm_pvalues", "cmlm_pvalues", "report", "environment", "log", "manifest",
|
||
|
|
}
|
||
|
|
query = {key: task["stage_identity"][key] for key in (
|
||
|
|
"genotype_scope_id", "phenotype_scope_id", "covariate_scope_id",
|
||
|
|
"model_scope_id", "analysis_scope_id",
|
||
|
|
)}
|
||
|
|
status = asyncio.run(api.get_stage_status(job_id="job", stage="gwas_mlm", prefix="cohort", **query))
|
||
|
|
results = asyncio.run(api.get_results(job_id="job", stage="gwas_mlm", **query))
|
||
|
|
assert status["completed"] is True
|
||
|
|
assert {item["role"] for item in results["gwas_mlm"]["files"]} >= {
|
||
|
|
"mlm_full", "cmlm_full", "mlm_pvalues", "cmlm_pvalues", "manifest",
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def test_backend_mlm_success_returns_only_manifest_changed_by_runner(monkeypatch, tmp_path):
|
||
|
|
workspace = tmp_path / "workspace"
|
||
|
|
decisions = {role: {"path": str(path)} for role, path in _four_upstreams(workspace).items()}
|
||
|
|
produced = workspace / "final" / "cohort.gwas_mlm.identity.manifest.json"
|
||
|
|
monkeypatch.setattr(backend, "_resolve_gwas_mlm_inputs", lambda **_kwargs: decisions)
|
||
|
|
|
||
|
|
def fake_run(_command, _log, cwd=None):
|
||
|
|
produced.write_text(json.dumps({
|
||
|
|
"stage": "gwas_mlm", "prefix": "cohort", "status": "success",
|
||
|
|
"genotype_scope_id": GENOTYPE, "phenotype_scope_id": PHENOTYPE,
|
||
|
|
"covariate_scope_id": COVARIATE, "model_scope_id": MODEL,
|
||
|
|
"analysis_scope_id": ANALYSIS,
|
||
|
|
}), encoding="utf-8")
|
||
|
|
return 0
|
||
|
|
|
||
|
|
monkeypatch.setattr(backend, "run_cmd_logged", fake_run)
|
||
|
|
message, manifest = backend.run_stage(
|
||
|
|
stage="gwas_mlm", 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,
|
||
|
|
)
|
||
|
|
assert message.startswith("✅")
|
||
|
|
assert manifest == produced
|
||
|
|
|
||
|
|
|
||
|
|
def test_backend_mlm_success_rejects_changed_manifest_with_wrong_identity(monkeypatch, tmp_path):
|
||
|
|
workspace = tmp_path / "workspace"
|
||
|
|
decisions = {role: {"path": str(path)} for role, path in _four_upstreams(workspace).items()}
|
||
|
|
wrong = workspace / "final" / "cohort.gwas_mlm.wrong.manifest.json"
|
||
|
|
monkeypatch.setattr(backend, "_resolve_gwas_mlm_inputs", lambda **_kwargs: decisions)
|
||
|
|
def fake_run(_command, _log, cwd=None):
|
||
|
|
wrong.write_text(json.dumps({
|
||
|
|
"stage": "gwas_mlm", "prefix": "cohort", "status": "success",
|
||
|
|
"genotype_scope_id": "genotype:" + "f" * 64,
|
||
|
|
"phenotype_scope_id": PHENOTYPE, "covariate_scope_id": COVARIATE,
|
||
|
|
"model_scope_id": MODEL, "analysis_scope_id": ANALYSIS,
|
||
|
|
}), encoding="utf-8")
|
||
|
|
return 0
|
||
|
|
monkeypatch.setattr(backend, "run_cmd_logged", fake_run)
|
||
|
|
message, manifest = backend.run_stage(
|
||
|
|
stage="gwas_mlm", 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,
|
||
|
|
)
|
||
|
|
assert message.startswith("❌")
|
||
|
|
assert manifest is None
|
||
|
|
|
||
|
|
|
||
|
|
def test_backend_mlm_failure_reports_log_and_never_returns_unrelated_manifest(monkeypatch, tmp_path):
|
||
|
|
workspace = tmp_path / "workspace"
|
||
|
|
decisions = {role: {"path": str(path)} for role, path in _four_upstreams(workspace).items()}
|
||
|
|
unrelated = workspace / "final" / "cohort.gwas_mlm.old.manifest.json"
|
||
|
|
unrelated.write_text(json.dumps({"stage": "gwas_mlm", "status": "success"}), encoding="utf-8")
|
||
|
|
monkeypatch.setattr(backend, "_resolve_gwas_mlm_inputs", lambda **_kwargs: decisions)
|
||
|
|
monkeypatch.setattr(backend, "run_cmd_logged", lambda *_args, **_kwargs: 7)
|
||
|
|
message, manifest = backend.run_stage(
|
||
|
|
stage="gwas_mlm", 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,
|
||
|
|
)
|
||
|
|
assert "exit code: 7" in message and "Log:" in message
|
||
|
|
assert manifest is None
|
||
|
|
|
||
|
|
|
||
|
|
def test_backend_mlm_real_cli_failure_log_has_clear_error_without_nul_bytes(monkeypatch, tmp_path):
|
||
|
|
workspace = tmp_path / "workspace"
|
||
|
|
decisions = {role: {"path": str(path)} for role, path in _four_upstreams(workspace).items()}
|
||
|
|
monkeypatch.setattr(backend, "_resolve_gwas_mlm_inputs", lambda **_kwargs: decisions)
|
||
|
|
|
||
|
|
message, manifest = backend.run_stage(
|
||
|
|
stage="gwas_mlm",
|
||
|
|
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=str(tmp_path / "missing_tassel"),
|
||
|
|
)
|
||
|
|
|
||
|
|
assert message.startswith("❌") and manifest is not None
|
||
|
|
log_files = list((workspace / "logs").glob("*_gwas_mlm.log"))
|
||
|
|
assert len(log_files) == 1
|
||
|
|
payload = log_files[0].read_bytes()
|
||
|
|
assert b"\x00" not in payload
|
||
|
|
assert payload.count(b"ERROR:") == 1
|
||
|
|
assert b"ERROR: prepared trait table header is invalid" in payload
|
||
|
|
|
||
|
|
|
||
|
|
def test_mlm_status_requires_full_exact_identity(monkeypatch, tmp_path):
|
||
|
|
workspace = tmp_path / "job"
|
||
|
|
final = workspace / "final"
|
||
|
|
final.mkdir(parents=True)
|
||
|
|
artifact = final / "cohort.gwas_mlm.out.tsv"
|
||
|
|
artifact.write_text("chr\tpos\tp\n1\t1\t0.5\n", encoding="utf-8")
|
||
|
|
manifest = {
|
||
|
|
"manifest_version": "1.0", "stage": "gwas_mlm", "prefix": "cohort", "status": "success",
|
||
|
|
"genotype_scope_id": GENOTYPE, "phenotype_scope_id": PHENOTYPE,
|
||
|
|
"covariate_scope_id": COVARIATE, "model_scope_id": MODEL, "analysis_scope_id": ANALYSIS,
|
||
|
|
"artifacts": {"mlm_pvalues": {"role": "mlm_pvalues", "path": f"final/{artifact.name}",
|
||
|
|
"full_path": str(artifact.resolve()), "size": artifact.stat().st_size, "sha256": _sha(artifact)}},
|
||
|
|
}
|
||
|
|
path = final / "cohort.gwas_mlm.abc.manifest.json"
|
||
|
|
path.write_text(json.dumps(manifest), encoding="utf-8")
|
||
|
|
identity = StageIdentity(stage="gwas_mlm", prefix="cohort", genotype_scope_id=GENOTYPE,
|
||
|
|
phenotype_scope_id=PHENOTYPE, covariate_scope_id=COVARIATE,
|
||
|
|
model_scope_id=MODEL, analysis_scope_id=ANALYSIS)
|
||
|
|
report = get_stage_status_for_identity("job", identity, workspace_root=tmp_path, strict_identity=True)
|
||
|
|
assert report.completed is True
|
||
|
|
identity.model_scope_id = "gwas-model:" + "f" * 64
|
||
|
|
wrong = get_stage_status_for_identity("job", identity, workspace_root=tmp_path, strict_identity=True)
|
||
|
|
assert wrong.completed is False
|
||
|
|
assert wrong.status == "not_found"
|
||
|
|
|
||
|
|
partial = StageIdentity(stage="gwas_mlm", prefix="cohort", genotype_scope_id=GENOTYPE,
|
||
|
|
phenotype_scope_id=PHENOTYPE, covariate_scope_id=COVARIATE)
|
||
|
|
partial_report = get_stage_status_for_identity("job", partial, workspace_root=tmp_path, strict_identity=True)
|
||
|
|
assert partial_report.completed is False
|
||
|
|
assert partial_report.status == "not_found"
|
||
|
|
|
||
|
|
|
||
|
|
def test_status_api_exposes_all_five_derived_mlm_scope_ids(monkeypatch, tmp_path):
|
||
|
|
import asyncio
|
||
|
|
|
||
|
|
_mlm_success_manifest(tmp_path / "workspace" / "job")
|
||
|
|
monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace")
|
||
|
|
exact = asyncio.run(api.get_stage_status(
|
||
|
|
job_id="job", stage="gwas_mlm", prefix="cohort",
|
||
|
|
genotype_scope_id=GENOTYPE, phenotype_scope_id=PHENOTYPE,
|
||
|
|
covariate_scope_id=COVARIATE, model_scope_id=MODEL, analysis_scope_id=ANALYSIS,
|
||
|
|
))
|
||
|
|
assert exact["completed"] is True
|
||
|
|
assert {key: exact[key] for key in (
|
||
|
|
"genotype_scope_id", "phenotype_scope_id", "covariate_scope_id",
|
||
|
|
"model_scope_id", "analysis_scope_id",
|
||
|
|
)} == {
|
||
|
|
"genotype_scope_id": GENOTYPE, "phenotype_scope_id": PHENOTYPE,
|
||
|
|
"covariate_scope_id": COVARIATE, "model_scope_id": MODEL,
|
||
|
|
"analysis_scope_id": ANALYSIS,
|
||
|
|
}
|
||
|
|
partial = asyncio.run(api.get_stage_status(
|
||
|
|
job_id="job", stage="gwas_mlm", prefix="cohort",
|
||
|
|
genotype_scope_id=GENOTYPE, phenotype_scope_id=PHENOTYPE,
|
||
|
|
covariate_scope_id=COVARIATE,
|
||
|
|
))
|
||
|
|
assert partial["completed"] is False and partial["status"] == "not_found"
|
||
|
|
|
||
|
|
|
||
|
|
def test_execute_mlm_never_reuses_completed_result_without_full_five_scope_identity(monkeypatch, tmp_path):
|
||
|
|
import asyncio
|
||
|
|
|
||
|
|
workspace = tmp_path / "workspace" / "job"
|
||
|
|
_four_upstreams(workspace)
|
||
|
|
_mlm_success_manifest(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)
|
||
|
|
request = api.ExecuteRequest(
|
||
|
|
stage="gwas_mlm", job_id="job", prefix="cohort",
|
||
|
|
genotypeScopeId=GENOTYPE, phenotypeScopeId=PHENOTYPE,
|
||
|
|
covariateScopeId=COVARIATE,
|
||
|
|
)
|
||
|
|
result = asyncio.run(api.execute_stage(request))
|
||
|
|
assert result.get("already_completed") is not True
|
||
|
|
assert result["async"] is True
|
||
|
|
assert len(dummy.calls) == 1
|
||
|
|
|
||
|
|
|
||
|
|
def test_execute_mlm_materializes_empty_request_scopes_before_submission(monkeypatch, tmp_path):
|
||
|
|
import asyncio
|
||
|
|
|
||
|
|
workspace = tmp_path / "workspace" / "job"
|
||
|
|
_four_upstreams(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)
|
||
|
|
request = api.ExecuteRequest(
|
||
|
|
stage="gwas_mlm", job_id="job", prefix="cohort",
|
||
|
|
)
|
||
|
|
|
||
|
|
result = asyncio.run(api.execute_stage(request))
|
||
|
|
|
||
|
|
assert result["async"] is True
|
||
|
|
assert request.genotypeScopeId == GENOTYPE
|
||
|
|
assert request.phenotypeScopeId == PHENOTYPE
|
||
|
|
assert request.covariateScopeId == COVARIATE
|
||
|
|
assert len(dummy.calls) == 1
|
||
|
|
|
||
|
|
|
||
|
|
def test_execute_mlm_reuses_completed_result_with_full_five_scope_identity(monkeypatch, tmp_path):
|
||
|
|
import asyncio
|
||
|
|
import gfa_core.mas.tools.stage_identity as stage_identity_module
|
||
|
|
|
||
|
|
workspace = tmp_path / "workspace" / "job"
|
||
|
|
_four_upstreams(workspace)
|
||
|
|
_mlm_success_manifest(workspace)
|
||
|
|
monkeypatch.setattr(api, "PROJECT_ROOT", tmp_path)
|
||
|
|
monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace")
|
||
|
|
monkeypatch.setattr(api, "INPUT_ROOT", tmp_path / "input")
|
||
|
|
real_status = stage_identity_module.get_stage_status_for_identity
|
||
|
|
monkeypatch.setattr(
|
||
|
|
stage_identity_module, "get_stage_status_for_identity",
|
||
|
|
lambda *args, **kwargs: real_status(
|
||
|
|
*args, **{**kwargs, "workspace_root": tmp_path / "workspace"},
|
||
|
|
),
|
||
|
|
)
|
||
|
|
result = asyncio.run(api.execute_stage(api.ExecuteRequest(
|
||
|
|
stage="gwas_mlm", job_id="job", prefix="cohort",
|
||
|
|
genotypeScopeId=GENOTYPE, phenotypeScopeId=PHENOTYPE,
|
||
|
|
covariateScopeId=COVARIATE, modelScopeId=MODEL,
|
||
|
|
analysisScopeId=ANALYSIS,
|
||
|
|
)))
|
||
|
|
assert result["already_completed"] is True
|
||
|
|
assert result["stage_status"]["model_scope_id"] == MODEL
|
||
|
|
|
||
|
|
|
||
|
|
def test_mlm_generic_no_scope_status_remains_available_for_legacy_ui(tmp_path):
|
||
|
|
_mlm_success_manifest(tmp_path / "job")
|
||
|
|
report = get_stage_status_for_identity(
|
||
|
|
"job", StageIdentity(stage="gwas_mlm", prefix="cohort"),
|
||
|
|
workspace_root=tmp_path, strict_identity=False,
|
||
|
|
)
|
||
|
|
assert report.completed is True
|
||
|
|
assert report.to_dict()["analysis_scope_id"] == ANALYSIS
|
||
|
|
|
||
|
|
|
||
|
|
def test_results_filters_mlm_by_all_scopes_and_preserves_roles_and_elapsed(monkeypatch, tmp_path):
|
||
|
|
import asyncio
|
||
|
|
|
||
|
|
workspace = tmp_path / "workspace" / "job"
|
||
|
|
final = workspace / "final"
|
||
|
|
final.mkdir(parents=True)
|
||
|
|
artifacts = {}
|
||
|
|
for role in ("mlm_full", "cmlm_full", "mlm_pvalues", "cmlm_pvalues", "report", "environment", "log"):
|
||
|
|
artifact = final / f"cohort.gwas_mlm.{role}.txt"
|
||
|
|
artifact.write_text(f"{role}\n", encoding="utf-8")
|
||
|
|
artifacts[role] = {"role": role, "path": f"final/{artifact.name}",
|
||
|
|
"full_path": str(artifact.resolve()), "size": artifact.stat().st_size, "sha256": _sha(artifact)}
|
||
|
|
manifest = {
|
||
|
|
"manifest_version": "1.0", "stage": "gwas_mlm", "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": 12.34, "artifacts": artifacts,
|
||
|
|
}
|
||
|
|
manifest_path = final / "cohort.gwas_mlm.identity.manifest.json"
|
||
|
|
manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
|
||
|
|
monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace")
|
||
|
|
payload = asyncio.run(api.get_results(
|
||
|
|
job_id="job", stage="gwas_mlm", genotype_scope_id=GENOTYPE,
|
||
|
|
phenotype_scope_id=PHENOTYPE, covariate_scope_id=COVARIATE,
|
||
|
|
model_scope_id=MODEL, analysis_scope_id=ANALYSIS,
|
||
|
|
))
|
||
|
|
assert payload["gwas_mlm"]["elapsed_time"] == 12.3
|
||
|
|
assert {item["role"] for item in payload["gwas_mlm"]["files"]} >= {
|
||
|
|
"mlm_full", "cmlm_full", "mlm_pvalues", "cmlm_pvalues", "report", "environment", "log", "manifest",
|
||
|
|
}
|
||
|
|
wrong = asyncio.run(api.get_results(
|
||
|
|
job_id="job", stage="gwas_mlm", genotype_scope_id=GENOTYPE,
|
||
|
|
phenotype_scope_id=PHENOTYPE, covariate_scope_id=COVARIATE,
|
||
|
|
model_scope_id="gwas-model:" + "f" * 64, analysis_scope_id=ANALYSIS,
|
||
|
|
))
|
||
|
|
assert wrong["gwas_mlm"]["files"] == []
|
||
|
|
partial = asyncio.run(api.get_results(
|
||
|
|
job_id="job", stage="gwas_mlm", genotype_scope_id=GENOTYPE,
|
||
|
|
phenotype_scope_id=PHENOTYPE, covariate_scope_id=COVARIATE,
|
||
|
|
))
|
||
|
|
assert partial["gwas_mlm"]["files"] == []
|