312 lines
11 KiB
Python
312 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
|
|
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.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
|
|
ALIGN_ANALYSIS = "wgs-analysis:" + "c" * 64
|
|
MARKDUP_ANALYSIS = "wgs-analysis:" + "d" * 64
|
|
ALIGN_RUN = "3" * 32
|
|
MARKDUP_RUN = "4" * 32
|
|
|
|
|
|
def _record(workspace: Path, path: Path, role: str, sample: str) -> dict:
|
|
return {
|
|
"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(),
|
|
"sample_id": sample,
|
|
}
|
|
|
|
|
|
def _align_authority(workspace: Path) -> Path:
|
|
final = workspace / "final"
|
|
bundle = final / ".wgs_align_runs" / "Cohort" / ALIGN_RUN
|
|
bundle.mkdir(parents=True)
|
|
artifacts = {}
|
|
for sample in ("S1", "S2", "S3"):
|
|
for role, suffix in (
|
|
("sorted_bam", "sorted.bam"),
|
|
("sorted_bam_index", "sorted.bam.bai"),
|
|
):
|
|
path = bundle / f"Cohort.wgs_align.{sample}.{suffix}"
|
|
path.write_bytes(f"{sample}-{role}".encode())
|
|
artifacts[f"{role}.{sample}"] = _record(
|
|
workspace, path, role, sample
|
|
)
|
|
manifest = {
|
|
"manifest_version": "1.0",
|
|
"stage": "wgs_align",
|
|
"status": "success",
|
|
"prefix": "Cohort",
|
|
"job_id": "job-1",
|
|
"run_id": ALIGN_RUN,
|
|
"workspace": str(workspace.resolve()),
|
|
"wgs_sample_set_id": SAMPLES,
|
|
"wgs_reference_id": REFERENCE,
|
|
"wgs_analysis_id": ALIGN_ANALYSIS,
|
|
"artifacts": artifacts,
|
|
}
|
|
authority = final / "Cohort.wgs_align.manifest.json"
|
|
authority.write_text(json.dumps(manifest))
|
|
return authority
|
|
|
|
|
|
def _publish_result(workspace: Path, *, bad_hash: bool = False) -> Path:
|
|
final = workspace / "final"
|
|
bundle = final / ".wgs_markdup_runs" / "Cohort" / MARKDUP_RUN
|
|
bundle.mkdir(parents=True, exist_ok=True)
|
|
artifacts = {}
|
|
suffixes = {
|
|
"markdup_bam": "markdup.bam",
|
|
"markdup_bam_index": "markdup.bam.bai",
|
|
"markdup_metrics": "markdup.metrics.txt",
|
|
"markdup_log": "markdup.log",
|
|
}
|
|
for sample in ("S1", "S2", "S3"):
|
|
for role, suffix in suffixes.items():
|
|
path = bundle / f"Cohort.wgs_markdup.{sample}.{suffix}"
|
|
path.write_bytes(f"{sample}-{role}".encode())
|
|
artifacts[f"{role}.{sample}"] = _record(
|
|
workspace, path, role, sample
|
|
)
|
|
if bad_hash:
|
|
artifacts["markdup_bam.S2"]["sha256"] = "0" * 64
|
|
manifest = {
|
|
"manifest_version": "1.0",
|
|
"stage": "wgs_markdup",
|
|
"status": "success",
|
|
"prefix": "Cohort",
|
|
"job_id": "job-1",
|
|
"run_id": MARKDUP_RUN,
|
|
"start_time": "2026-07-23T00:00:00+00:00",
|
|
"workspace": str(workspace.resolve()),
|
|
"wgs_sample_set_id": SAMPLES,
|
|
"wgs_reference_id": REFERENCE,
|
|
"upstream_wgs_analysis_id": ALIGN_ANALYSIS,
|
|
"wgs_analysis_id": MARKDUP_ANALYSIS,
|
|
"metrics": {
|
|
"input_sample_ids": ["S1", "S2", "S3"],
|
|
"output_sample_ids": ["S1", "S2", "S3"],
|
|
},
|
|
"artifacts": artifacts,
|
|
}
|
|
authority = final / "Cohort.wgs_markdup.manifest.json"
|
|
authority.write_text(json.dumps(manifest))
|
|
return authority
|
|
|
|
|
|
def test_wgs_markdup_vertical_registration_is_exact() -> None:
|
|
assert STAGE_DEPENDENCIES["wgs_markdup"] == ["wgs_align"]
|
|
assert "wgs_markdup" in EXECUTABLE_STAGES
|
|
capability = CAPABILITIES["wgs_markdup"]
|
|
assert {item.role for item in capability.consumes} == {
|
|
"sorted_bam",
|
|
"sorted_bam_index",
|
|
}
|
|
assert {item.role for item in capability.provides} == {
|
|
"markdup_bam",
|
|
"markdup_bam_index",
|
|
"markdup_metrics",
|
|
"markdup_log",
|
|
}
|
|
definition = next(
|
|
item
|
|
for item in StageToolRegistry.STAGE_DEFINITIONS
|
|
if item["id"] == "wgs_markdup"
|
|
)
|
|
assert definition["script"] == "run_wgs_markdup.py"
|
|
assert PlanningAgentV2.REQUIRED_INPUTS["wgs_markdup"] == []
|
|
|
|
|
|
def test_backend_uses_exact_align_authority_command_and_deep_fresh_result(
|
|
monkeypatch, tmp_path: Path
|
|
) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
align = _align_authority(workspace)
|
|
output = workspace / "final" / "Cohort.wgs_markdup.manifest.json"
|
|
output.write_text('{"status":"old"}')
|
|
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, returned = pipeline_backend.run_stage(
|
|
stage="wgs_markdup",
|
|
job=pipeline_backend.JobInfo(
|
|
"job-1",
|
|
str(tmp_path / "input"),
|
|
str(workspace),
|
|
str(workspace / "logs"),
|
|
"now",
|
|
),
|
|
prefix="Cohort",
|
|
wgs_align_manifest=str(align),
|
|
wgs_sample_set_id=SAMPLES,
|
|
wgs_reference_id=REFERENCE,
|
|
wgs_analysis_id=ALIGN_ANALYSIS,
|
|
wgs_gatk="/opt/gatk",
|
|
wgs_samtools="/opt/samtools",
|
|
wgs_max_jobs=2,
|
|
)
|
|
assert "completed" in message.lower()
|
|
assert returned == output
|
|
command = calls[0]
|
|
assert command[:3] == [
|
|
pipeline_backend.sys.executable,
|
|
"-m",
|
|
"apps.cli.run_wgs_markdup",
|
|
]
|
|
assert command[command.index("--align-manifest") + 1] == str(align)
|
|
assert command[command.index("--gatk") + 1] == "/opt/gatk"
|
|
assert command[command.index("--samtools") + 1] == "/opt/samtools"
|
|
assert "--max-jobs" in command
|
|
assert "--remove-duplicates=true" in command
|
|
|
|
partial = json.loads(output.read_text())
|
|
partial["artifacts"].pop("markdup_metrics.S2")
|
|
assert not pipeline_backend._valid_wgs_markdup_manifest(
|
|
partial, job_id="job-1", prefix="Cohort"
|
|
)
|
|
|
|
|
|
def test_backend_rejects_missing_authority_and_tampered_fresh_hash(
|
|
monkeypatch, tmp_path: Path
|
|
) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
(workspace / "final").mkdir(parents=True)
|
|
message, returned = pipeline_backend.run_stage(
|
|
stage="wgs_markdup",
|
|
job=pipeline_backend.JobInfo(
|
|
"job-1", "", str(workspace), str(workspace / "logs"), "now"
|
|
),
|
|
prefix="Cohort",
|
|
)
|
|
assert "explicit" in message.lower()
|
|
assert returned is None
|
|
|
|
align = _align_authority(workspace)
|
|
monkeypatch.setattr(
|
|
pipeline_backend,
|
|
"run_cmd_logged",
|
|
lambda *_args, **_kwargs: (_publish_result(workspace, bad_hash=True) and 0),
|
|
)
|
|
message, returned = pipeline_backend.run_stage(
|
|
stage="wgs_markdup",
|
|
job=pipeline_backend.JobInfo(
|
|
"job-1", "", str(workspace), str(workspace / "logs"), "now"
|
|
),
|
|
prefix="Cohort",
|
|
wgs_align_manifest=str(align),
|
|
wgs_sample_set_id=SAMPLES,
|
|
wgs_reference_id=REFERENCE,
|
|
wgs_analysis_id=ALIGN_ANALYSIS,
|
|
)
|
|
assert "failed" in message.lower()
|
|
assert returned == workspace / "final" / "Cohort.wgs_markdup.manifest.json"
|
|
|
|
|
|
def test_api_worker_and_frontend_carry_stage04_authority_and_identities() -> None:
|
|
request = ExecuteRequest(
|
|
stage="wgs_markdup",
|
|
wgsAlignManifest="final/Cohort.wgs_align.manifest.json",
|
|
wgsSampleSetId=SAMPLES,
|
|
wgsReferenceId=REFERENCE,
|
|
wgsAnalysisId=ALIGN_ANALYSIS,
|
|
wgsGatk="gatk",
|
|
wgsMaxJobs=2,
|
|
)
|
|
assert request.wgsAlignManifest
|
|
root = Path(__file__).resolve().parents[2]
|
|
frontend = (root / "apps/web/vue3/index.html").read_text(encoding="utf-8")
|
|
for token in (
|
|
"wgs_markdup",
|
|
"wgsAlignManifest",
|
|
"wgsSampleSetId",
|
|
"wgsReferenceId",
|
|
"wgsAnalysisId",
|
|
):
|
|
assert token in frontend
|
|
worker = (root / "gfa_core/mas/tools/stage_tools.py").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
for token in (
|
|
"wgs_align_manifest=",
|
|
"wgs_sample_set_id=",
|
|
"wgs_reference_id=",
|
|
"wgs_analysis_id=",
|
|
):
|
|
assert token in worker
|
|
|
|
|
|
def test_frontend_builds_active_params_and_hands_align_identity_to_batch_markdup() -> None:
|
|
root = Path(__file__).resolve().parents[2]
|
|
text = (root / "apps/web/vue3/index.html").read_text(encoding="utf-8")
|
|
apply_start = text.index("const applyWgsExecutionParams =")
|
|
apply_end = text.index("const applyRnaseqExecutionParams =", apply_start)
|
|
handoff_start = text.index("const wgsMarkdupHandoffUpdates =")
|
|
handoff_end = text.index(
|
|
"const handoffWgsMarkdupProducerAuthority =", handoff_start
|
|
)
|
|
script = f"""
|
|
const firstDefinedParam = (...values) => values.find(v => v !== undefined && v !== null && v !== '');
|
|
const numberParam = (value, fallback) => Number(value || fallback);
|
|
const confirmParams = {{}};
|
|
{text[apply_start:apply_end]}
|
|
{text[handoff_start:handoff_end]}
|
|
const body = applyWgsExecutionParams({{}}, 'wgs_markdup', {{
|
|
wgsAlignManifest: '/workspace/job/final/Cohort.wgs_align.manifest.json',
|
|
wgsSampleSetId: '{SAMPLES}', wgsReferenceId: '{REFERENCE}',
|
|
wgsAnalysisId: '{ALIGN_ANALYSIS}', wgsGatk: '/opt/gatk',
|
|
wgsSamtools: '/opt/samtools-1.22/bin/samtools', wgsMaxJobs: 3
|
|
}});
|
|
const handoff = wgsMarkdupHandoffUpdates('wgs_align', {{
|
|
result: {{manifest_path: '/workspace/job/final/Cohort.wgs_align.manifest.json',
|
|
stage_status: {{completed: true, stage: 'wgs_align', prefix: 'Cohort',
|
|
wgs_sample_set_id: '{SAMPLES}', wgs_reference_id: '{REFERENCE}',
|
|
wgs_analysis_id: '{ALIGN_ANALYSIS}'}}}}
|
|
}}, 'Cohort');
|
|
const rejected = wgsMarkdupHandoffUpdates('wgs_align', {{
|
|
result: {{manifest_path: '/workspace/job/final/Other.wgs_align.manifest.json',
|
|
stage_status: {{completed: true, stage: 'wgs_align', prefix: 'Other',
|
|
wgs_sample_set_id: '{SAMPLES}', wgs_reference_id: '{REFERENCE}',
|
|
wgs_analysis_id: '{ALIGN_ANALYSIS}'}}}}
|
|
}}, 'Cohort');
|
|
console.log(JSON.stringify({{body, handoff, rejected}}));
|
|
"""
|
|
result = subprocess.run(
|
|
["/usr/bin/node", "-e", script], text=True, capture_output=True, check=True
|
|
)
|
|
payload = json.loads(result.stdout)
|
|
assert payload["body"]["wgsAlignManifest"].endswith(
|
|
"Cohort.wgs_align.manifest.json"
|
|
)
|
|
assert payload["body"]["wgsGatk"] == "/opt/gatk"
|
|
assert payload["body"]["wgsSamtools"] == "/opt/samtools-1.22/bin/samtools"
|
|
assert payload["body"]["wgsMaxJobs"] == 3
|
|
assert payload["handoff"] == {
|
|
"wgsAlignManifest": "/workspace/job/final/Cohort.wgs_align.manifest.json",
|
|
"wgsSampleSetId": SAMPLES,
|
|
"wgsReferenceId": REFERENCE,
|
|
"wgsAnalysisId": ALIGN_ANALYSIS,
|
|
}
|
|
assert payload["rejected"] is None
|