238 lines
8.8 KiB
Python
238 lines
8.8 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
|
|
from apps.web import pipeline_backend
|
|
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 import project_paths
|
|
from gfa_core.mas.stage_catalog import EXECUTABLE_STAGES, STAGE_DEPENDENCIES
|
|
from gfa_core.mas.tools.stage_tools import StageToolRegistry
|
|
|
|
|
|
SAMPLES = "wgs-samples:" + "a" * 64
|
|
REFERENCE = "wgs-reference:" + "b" * 64
|
|
MARKDUP_ANALYSIS = "wgs-analysis:" + "c" * 64
|
|
INDEX_ANALYSIS = "wgs-analysis:" + "d" * 64
|
|
|
|
|
|
def _record(workspace: Path, path: Path, role: str, sample: str = "") -> dict:
|
|
value = {
|
|
"artifact": role,
|
|
"role": role,
|
|
"path": str(path.relative_to(workspace)),
|
|
"full_path": str(path.resolve()),
|
|
"size": path.stat().st_size,
|
|
"size_bytes": path.stat().st_size,
|
|
"sha256": hashlib.sha256(path.read_bytes()).hexdigest(),
|
|
}
|
|
if sample:
|
|
value["sample_id"] = sample
|
|
return value
|
|
|
|
|
|
def _authorities(workspace: Path) -> tuple[Path, Path]:
|
|
final = workspace / "final"
|
|
markdup_bundle = final / ".wgs_markdup_runs" / "Cohort" / ("4" * 32)
|
|
markdup_bundle.mkdir(parents=True)
|
|
artifacts = {}
|
|
for sample in ("S1", "S2", "S3"):
|
|
for role, suffix in (
|
|
("markdup_bam", "markdup.bam"),
|
|
("markdup_bam_index", "markdup.bam.bai"),
|
|
):
|
|
path = markdup_bundle / f"Cohort.wgs_markdup.{sample}.{suffix}"
|
|
path.write_bytes(f"{sample}-{role}".encode())
|
|
artifacts[f"{role}.{sample}"] = _record(
|
|
workspace, path, role, sample
|
|
)
|
|
markdup = final / "Cohort.wgs_markdup.manifest.json"
|
|
markdup.write_text(json.dumps({
|
|
"manifest_version": "1.0", "stage": "wgs_markdup", "status": "success",
|
|
"prefix": "Cohort", "job_id": "job-1", "run_id": "4" * 32,
|
|
"workspace": str(workspace.resolve()), "wgs_sample_set_id": SAMPLES,
|
|
"wgs_reference_id": REFERENCE, "wgs_analysis_id": MARKDUP_ANALYSIS,
|
|
"artifacts": artifacts,
|
|
}))
|
|
index_bundle = final / ".wgs_index_runs" / "Cohort" / ("2" * 32)
|
|
index_bundle.mkdir(parents=True)
|
|
fai = index_bundle / "Cohort.wgs_index.reference.fasta.fai"
|
|
fai.write_text("chr1\t10\t0\t0\t0\n")
|
|
index = final / "Cohort.wgs_index.manifest.json"
|
|
index.write_text(json.dumps({
|
|
"manifest_version": "1.0", "stage": "wgs_index", "status": "success",
|
|
"prefix": "Cohort", "job_id": "job-1", "run_id": "2" * 32,
|
|
"workspace": str(workspace.resolve()), "wgs_reference_id": REFERENCE,
|
|
"wgs_analysis_id": INDEX_ANALYSIS,
|
|
"artifacts": {"reference_fai": _record(workspace, fai, "reference_fai")},
|
|
}))
|
|
return markdup, index
|
|
|
|
|
|
def _publish_result(workspace: Path) -> Path:
|
|
final = workspace / "final"
|
|
bundle = final / ".wgs_bam_qc_runs" / "Cohort" / ("5" * 32)
|
|
bundle.mkdir(parents=True)
|
|
artifacts = {}
|
|
for sample in ("S1", "S2", "S3"):
|
|
for role, suffix in (
|
|
("bam_flagstat", "flagstat.txt"),
|
|
("bam_depth_summary", "depth.tsv"),
|
|
):
|
|
path = bundle / f"Cohort.wgs_bam_qc.{sample}.{suffix}"
|
|
path.write_text(f"{sample}-{role}\n")
|
|
artifacts[f"{role}.{sample}"] = _record(
|
|
workspace, path, role, sample
|
|
)
|
|
for role, suffix in (
|
|
("bam_qc_summary", "cohort.summary.tsv"),
|
|
("bam_qc_report", "cohort.report.md"),
|
|
):
|
|
path = bundle / f"Cohort.wgs_bam_qc.{suffix}"
|
|
path.write_text(f"{role}\n")
|
|
artifacts[role] = _record(workspace, path, role)
|
|
output = final / "Cohort.wgs_bam_qc.manifest.json"
|
|
output.write_text(json.dumps({
|
|
"manifest_version": "1.0", "stage": "wgs_bam_qc", "status": "success",
|
|
"prefix": "Cohort", "job_id": "job-1", "run_id": "5" * 32,
|
|
"workspace": str(workspace.resolve()), "wgs_sample_set_id": SAMPLES,
|
|
"wgs_reference_id": REFERENCE,
|
|
"upstream_wgs_analysis_ids": {
|
|
"wgs_markdup": MARKDUP_ANALYSIS, "wgs_index": INDEX_ANALYSIS,
|
|
},
|
|
"wgs_analysis_id": "wgs-analysis:" + "e" * 64,
|
|
"parameters": {"metric_name": "coverage_ge_1x"},
|
|
"metrics": {
|
|
"input_sample_ids": ["S1", "S2", "S3"],
|
|
"output_sample_ids": ["S1", "S2", "S3"],
|
|
"total_reference_bases": 10, "coverage_ge_1x": 0.2,
|
|
},
|
|
"artifacts": artifacts,
|
|
}))
|
|
return output
|
|
|
|
|
|
def test_wgs_bam_qc_vertical_registration_is_exact() -> None:
|
|
assert STAGE_DEPENDENCIES["wgs_bam_qc"] == ["wgs_markdup", "wgs_index"]
|
|
assert "wgs_bam_qc" in EXECUTABLE_STAGES
|
|
capability = CAPABILITIES["wgs_bam_qc"]
|
|
assert {item.role for item in capability.consumes} == {
|
|
"markdup_bam", "markdup_bam_index", "reference_fai"
|
|
}
|
|
assert {item.role for item in capability.provides} == {
|
|
"bam_flagstat", "bam_depth_summary", "bam_qc_summary", "bam_qc_report"
|
|
}
|
|
definition = next(
|
|
item
|
|
for item in StageToolRegistry.STAGE_DEFINITIONS
|
|
if item["id"] == "wgs_bam_qc"
|
|
)
|
|
assert definition["script"] == "run_wgs_bam_qc.py"
|
|
assert definition["conda_env"] == "rnaseq"
|
|
assert capability.env == "rnaseq"
|
|
assert definition["required_inputs"] == [
|
|
"wgs_markdup_manifest", "wgs_index_manifest"
|
|
]
|
|
assert PlanningAgentV2.REQUIRED_INPUTS["wgs_bam_qc"] == []
|
|
|
|
|
|
def test_wgs_bam_qc_resolves_samtools_from_declared_rnaseq_environment() -> None:
|
|
resolver = getattr(project_paths, "get_wgs_samtools", None)
|
|
assert resolver is not None
|
|
assert resolver("wgs_bam_qc", "").endswith(
|
|
"/envs/rnaseq/bin/samtools"
|
|
)
|
|
explicit = "/opt/asaw/bin/samtools"
|
|
assert resolver("wgs_align", explicit) == explicit
|
|
|
|
|
|
def test_backend_uses_both_authorities_and_exact_canonical_command(
|
|
monkeypatch, tmp_path: Path
|
|
) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
markdup, index = _authorities(workspace)
|
|
calls = []
|
|
|
|
def fake_run(command, *_args, **_kwargs):
|
|
calls.append(command)
|
|
_publish_result(workspace)
|
|
return 0
|
|
|
|
monkeypatch.setattr(pipeline_backend, "run_cmd_logged", fake_run)
|
|
message, output = pipeline_backend.run_stage(
|
|
stage="wgs_bam_qc",
|
|
job=pipeline_backend.JobInfo(
|
|
"job-1", "", str(workspace), str(workspace / "logs"), "now"
|
|
),
|
|
prefix="Cohort",
|
|
wgs_markdup_manifest=str(markdup),
|
|
wgs_index_manifest=str(index),
|
|
wgs_sample_set_id=SAMPLES,
|
|
wgs_reference_id=REFERENCE,
|
|
wgs_analysis_id=MARKDUP_ANALYSIS,
|
|
wgs_index_analysis_id=INDEX_ANALYSIS,
|
|
wgs_samtools="/opt/samtools",
|
|
wgs_coverage_threshold=1,
|
|
wgs_threads=2,
|
|
wgs_max_jobs=3,
|
|
)
|
|
assert "completed" in message.lower()
|
|
assert output == workspace / "final" / "Cohort.wgs_bam_qc.manifest.json"
|
|
command = calls[0]
|
|
assert command[:3] == [
|
|
pipeline_backend.sys.executable, "-m", "apps.cli.run_wgs_bam_qc"
|
|
]
|
|
assert command[command.index("--markdup-manifest") + 1] == str(markdup)
|
|
assert command[command.index("--index-manifest") + 1] == str(index)
|
|
assert command[command.index("--coverage-threshold") + 1] == "1"
|
|
assert command[command.index("--threads") + 1] == "2"
|
|
assert command[command.index("--max-jobs") + 1] == "3"
|
|
|
|
|
|
def test_api_contract_carries_stage05_authorities_and_identity_pair() -> None:
|
|
request = ExecuteRequest(
|
|
stage="wgs_bam_qc",
|
|
wgsMarkdupManifest="final/Cohort.wgs_markdup.manifest.json",
|
|
wgsIndexManifest="final/Cohort.wgs_index.manifest.json",
|
|
wgsSampleSetId=SAMPLES,
|
|
wgsReferenceId=REFERENCE,
|
|
wgsAnalysisId=MARKDUP_ANALYSIS,
|
|
wgsIndexAnalysisId=INDEX_ANALYSIS,
|
|
wgsCoverageThreshold=1,
|
|
)
|
|
assert request.wgsMarkdupManifest
|
|
assert request.wgsIndexAnalysisId == INDEX_ANALYSIS
|
|
|
|
|
|
def test_planning_status_keeps_completed_bam_qc_after_upstreams_are_deleted(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
from gfa_core.mas.tools.stage_status import get_stage_status
|
|
|
|
workspace_root = tmp_path / "workspace"
|
|
workspace = workspace_root / "job-1"
|
|
markdup, index = _authorities(workspace)
|
|
_publish_result(workspace)
|
|
markdup.unlink()
|
|
index.unlink()
|
|
|
|
strict_status = get_stage_status(
|
|
"job-1", "Cohort", "wgs_bam_qc", workspace_root,
|
|
wgs_sample_set_id=SAMPLES,
|
|
wgs_reference_id=REFERENCE,
|
|
wgs_analysis_id="wgs-analysis:" + "e" * 64,
|
|
)
|
|
planning_status = get_stage_status(
|
|
"job-1", "Cohort", "wgs_bam_qc", workspace_root,
|
|
wgs_sample_set_id=SAMPLES,
|
|
wgs_reference_id=REFERENCE,
|
|
wgs_analysis_id="wgs-analysis:" + "e" * 64,
|
|
validate_wgs_upstream_authority=False,
|
|
)
|
|
|
|
assert strict_status.completed is False
|
|
assert planning_status.completed is True
|