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"] == []