211 lines
7.0 KiB
Python
211 lines
7.0 KiB
Python
from __future__ import annotations
|
|
|
|
import inspect
|
|
import json
|
|
from pathlib import Path
|
|
import runpy
|
|
|
|
from apps.web.vue3.main import ExecuteRequest
|
|
from gfa_core.mas.agents.planning_v2 import PlanningAgentV2
|
|
from gfa_core.mas.stage_capabilities import CAPABILITIES
|
|
from gfa_core.mas.stage_catalog import EXECUTABLE_STAGES, STAGE_DEPENDENCIES
|
|
from gfa_core.mas.tools.stage_tools import StageToolRegistry
|
|
_CLI_FIXTURES = runpy.run_path(
|
|
str(Path(__file__).with_name("test_wgs_variant_select_cli.py"))
|
|
)
|
|
JOINT_ANALYSIS = _CLI_FIXTURES["JOINT_ANALYSIS"]
|
|
REFERENCE = _CLI_FIXTURES["REFERENCE"]
|
|
SAMPLES = _CLI_FIXTURES["SAMPLES"]
|
|
_joint_authority = _CLI_FIXTURES["_joint_authority"]
|
|
_tool = _CLI_FIXTURES["_tool"]
|
|
|
|
|
|
def test_wgs_variant_select_vertical_registration_is_exact() -> None:
|
|
assert STAGE_DEPENDENCIES["wgs_variant_select"] == [
|
|
"wgs_joint_genotyping"
|
|
]
|
|
assert "wgs_variant_select" in EXECUTABLE_STAGES
|
|
capability = CAPABILITIES["wgs_variant_select"]
|
|
assert {item.role for item in capability.consumes} == {
|
|
"joint_vcf",
|
|
"joint_vcf_index",
|
|
}
|
|
assert {item.role for item in capability.provides} == {
|
|
"selected_snp_vcf",
|
|
"selected_snp_vcf_index",
|
|
"selected_indel_vcf",
|
|
"selected_indel_vcf_index",
|
|
"variant_select_report",
|
|
}
|
|
definition = next(
|
|
item
|
|
for item in StageToolRegistry.STAGE_DEFINITIONS
|
|
if item["id"] == "wgs_variant_select"
|
|
)
|
|
assert definition["script"] == "run_wgs_variant_select.py"
|
|
assert definition["required_inputs"] == ["wgs_joint_manifest"]
|
|
assert PlanningAgentV2.REQUIRED_INPUTS["wgs_variant_select"] == []
|
|
|
|
|
|
def test_api_and_worker_boundary_carry_method_and_both_tools() -> None:
|
|
request = ExecuteRequest(
|
|
stage="wgs_variant_select",
|
|
wgsJointManifest="final/Cohort.wgs_joint_genotyping.manifest.json",
|
|
wgsMethod="bcftools",
|
|
wgsGatk="/opt/gatk",
|
|
wgsBcftools="/opt/bcftools",
|
|
)
|
|
assert request.wgsJointManifest.endswith(
|
|
".wgs_joint_genotyping.manifest.json"
|
|
)
|
|
assert request.wgsMethod == "bcftools"
|
|
assert request.wgsGatk == "/opt/gatk"
|
|
assert request.wgsBcftools == "/opt/bcftools"
|
|
parameters = inspect.signature(
|
|
__import__("apps.web.pipeline_backend", fromlist=["run_stage"]).run_stage
|
|
).parameters
|
|
assert "wgs_joint_manifest" in parameters
|
|
assert "wgs_method" in parameters
|
|
assert "wgs_bcftools" in parameters
|
|
|
|
|
|
def _backend_kwargs(
|
|
workspace: Path,
|
|
joint: Path,
|
|
gatk: Path,
|
|
bcftools: Path,
|
|
method: str,
|
|
) -> dict:
|
|
from apps.web import pipeline_backend
|
|
|
|
return {
|
|
"stage": "wgs_variant_select",
|
|
"job": pipeline_backend.JobInfo(
|
|
"job-1", "", str(workspace), str(workspace / "logs"), "now"
|
|
),
|
|
"prefix": "Cohort",
|
|
"wgs_joint_manifest": str(joint),
|
|
"wgs_sample_set_id": SAMPLES,
|
|
"wgs_reference_id": REFERENCE,
|
|
"wgs_joint_analysis_id": JOINT_ANALYSIS,
|
|
"wgs_method": method,
|
|
"wgs_gatk": str(gatk),
|
|
"wgs_bcftools": str(bcftools),
|
|
}
|
|
|
|
|
|
def test_backend_builds_exact_method_command_and_accepts_fresh_manifest(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
from apps.cli.run_wgs_variant_select import main
|
|
from apps.web import pipeline_backend
|
|
|
|
workspace = tmp_path / "workspace"
|
|
joint = _joint_authority(workspace)
|
|
gatk = _tool(tmp_path, "gatk")
|
|
bcftools = _tool(tmp_path, "bcftools")
|
|
calls: list[list[str]] = []
|
|
|
|
def fake_run(command, *_args, **_kwargs):
|
|
calls.append(command)
|
|
return main(command[3:])
|
|
|
|
monkeypatch.setattr(pipeline_backend, "run_cmd_logged", fake_run)
|
|
for method in ("gatk", "bcftools"):
|
|
message, output = pipeline_backend.run_stage(
|
|
**_backend_kwargs(
|
|
workspace, joint, gatk, bcftools, method
|
|
)
|
|
)
|
|
assert "completed" in message.lower()
|
|
assert output == (
|
|
workspace
|
|
/ "final"
|
|
/ f"Cohort.wgs_variant_select.{method}.manifest.json"
|
|
)
|
|
manifest = json.loads(output.read_text())
|
|
assert manifest["method"] == method
|
|
assert manifest["parameters"]["method"] == method
|
|
assert set(manifest["software"]) == {method}
|
|
command = calls[-1]
|
|
assert command[:3] == [
|
|
pipeline_backend.sys.executable,
|
|
"-m",
|
|
"apps.cli.run_wgs_variant_select",
|
|
]
|
|
assert command[command.index("--joint-manifest") + 1] == str(joint)
|
|
assert command[command.index("--method") + 1] == method
|
|
|
|
|
|
def test_only_runtime_gatk_failure_returns_real_reason_and_fallback_metadata(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
from apps.cli.run_wgs_variant_select import main
|
|
from apps.web import pipeline_backend
|
|
|
|
workspace = tmp_path / "workspace"
|
|
joint = _joint_authority(workspace)
|
|
gatk = _tool(tmp_path, "gatk")
|
|
bcftools = _tool(tmp_path, "bcftools")
|
|
monkeypatch.setenv("FAIL_POINT", "gatk:select:SNP")
|
|
monkeypatch.setattr(
|
|
pipeline_backend,
|
|
"run_cmd_logged",
|
|
lambda command, *_args, **_kwargs: main(command[3:]),
|
|
)
|
|
message, failure_path = pipeline_backend.run_stage(
|
|
**_backend_kwargs(
|
|
workspace, joint, gatk, bcftools, "gatk"
|
|
)
|
|
)
|
|
failure = json.loads(failure_path.read_text())
|
|
assert "SelectVariants SNP failed with exit code 23" in message
|
|
assert "SelectVariants SNP failed with exit code 23" in failure["errors"][0]
|
|
assert failure["fallback_method"] == "bcftools"
|
|
assert failure["method"] == "gatk"
|
|
|
|
|
|
def test_bcftools_failure_and_upstream_preflight_are_not_fallback_eligible(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
from apps.cli.run_wgs_variant_select import main
|
|
from apps.web import pipeline_backend
|
|
|
|
workspace = tmp_path / "workspace"
|
|
joint = _joint_authority(workspace)
|
|
gatk = _tool(tmp_path, "gatk")
|
|
bcftools = _tool(tmp_path, "bcftools")
|
|
monkeypatch.setenv("FAIL_POINT", "bcftools:select:SNP")
|
|
monkeypatch.setattr(
|
|
pipeline_backend,
|
|
"run_cmd_logged",
|
|
lambda command, *_args, **_kwargs: main(command[3:]),
|
|
)
|
|
_, failure_path = pipeline_backend.run_stage(
|
|
**_backend_kwargs(
|
|
workspace, joint, gatk, bcftools, "bcftools"
|
|
)
|
|
)
|
|
failure = json.loads(failure_path.read_text())
|
|
assert "fallback_method" not in failure
|
|
payload = json.loads(joint.read_text())
|
|
payload.pop("wgs_analysis_id")
|
|
joint.write_text(json.dumps(payload))
|
|
message, output = pipeline_backend.run_stage(
|
|
**_backend_kwargs(
|
|
workspace, joint, gatk, bcftools, "gatk"
|
|
)
|
|
)
|
|
assert output is None
|
|
assert "fallback" not in message.lower()
|
|
|
|
|
|
def test_worker_failure_payload_surfaces_only_manifest_approved_fallback() -> None:
|
|
source = inspect.getsource(
|
|
__import__(
|
|
"gfa_core.mas.tools.stage_tools",
|
|
fromlist=["StageToolExecutor"],
|
|
).StageToolExecutor.execute
|
|
)
|
|
assert 'payload["fallback_method"] = manifest["fallback_method"]' in source
|