348 lines
11 KiB
Python
348 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
|
|
|
|
def _write_identify_manifest(workspace: Path, prefix: str = "Ft.SPL") -> Path:
|
|
final_dir = workspace / "final"
|
|
final_dir.mkdir(parents=True, exist_ok=True)
|
|
table = final_dir / f"{prefix}.identify.tsv"
|
|
table.write_text("gene\tvalue\nFtSPL1\t1\n", encoding="utf-8")
|
|
manifest = final_dir / f"{prefix}.identify.manifest.json"
|
|
manifest.write_text(
|
|
json.dumps(
|
|
{
|
|
"manifest_version": "1.0",
|
|
"stage": "identify",
|
|
"status": "success",
|
|
"prefix": prefix,
|
|
"metrics": {"member_count": 1},
|
|
"artifacts": {
|
|
"members": {
|
|
"path": f"final/{table.name}",
|
|
"full_path": str(table),
|
|
"type": "table",
|
|
}
|
|
},
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
return manifest
|
|
|
|
|
|
def _flag_value(command: list[str], flag: str) -> str:
|
|
return command[command.index(flag) + 1]
|
|
|
|
|
|
def test_report_stage_is_registered_without_analysis_dependencies() -> None:
|
|
from gfa_core.mas.stage_capabilities import CAPABILITIES
|
|
from gfa_core.mas.stage_catalog import STAGE_ORDER, is_stage_executable
|
|
from gfa_core.mas.tools.stage_tools import create_stage_registry
|
|
|
|
capability = CAPABILITIES["gene_family_report"]
|
|
stage_tool = create_stage_registry().get("gene_family_report")
|
|
|
|
assert "gene_family_report" in STAGE_ORDER
|
|
assert is_stage_executable("gene_family_report")
|
|
assert capability.consumes == []
|
|
assert stage_tool is not None
|
|
assert stage_tool.required_inputs == []
|
|
assert stage_tool.script == "run_gene_family_report.py"
|
|
|
|
|
|
def test_report_cli_can_start_as_a_script() -> None:
|
|
root = Path(__file__).resolve().parents[2]
|
|
completed = subprocess.run(
|
|
[sys.executable, "apps/cli/run_gene_family_report.py", "--help"],
|
|
cwd=root,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
|
|
assert completed.returncode == 0, completed.stderr
|
|
|
|
|
|
def test_report_cli_writes_report_and_manifest(tmp_path: Path) -> None:
|
|
from apps.cli.run_gene_family_report import main
|
|
|
|
workspace = tmp_path / "job"
|
|
_write_identify_manifest(workspace)
|
|
result = main(
|
|
[
|
|
"--workspace",
|
|
str(workspace),
|
|
"--prefix",
|
|
"Ft.SPL",
|
|
"--job-id",
|
|
"job-wiring",
|
|
"--expected-stages",
|
|
"identify",
|
|
"--stage-statuses-json",
|
|
'{"identify":"success"}',
|
|
"--no-model",
|
|
]
|
|
)
|
|
|
|
manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
|
|
assert result.report_path.is_file()
|
|
assert result.manifest_path.is_file()
|
|
assert "gene_family_report" in manifest["artifacts"]
|
|
|
|
|
|
def test_pipeline_backend_passes_plan_status_and_identity_to_report_cli(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
from apps.web import pipeline_backend
|
|
|
|
workspace = tmp_path / "job"
|
|
input_dir = workspace / "input"
|
|
logs_dir = workspace / "logs"
|
|
input_dir.mkdir(parents=True)
|
|
logs_dir.mkdir(parents=True)
|
|
_write_identify_manifest(workspace)
|
|
captured: dict[str, list[str]] = {}
|
|
|
|
def capture(command, _log_file, cwd=None):
|
|
captured["command"] = [str(value) for value in command]
|
|
return 0
|
|
|
|
monkeypatch.setattr(pipeline_backend, "can_start_new_task", lambda: (True, ""))
|
|
monkeypatch.setattr(pipeline_backend, "run_cmd_logged", capture)
|
|
job = pipeline_backend.JobInfo(
|
|
job_id="job-backend",
|
|
input_dir=str(input_dir),
|
|
workspace_dir=str(workspace),
|
|
logs_dir=str(logs_dir),
|
|
created_at="2026-08-08 00:00:00",
|
|
)
|
|
|
|
pipeline_backend.run_stage(
|
|
"gene_family_report",
|
|
job,
|
|
"Ft.SPL",
|
|
expected_stages=["identify", "motif"],
|
|
stage_statuses={"identify": "success", "motif": "failed"},
|
|
report_action="open",
|
|
report_use_model=False,
|
|
scope_id="scope-a",
|
|
analysis_route="post_subfamily_confirm",
|
|
member_set_id="members-a",
|
|
rnaseq_scope_id="rna-a",
|
|
)
|
|
|
|
command = captured["command"]
|
|
assert command[1:3] == ["apps/cli/run_gene_family_report.py", "--workspace"]
|
|
assert _flag_value(command, "--expected-stages") == "identify,motif"
|
|
assert json.loads(_flag_value(command, "--stage-statuses-json")) == {
|
|
"identify": "success",
|
|
"motif": "failed",
|
|
}
|
|
assert _flag_value(command, "--action") == "open"
|
|
assert _flag_value(command, "--job-id") == "job-backend"
|
|
assert _flag_value(command, "--scope-id") == "scope-a"
|
|
assert _flag_value(command, "--analysis-route") == "post_subfamily_confirm"
|
|
assert _flag_value(command, "--member-set-id") == "members-a"
|
|
assert _flag_value(command, "--rnaseq-scope-id") == "rna-a"
|
|
assert "--no-model" in command
|
|
assert not any("run_identify.py" in value or "run_motif.py" in value for value in command)
|
|
|
|
|
|
def test_pipeline_backend_rejects_empty_plan_before_starting_report_cli(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
from apps.web import pipeline_backend
|
|
|
|
workspace = tmp_path / "job"
|
|
input_dir = workspace / "input"
|
|
logs_dir = workspace / "logs"
|
|
input_dir.mkdir(parents=True)
|
|
logs_dir.mkdir(parents=True)
|
|
invoked = False
|
|
|
|
def capture(*_args, **_kwargs):
|
|
nonlocal invoked
|
|
invoked = True
|
|
return 0
|
|
|
|
monkeypatch.setattr(pipeline_backend, "can_start_new_task", lambda: (True, ""))
|
|
monkeypatch.setattr(pipeline_backend, "run_cmd_logged", capture)
|
|
job = pipeline_backend.JobInfo(
|
|
job_id="job-empty-plan",
|
|
input_dir=str(input_dir),
|
|
workspace_dir=str(workspace),
|
|
logs_dir=str(logs_dir),
|
|
created_at="2026-08-08 00:00:00",
|
|
)
|
|
|
|
message, _manifest = pipeline_backend.run_stage(
|
|
"gene_family_report",
|
|
job,
|
|
"Ft.SPL",
|
|
expected_stages=[],
|
|
report_action="generate",
|
|
report_use_model=False,
|
|
)
|
|
|
|
assert "expected_stages" in message
|
|
assert invoked is False
|
|
|
|
|
|
def test_stage_tool_executor_forwards_gene_family_report_context(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
import asyncio
|
|
import types
|
|
from dataclasses import dataclass
|
|
|
|
from gfa_core.mas.tools.stage_tools import StageTool, StageToolExecutor
|
|
|
|
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(kwargs["job"].workspace_dir) / "final" / "Ft.SPL.gene_family_report.manifest.json"
|
|
manifest.parent.mkdir(parents=True, exist_ok=True)
|
|
manifest.write_text(
|
|
json.dumps({"stage": "gene_family_report", "status": "success"}),
|
|
encoding="utf-8",
|
|
)
|
|
return "报告已生成", manifest
|
|
|
|
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)
|
|
|
|
result = asyncio.run(
|
|
StageToolExecutor(project_root=tmp_path).execute(
|
|
StageTool(
|
|
"gene_family_report",
|
|
"Gene Family Report",
|
|
"run_gene_family_report.py",
|
|
"report",
|
|
"test",
|
|
timeout=5,
|
|
),
|
|
"job-report",
|
|
"Ft.SPL",
|
|
{
|
|
"expectedStages": ["identify", "motif"],
|
|
"stageStatuses": {"identify": "success", "motif": "failed"},
|
|
"reportAction": "open",
|
|
"reportUseModel": False,
|
|
"scopeId": "scope-a",
|
|
"rnaseqScopeId": "rna-a",
|
|
},
|
|
)
|
|
)
|
|
|
|
assert result["status"] == "success"
|
|
assert captured["expected_stages"] == ["identify", "motif"]
|
|
assert captured["stage_statuses"] == {"identify": "success", "motif": "failed"}
|
|
assert captured["report_action"] == "open"
|
|
assert captured["report_use_model"] is False
|
|
assert captured["scope_id"] == "scope-a"
|
|
assert captured["rnaseq_scope_id"] == "rna-a"
|
|
|
|
|
|
def test_execute_report_does_not_invent_scope_identity(monkeypatch) -> None:
|
|
import asyncio
|
|
from unittest.mock import Mock
|
|
|
|
from apps.web.vue3 import main as web_main
|
|
from gfa_core.mas import resource_governor
|
|
from tasks import execute_single_stage_task
|
|
|
|
submitted = Mock()
|
|
submitted.return_value.id = "report-task"
|
|
monkeypatch.setattr(execute_single_stage_task, "apply_async", submitted)
|
|
monkeypatch.setattr(
|
|
web_main,
|
|
"_ensure_celery_queue_health",
|
|
lambda *_args, **_kwargs: ({"available": True}, None),
|
|
)
|
|
monkeypatch.setattr(
|
|
web_main,
|
|
"_ensure_gene_family_scope_file",
|
|
lambda **_kwargs: ("", ""),
|
|
)
|
|
monkeypatch.setattr(resource_governor, "record_task", lambda **_kwargs: {})
|
|
|
|
request = web_main.ExecuteRequest(
|
|
stage="gene_family_report",
|
|
job_id="job-report-no-invented-identity",
|
|
prefix="Ft.SPL",
|
|
expectedStages=["identify", "motif"],
|
|
stageStatuses={"identify": "success", "motif": "success"},
|
|
)
|
|
result = asyncio.run(web_main.execute_stage(request))
|
|
|
|
assert result["success"] is True
|
|
params = submitted.call_args.kwargs["kwargs"]["params"]
|
|
assert params["scopeId"] == ""
|
|
assert params["analysisRoute"] == ""
|
|
assert params["memberSetId"] == ""
|
|
assert params["rnaseqScopeId"] == ""
|
|
|
|
|
|
def test_report_artifact_validation_does_not_require_invented_identity(monkeypatch) -> None:
|
|
from types import SimpleNamespace
|
|
|
|
import tasks
|
|
from gfa_core.mas.tools import stage_identity
|
|
|
|
captured = {}
|
|
identity = SimpleNamespace(
|
|
scope_id="",
|
|
analysis_route="custom",
|
|
member_set_id="Ft.SPL:custom:all_members",
|
|
wgs_sample_set_id="",
|
|
wgs_reference_id="",
|
|
wgs_analysis_id="",
|
|
genotype_scope_id="",
|
|
phenotype_scope_id="",
|
|
covariate_scope_id="",
|
|
model_scope_id="",
|
|
source_model_scope_id="",
|
|
analysis_scope_id="",
|
|
)
|
|
monkeypatch.setattr(stage_identity, "resolve_stage_identity", lambda **_kwargs: identity)
|
|
|
|
def capture_status(**kwargs):
|
|
captured.update(kwargs)
|
|
return SimpleNamespace(completed=True)
|
|
|
|
monkeypatch.setattr(stage_identity, "get_stage_status_for_identity", capture_status)
|
|
|
|
failure = tasks.validate_single_stage_artifacts(
|
|
"job-report",
|
|
"Ft.SPL",
|
|
"gene_family_report",
|
|
{
|
|
"manifest": {
|
|
"stage": "gene_family_report",
|
|
"status": "success",
|
|
"prefix": "Ft.SPL",
|
|
"analysis_route": "",
|
|
"member_set_id": "",
|
|
}
|
|
},
|
|
)
|
|
|
|
assert failure is None
|
|
assert captured["strict_identity"] is False
|