451 lines
18 KiB
Python
451 lines
18 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
import sys
|
|
import types
|
|
from dataclasses import dataclass
|
|
|
|
import pytest
|
|
|
|
from apps.web import pipeline_backend as backend
|
|
from apps.web.vue3 import main as api
|
|
|
|
|
|
SCOPE = "genotype:" + "a" * 64
|
|
|
|
|
|
def _record(path: Path, **extra) -> dict:
|
|
return {
|
|
"path": str(path.resolve()), "size": path.stat().st_size,
|
|
"sha256": hashlib.sha256(path.read_bytes()).hexdigest(), **extra,
|
|
}
|
|
|
|
|
|
def _bound_inputs(project: Path, job_id: str = "job") -> tuple[Path, Path, Path]:
|
|
data = project.parent / "data"
|
|
data.mkdir(parents=True)
|
|
phenotype = data / "phenotype.tsv"
|
|
q_file = data / "cohort.2.Q"
|
|
fam_file = data / "cohort.fam"
|
|
phenotype.write_text("ID\tGC_BLUP\nS1\t1\nS2\t2\n", encoding="utf-8")
|
|
q_file.write_text("0.4 0.6\n0.7 0.3\n", encoding="utf-8")
|
|
fam_file.write_text("F1 S1 0 0 0 -9\nF2 S2 0 0 0 -9\n", encoding="utf-8")
|
|
workspace = project / "workspace" / job_id
|
|
workspace.mkdir(parents=True)
|
|
binding = {
|
|
"manifest_version": "1.0", "job_id": job_id,
|
|
"inputs": {
|
|
"phenotype_table": _record(phenotype),
|
|
"admixture_q": _record(q_file, pair_id="k2", source_genotype_scope_id=SCOPE),
|
|
"sample_fam": _record(fam_file, pair_id="k2", source_genotype_scope_id=SCOPE),
|
|
},
|
|
}
|
|
(workspace / "external_inputs.manifest.json").write_text(json.dumps(binding), encoding="utf-8")
|
|
final = workspace / "final"
|
|
final.mkdir()
|
|
producer = {
|
|
"manifest_version": "1.0", "stage": "pop_struct", "status": "success",
|
|
"run_id": "pop_struct_k2", "genotype_scope_id": SCOPE,
|
|
"artifacts": {
|
|
"admixture_q": _record(q_file, role="admixture_q"),
|
|
"sample_fam": _record(fam_file, role="sample_fam"),
|
|
},
|
|
}
|
|
(final / "cohort.pop_struct.manifest.json").write_text(json.dumps(producer), encoding="utf-8")
|
|
return phenotype, q_file, fam_file
|
|
|
|
|
|
def test_trait_and_cov_preflight_materialize_exact_root_inputs(monkeypatch, tmp_path: Path) -> None:
|
|
project = tmp_path / "sandbox"
|
|
phenotype, q_file, fam_file = _bound_inputs(project)
|
|
monkeypatch.setattr(api, "PROJECT_ROOT", project)
|
|
monkeypatch.setattr(api, "WORKSPACE_ROOT", project / "workspace")
|
|
|
|
trait = api._api_semantic_preflight("job", "gwas_trait", "cohort", {})
|
|
assert trait["valid"] is True
|
|
assert trait["materialized_params"] == {
|
|
"phenotypeFile": str(phenotype.resolve()),
|
|
"phenotypeSampleColumn": "ID",
|
|
"phenotypeTraitColumn": "GC_BLUP",
|
|
"phenotypeTraitName": "GC_BLUP",
|
|
}
|
|
|
|
cov = api._api_semantic_preflight("job", "gwas_cov", "cohort", {"gwasQK": 2, "genotypeScopeId": SCOPE})
|
|
assert cov["valid"] is True
|
|
assert cov["materialized_params"] == {
|
|
"gwasQFile": str(q_file.resolve()),
|
|
"gwasFamFile": str(fam_file.resolve()),
|
|
"gwasQK": 2,
|
|
"genotypeScopeId": SCOPE,
|
|
}
|
|
|
|
|
|
def test_trait_preflight_rejects_authoritative_comma_delimited_txt(monkeypatch, tmp_path: Path) -> None:
|
|
project = tmp_path / "sandbox"
|
|
workspace = project / "workspace" / "job"
|
|
phenotype = project.parent / "data" / "phenotype.txt"
|
|
workspace.mkdir(parents=True)
|
|
phenotype.parent.mkdir()
|
|
phenotype.write_text("ID,trait\nS1,1\n", encoding="utf-8")
|
|
binding = {
|
|
"manifest_version": "1.0", "job_id": "job",
|
|
"inputs": {"phenotype_table": _record(phenotype)},
|
|
}
|
|
(workspace / "external_inputs.manifest.json").write_text(json.dumps(binding), encoding="utf-8")
|
|
monkeypatch.setattr(api, "PROJECT_ROOT", project)
|
|
monkeypatch.setattr(api, "WORKSPACE_ROOT", project / "workspace")
|
|
|
|
result = api._api_semantic_preflight("job", "gwas_trait", "cohort", {})
|
|
|
|
assert result["valid"] is False
|
|
assert result["needs_confirmation"] is False
|
|
assert result["issues"][0]["code"] == "invalid_external_binding"
|
|
|
|
|
|
def test_trait_preflight_rejects_authoritative_leading_blank_line(monkeypatch, tmp_path: Path) -> None:
|
|
project = tmp_path / "sandbox"
|
|
workspace = project / "workspace" / "job"
|
|
phenotype = project.parent / "data" / "phenotype.tsv"
|
|
workspace.mkdir(parents=True)
|
|
phenotype.parent.mkdir()
|
|
phenotype.write_text("\nID\ttrait\nS1\t1\n", encoding="utf-8")
|
|
binding = {
|
|
"manifest_version": "1.0", "job_id": "job",
|
|
"inputs": {"phenotype_table": _record(phenotype)},
|
|
}
|
|
(workspace / "external_inputs.manifest.json").write_text(json.dumps(binding), encoding="utf-8")
|
|
monkeypatch.setattr(api, "PROJECT_ROOT", project)
|
|
monkeypatch.setattr(api, "WORKSPACE_ROOT", project / "workspace")
|
|
|
|
result = api._api_semantic_preflight("job", "gwas_trait", "cohort", {})
|
|
|
|
assert result["valid"] is False
|
|
assert "materialized_params" not in result
|
|
assert result["issues"][0]["code"] == "invalid_external_binding"
|
|
|
|
|
|
def test_trait_and_cov_backend_commands_invoke_only_canonical_runners(tmp_path: Path) -> None:
|
|
trait = backend._gwas_trait_cli_command(
|
|
workspace=tmp_path, job_id="job", prefix="cohort",
|
|
phenotype=tmp_path / "p.tsv", sample_column="ID",
|
|
trait_column="GC_BLUP", trait_name="heading_days",
|
|
log_file=tmp_path / "trait.log",
|
|
)
|
|
assert trait.count("apps.cli.run_gwas_trait") == 1
|
|
assert "awk" not in trait
|
|
assert "--trait-name" in trait
|
|
|
|
cov = backend._gwas_cov_cli_command(
|
|
workspace=tmp_path, job_id="job", prefix="cohort",
|
|
q_file=tmp_path / "a.Q", fam_file=tmp_path / "a.fam", k=2,
|
|
genotype_scope_id=SCOPE, log_file=tmp_path / "cov.log",
|
|
)
|
|
assert cov.count("apps.cli.run_gwas_covariate") == 1
|
|
assert "paste" not in cov
|
|
assert "--source-genotype-scope-id" in cov
|
|
|
|
|
|
def test_run_stage_dispatches_materialized_trait_values_without_protein_gate(monkeypatch, tmp_path: Path) -> None:
|
|
workspace = tmp_path / "workspace" / "job"
|
|
input_dir = tmp_path / "input" / "job"
|
|
logs = workspace / "logs"
|
|
final = workspace / "final"
|
|
input_dir.mkdir(parents=True)
|
|
logs.mkdir(parents=True)
|
|
final.mkdir()
|
|
phenotype = tmp_path / "phenotype.tsv"
|
|
phenotype.write_text("ID\tvalue\nS1\t1\nS2\t2\n", encoding="utf-8")
|
|
captured: list[str] = []
|
|
|
|
def fake_run(command, _log_file, cwd=None):
|
|
captured.extend(command)
|
|
(final / "cohort.gwas_trait.abc.manifest.json").write_text("{}", encoding="utf-8")
|
|
return 0
|
|
|
|
monkeypatch.setattr(backend, "can_start_new_task", lambda: (True, ""))
|
|
monkeypatch.setattr(backend, "run_cmd_logged", fake_run)
|
|
message, manifest = backend.run_stage(
|
|
stage="gwas_trait",
|
|
job=backend.JobInfo("job", str(input_dir), str(workspace), str(logs), "now"),
|
|
prefix="cohort",
|
|
phenotype_file=str(phenotype),
|
|
phenotype_sample_column="ID",
|
|
phenotype_trait_column="value",
|
|
phenotype_trait_name="heading_days",
|
|
)
|
|
|
|
assert message.startswith("✅")
|
|
assert manifest == final / "cohort.gwas_trait.abc.manifest.json"
|
|
assert captured[:3] == [backend.sys.executable, "-m", "apps.cli.run_gwas_trait"]
|
|
assert captured[captured.index("--phenotype") + 1] == str(phenotype)
|
|
assert captured[captured.index("--sample-column") + 1] == "ID"
|
|
assert captured[captured.index("--trait-column") + 1] == "value"
|
|
assert captured[captured.index("--trait-name") + 1] == "heading_days"
|
|
|
|
|
|
def test_run_stage_dispatches_materialized_covariate_values_without_protein_gate(monkeypatch, tmp_path: Path) -> None:
|
|
workspace = tmp_path / "workspace" / "job"
|
|
input_dir = tmp_path / "input" / "job"
|
|
logs = workspace / "logs"
|
|
final = workspace / "final"
|
|
input_dir.mkdir(parents=True)
|
|
logs.mkdir(parents=True)
|
|
final.mkdir()
|
|
q_file = tmp_path / "cohort.2.Q"
|
|
fam_file = tmp_path / "cohort.fam"
|
|
q_file.write_text("0.4 0.6\n", encoding="utf-8")
|
|
fam_file.write_text("F1 S1 0 0 0 -9\n", encoding="utf-8")
|
|
captured: list[str] = []
|
|
|
|
def fake_run(command, _log_file, cwd=None):
|
|
captured.extend(command)
|
|
(final / "cohort.gwas_cov.abc.manifest.json").write_text("{}", encoding="utf-8")
|
|
return 0
|
|
|
|
monkeypatch.setattr(backend, "can_start_new_task", lambda: (True, ""))
|
|
monkeypatch.setattr(backend, "run_cmd_logged", fake_run)
|
|
message, manifest = backend.run_stage(
|
|
stage="gwas_cov",
|
|
job=backend.JobInfo("job", str(input_dir), str(workspace), str(logs), "now"),
|
|
prefix="cohort",
|
|
gwas_q_file=str(q_file),
|
|
gwas_fam_file=str(fam_file),
|
|
gwas_q_k=2,
|
|
genotype_scope_id=SCOPE,
|
|
)
|
|
|
|
assert message.startswith("✅")
|
|
assert manifest == final / "cohort.gwas_cov.abc.manifest.json"
|
|
assert captured[:3] == [backend.sys.executable, "-m", "apps.cli.run_gwas_covariate"]
|
|
assert captured[captured.index("--q") + 1] == str(q_file)
|
|
assert captured[captured.index("--fam") + 1] == str(fam_file)
|
|
assert captured[captured.index("--k") + 1] == "2"
|
|
assert captured[captured.index("--source-genotype-scope-id") + 1] == SCOPE
|
|
|
|
|
|
def test_run_stage_sync_propagates_materialized_covariate_values(monkeypatch, tmp_path: Path) -> None:
|
|
resource_governor = __import__("gfa_core.mas.resource_governor", fromlist=["ResourceLease"])
|
|
workspace_root = tmp_path / "workspace"
|
|
input_root = tmp_path / "input"
|
|
(input_root / "job").mkdir(parents=True)
|
|
q_file = tmp_path / "cohort.2.Q"
|
|
fam_file = tmp_path / "cohort.fam"
|
|
q_file.write_text("0.4 0.6\n", encoding="utf-8")
|
|
fam_file.write_text("F1 S1 0 0 0 -9\n", encoding="utf-8")
|
|
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)
|
|
manifest = workspace_root / "job" / "final" / "cohort.gwas_cov.abc.manifest.json"
|
|
manifest.parent.mkdir(parents=True, exist_ok=True)
|
|
manifest.write_text(json.dumps({"status": "success", "stage": "gwas_cov", "artifacts": {}}), encoding="utf-8")
|
|
return "✅ done", manifest
|
|
|
|
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", input_root)
|
|
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["task"] = {
|
|
"id": "task", "stage": "gwas_cov", "job_id": "job", "status": "pending",
|
|
"message": "", "created_at": "2026-07-21T00:00:00",
|
|
}
|
|
|
|
api.run_stage_sync("task", "gwas_cov", "job", {
|
|
"prefix": "cohort", "gwasQFile": str(q_file), "gwasFamFile": str(fam_file),
|
|
"gwasQK": 2, "genotypeScopeId": SCOPE,
|
|
})
|
|
|
|
assert captured["gwas_q_file"] == str(q_file)
|
|
assert captured["gwas_fam_file"] == str(fam_file)
|
|
assert captured["gwas_q_k"] == 2
|
|
assert captured["genotype_scope_id"] == SCOPE
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("stage", "scope_params"),
|
|
[
|
|
("gwas_trait", {"phenotypeScopeId": "phenotype:" + "b" * 64, "analysisScopeId": "gwas-analysis:" + "c" * 64}),
|
|
("gwas_cov", {"genotypeScopeId": SCOPE, "covariateScopeId": "covariate:" + "d" * 64, "analysisScopeId": "gwas-analysis:" + "e" * 64}),
|
|
],
|
|
)
|
|
def test_execute_strict_lookup_receives_requested_gwas_root_identity(monkeypatch, tmp_path: Path, stage: str, scope_params: dict) -> None:
|
|
import asyncio
|
|
import gfa_core.mas.tools.stage_identity as stage_identity
|
|
|
|
captured = {}
|
|
|
|
def fake_status(*, identity, **_kwargs):
|
|
captured.update(identity.to_dict())
|
|
return SimpleNamespace(completed=True, errors=[], to_dict=lambda: {"completed": True})
|
|
|
|
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(stage_identity, "get_stage_status_for_identity", fake_status)
|
|
request = api.ExecuteRequest(stage=stage, job_id="job", prefix="cohort", **scope_params)
|
|
|
|
result = asyncio.run(api.execute_stage(request))
|
|
|
|
assert result["already_completed"] is True
|
|
for key, value in scope_params.items():
|
|
snake = {
|
|
"genotypeScopeId": "genotype_scope_id",
|
|
"phenotypeScopeId": "phenotype_scope_id",
|
|
"covariateScopeId": "covariate_scope_id",
|
|
"analysisScopeId": "analysis_scope_id",
|
|
}[key]
|
|
assert captured[snake] == value
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("stage", "artifact_name"),
|
|
[("gwas_trait", "cohort.gwas_trait.abc.trait.txt"), ("gwas_cov", "cohort.gwas_cov.abc.covariates.txt")],
|
|
)
|
|
def test_run_stage_sync_reports_success_after_gwas_root_cli(monkeypatch, tmp_path: Path, stage: str, artifact_name: str) -> None:
|
|
import gfa_core.mas.resource_governor as resource_governor
|
|
import gfa_core.mas.tools.stage_identity as stage_identity
|
|
|
|
workspace_root = tmp_path / "workspace"
|
|
input_root = tmp_path / "input"
|
|
workspace = workspace_root / "job"
|
|
(input_root / "job").mkdir(parents=True)
|
|
final = workspace / "final"
|
|
final.mkdir(parents=True)
|
|
|
|
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):
|
|
artifact = final / artifact_name
|
|
artifact.write_text("result\n", encoding="utf-8")
|
|
manifest = final / f"cohort.{stage}.abc.manifest.json"
|
|
manifest.write_text(json.dumps({
|
|
"manifest_version": "1.0", "stage": stage, "prefix": "cohort",
|
|
"status": "success", "artifacts": {},
|
|
}), encoding="utf-8")
|
|
return f"✅ {stage} done", manifest
|
|
|
|
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(stage_identity, "get_stage_status_for_identity", lambda **_kwargs: SimpleNamespace(
|
|
completed=True, errors=[], to_dict=lambda: {"completed": True},
|
|
))
|
|
monkeypatch.setattr(api, "WORKSPACE_ROOT", workspace_root)
|
|
monkeypatch.setattr(api, "INPUT_ROOT", input_root)
|
|
monkeypatch.setattr(api, "MAS_AVAILABLE", False)
|
|
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)
|
|
task_id = f"task-{stage}"
|
|
api.tasks[task_id] = {
|
|
"id": task_id, "stage": stage, "job_id": "job", "status": "pending",
|
|
"message": "", "created_at": "2026-07-21T00:00:00",
|
|
}
|
|
|
|
result = api.run_stage_sync(task_id, stage, "job", {"prefix": "cohort"})
|
|
|
|
assert result["success"] is True
|
|
assert result["stage"] == stage
|
|
assert api.tasks[task_id]["status"] == "success"
|
|
assert any(item["name"] == artifact_name for item in result["stage_files"][stage]["files"])
|
|
|
|
|
|
def _execute_registry_with_captured_backend(monkeypatch, tmp_path: Path, stage: str, params: dict) -> tuple[dict, dict]:
|
|
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)
|
|
manifest_path = Path(kwargs["job"].workspace_dir) / "final" / f"cohort.{stage}.manifest.json"
|
|
manifest_path.parent.mkdir(parents=True, exist_ok=True)
|
|
manifest_path.write_text(json.dumps({
|
|
"manifest_version": "1.0", "stage": stage, "prefix": "cohort",
|
|
"status": "success", "artifacts": {},
|
|
}), encoding="utf-8")
|
|
return f"✅ {stage} done", manifest_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)
|
|
result = asyncio.run(registry.execute(stage, "job", "cohort", params))
|
|
return result, captured
|
|
|
|
|
|
def test_registry_forwards_exact_stage20_materialized_values(monkeypatch, tmp_path: Path) -> None:
|
|
phenotype = tmp_path / "phenotype.tsv"
|
|
params = {
|
|
"phenotypeFile": str(phenotype),
|
|
"phenotype_sample_column": "ID",
|
|
"phenotypeTraitColumn": "height",
|
|
"phenotype_trait_name": "plant_height",
|
|
}
|
|
|
|
result, captured = _execute_registry_with_captured_backend(
|
|
monkeypatch, tmp_path, "gwas_trait", params,
|
|
)
|
|
|
|
assert result["status"] == "success"
|
|
assert captured["phenotype_file"] == str(phenotype)
|
|
assert captured["phenotype_sample_column"] == "ID"
|
|
assert captured["phenotype_trait_column"] == "height"
|
|
assert captured["phenotype_trait_name"] == "plant_height"
|
|
|
|
|
|
def test_registry_forwards_exact_stage21_materialized_values(monkeypatch, tmp_path: Path) -> None:
|
|
q_file = tmp_path / "cohort.3.Q"
|
|
fam_file = tmp_path / "cohort.fam"
|
|
params = {
|
|
"gwas_q_file": str(q_file),
|
|
"gwasFamFile": str(fam_file),
|
|
"gwas_q_k": 3,
|
|
}
|
|
|
|
result, captured = _execute_registry_with_captured_backend(
|
|
monkeypatch, tmp_path, "gwas_cov", params,
|
|
)
|
|
|
|
assert result["status"] == "success"
|
|
assert captured["gwas_q_file"] == str(q_file)
|
|
assert captured["gwas_fam_file"] == str(fam_file)
|
|
assert captured["gwas_q_k"] == 3
|