449 lines
15 KiB
Python
449 lines
15 KiB
Python
from __future__ import annotations
|
|
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
from threading import Event
|
|
|
|
import pytest
|
|
|
|
from apps.cli.run_wgs_markdup import build_parser, main
|
|
from apps.cli import run_wgs_markdup
|
|
|
|
|
|
ALIGN_RUN = "3" * 32
|
|
|
|
|
|
def _record(workspace: Path, path: Path, role: str, sample_id: 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_id,
|
|
}
|
|
|
|
|
|
def _alignment_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"):
|
|
bam = bundle / f"Cohort.wgs_align.{sample}.sorted.bam"
|
|
bam.write_text(
|
|
"@HD\tVN:1.6\tSO:coordinate\n"
|
|
"@SQ\tSN:chr1\tLN:8\n"
|
|
f"@RG\tID:{sample}\tSM:{sample}\tPL:illumina\n"
|
|
f"{sample}.read\t0\tchr1\t1\t60\t4M\t*\t0\t0\tACGT\tIIII\n"
|
|
)
|
|
bai = Path(f"{bam}.bai")
|
|
bai.write_bytes(b"BAI")
|
|
artifacts[f"sorted_bam.{sample}"] = _record(workspace, bam, "sorted_bam", sample)
|
|
artifacts[f"sorted_bam_index.{sample}"] = _record(
|
|
workspace, bai, "sorted_bam_index", 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": "wgs-samples:" + "a" * 64,
|
|
"wgs_reference_id": "wgs-reference:" + "b" * 64,
|
|
"wgs_analysis_id": "wgs-analysis:" + "c" * 64,
|
|
"artifacts": artifacts,
|
|
}
|
|
authority = final / "Cohort.wgs_align.manifest.json"
|
|
authority.write_text(json.dumps(manifest))
|
|
return authority
|
|
|
|
|
|
def _fake_tools(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> tuple[Path, Path]:
|
|
tool_dir = tmp_path / "bin"
|
|
tool_dir.mkdir()
|
|
argv_log = tmp_path / "gatk.argv.jsonl"
|
|
gatk = tool_dir / "gatk"
|
|
gatk.write_text(
|
|
"""#!/usr/bin/env python3
|
|
import json, os, pathlib, sys
|
|
with open(os.environ['GATK_ARGV_LOG'], 'a') as out:
|
|
out.write(json.dumps(sys.argv[1:]) + '\\n')
|
|
if sys.argv[1:] == ['--version']:
|
|
print('The Genome Analysis Toolkit (GATK) v4.6.1.0')
|
|
raise SystemExit(0)
|
|
if not sys.argv[1:] or sys.argv[1] != 'MarkDuplicates':
|
|
raise SystemExit(2)
|
|
values = dict(item.split('=', 1) for item in sys.argv[2:] if '=' in item)
|
|
source = pathlib.Path(values['INPUT'])
|
|
output = pathlib.Path(values['OUTPUT'])
|
|
metrics = pathlib.Path(values['METRICS_FILE'])
|
|
sample = source.name.split('.')[2]
|
|
if os.environ.get('WGS_FAIL_SAMPLE') == sample:
|
|
print('controlled markdup failure', file=sys.stderr)
|
|
raise SystemExit(23)
|
|
data = source.read_text()
|
|
if os.environ.get('WGS_BAD_RG_SAMPLE') == sample:
|
|
data = data.replace(f'SM:{sample}', 'SM:Wrong')
|
|
output.write_text(data)
|
|
if os.environ.get('WGS_BAD_INDEX_SAMPLE') != sample:
|
|
pathlib.Path(f'{output}.bai').write_bytes(b'BAI')
|
|
if os.environ.get('WGS_BAD_METRICS_SAMPLE') == sample:
|
|
metrics.write_text('old or malformed metrics\\n')
|
|
else:
|
|
metrics.write_text(
|
|
'## METRICS CLASS\\tpicard.sam.DuplicationMetrics\\n'
|
|
'LIBRARY\\tUNPAIRED_READS_EXAMINED\\tREAD_PAIRS_EXAMINED\\tPERCENT_DUPLICATION\\n'
|
|
f'{sample}\\t0\\t1\\t0.0\\n'
|
|
)
|
|
print(f'completed {sample}')
|
|
"""
|
|
)
|
|
samtools = tool_dir / "samtools"
|
|
samtools.write_text(
|
|
"""#!/usr/bin/env python3
|
|
import json, os, pathlib, sys
|
|
with open(os.environ['SAMTOOLS_ARGV_LOG'], 'a') as out:
|
|
out.write(json.dumps(sys.argv[1:]) + '\\n')
|
|
if sys.argv[1:] == ['--version']:
|
|
print('samtools 1.17')
|
|
raise SystemExit(0)
|
|
cmd = sys.argv[1]
|
|
bam = pathlib.Path(sys.argv[-1])
|
|
sample = bam.name.split('.')[2]
|
|
if cmd == 'quickcheck':
|
|
raise SystemExit(9 if os.environ.get('WGS_BAD_QUICKCHECK_SAMPLE') == sample else 0)
|
|
if cmd == 'idxstats':
|
|
index = pathlib.Path(f'{bam}.bai')
|
|
if not index.is_file() or not index.stat().st_size:
|
|
raise SystemExit(4)
|
|
print('chr1\\t8\\t1\\t0')
|
|
raise SystemExit(0)
|
|
if cmd == 'view' and '-H' in sys.argv:
|
|
print('\\n'.join(line for line in bam.read_text().splitlines() if line.startswith('@')))
|
|
raise SystemExit(0)
|
|
raise SystemExit(2)
|
|
"""
|
|
)
|
|
gatk.chmod(0o755)
|
|
samtools.chmod(0o755)
|
|
monkeypatch.setenv("GATK_ARGV_LOG", str(argv_log))
|
|
monkeypatch.setenv("SAMTOOLS_ARGV_LOG", str(tmp_path / "samtools.argv.jsonl"))
|
|
monkeypatch.setenv("PATH", f"{tool_dir}:{Path('/usr/bin')}")
|
|
return gatk, argv_log
|
|
|
|
|
|
def _argv(
|
|
workspace: Path,
|
|
authority: Path,
|
|
gatk: Path,
|
|
samtools: Path | None = None,
|
|
) -> list[str]:
|
|
return [
|
|
"--workspace",
|
|
str(workspace),
|
|
"--job-id",
|
|
"job-1",
|
|
"--prefix",
|
|
"Cohort",
|
|
"--align-manifest",
|
|
str(authority),
|
|
"--gatk",
|
|
str(gatk),
|
|
"--samtools",
|
|
str(samtools or gatk.parent / "samtools"),
|
|
"--max-jobs",
|
|
"2",
|
|
"--remove-duplicates=true",
|
|
]
|
|
|
|
|
|
def test_wgs_markdup_exposes_import_safe_cli() -> None:
|
|
assert callable(build_parser)
|
|
assert callable(main)
|
|
|
|
|
|
def test_parser_requires_one_unambiguous_true_boolean() -> None:
|
|
parser = build_parser()
|
|
args = parser.parse_args(
|
|
[
|
|
"--workspace",
|
|
"w",
|
|
"--job-id",
|
|
"j",
|
|
"--prefix",
|
|
"p",
|
|
"--align-manifest",
|
|
"a",
|
|
"--gatk",
|
|
"g",
|
|
"--samtools",
|
|
"s",
|
|
"--max-jobs",
|
|
"1",
|
|
"--remove-duplicates=true",
|
|
]
|
|
)
|
|
assert args.remove_duplicates == "true"
|
|
with pytest.raises(SystemExit):
|
|
parser.parse_args(
|
|
[
|
|
"--workspace",
|
|
"w",
|
|
"--job-id",
|
|
"j",
|
|
"--prefix",
|
|
"p",
|
|
"--align-manifest",
|
|
"a",
|
|
"--gatk",
|
|
"g",
|
|
"--samtools",
|
|
"s",
|
|
"--max-jobs",
|
|
"1",
|
|
"--remove-duplicates=false",
|
|
]
|
|
)
|
|
|
|
|
|
def test_three_samples_publish_four_valid_roles_and_exact_gatk_argv(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
authority = _alignment_authority(workspace)
|
|
gatk, argv_log = _fake_tools(tmp_path, monkeypatch)
|
|
assert main(_argv(workspace, authority, gatk)) == 0
|
|
|
|
manifest = json.loads(
|
|
(workspace / "final" / "Cohort.wgs_markdup.manifest.json").read_text()
|
|
)
|
|
assert manifest["status"] == "success"
|
|
assert manifest["metrics"]["input_sample_ids"] == ["S1", "S2", "S3"]
|
|
assert manifest["metrics"]["output_sample_ids"] == ["S1", "S2", "S3"]
|
|
assert manifest["wgs_sample_set_id"] == "wgs-samples:" + "a" * 64
|
|
assert manifest["wgs_reference_id"] == "wgs-reference:" + "b" * 64
|
|
assert manifest["wgs_analysis_id"].startswith("wgs-analysis:")
|
|
assert manifest["parameters"]["samtools"] == str(gatk.parent / "samtools")
|
|
assert manifest["software"]["samtools"]["version"] == "samtools 1.17"
|
|
assert len(manifest["artifacts"]) == 12
|
|
assert {
|
|
item["role"] for item in manifest["artifacts"].values()
|
|
} == {"markdup_bam", "markdup_bam_index", "markdup_metrics", "markdup_log"}
|
|
assert {
|
|
item["sample_id"] for item in manifest["artifacts"].values()
|
|
} == {"S1", "S2", "S3"}
|
|
run_root = (
|
|
workspace
|
|
/ "final"
|
|
/ ".wgs_markdup_runs"
|
|
/ "Cohort"
|
|
/ manifest["run_id"]
|
|
)
|
|
assert {
|
|
Path(item["full_path"]).parent for item in manifest["artifacts"].values()
|
|
} == {run_root}
|
|
calls = [
|
|
value
|
|
for value in map(json.loads, argv_log.read_text().splitlines())
|
|
if value != ["--version"]
|
|
]
|
|
assert len(calls) == 3
|
|
for call in calls:
|
|
assert call[0] == "MarkDuplicates"
|
|
assert "REMOVE_DUPLICATES=true" in call
|
|
assert "CREATE_INDEX=true" in call
|
|
assert "VALIDATION_STRINGENCY=LENIENT" in call
|
|
assert all(not token.startswith((";", "|", "&&")) for token in call)
|
|
|
|
samtools_calls = [
|
|
value
|
|
for value in map(
|
|
json.loads,
|
|
(tmp_path / "samtools.argv.jsonl").read_text().splitlines(),
|
|
)
|
|
if value != ["--version"]
|
|
]
|
|
assert len(samtools_calls) == 12
|
|
|
|
|
|
def test_custom_samtools_path_and_version_affect_analysis_identity(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
authority = _alignment_authority(workspace)
|
|
gatk, _ = _fake_tools(tmp_path, monkeypatch)
|
|
samtools = gatk.parent / "samtools"
|
|
monkeypatch.setenv("PATH", "/usr/bin")
|
|
assert main(_argv(workspace, authority, gatk, samtools)) == 0
|
|
manifest_path = workspace / "final" / "Cohort.wgs_markdup.manifest.json"
|
|
first = json.loads(manifest_path.read_text())
|
|
|
|
samtools.write_text(
|
|
samtools.read_text().replace("samtools 1.17", "samtools 1.18")
|
|
)
|
|
assert main(_argv(workspace, authority, gatk, samtools)) == 0
|
|
second = json.loads(manifest_path.read_text())
|
|
assert first["wgs_analysis_id"] != second["wgs_analysis_id"]
|
|
assert second["software"]["samtools"] == {
|
|
"path": str(samtools),
|
|
"version": "samtools 1.18",
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("environment", "sample", "message"),
|
|
[
|
|
("WGS_FAIL_SAMPLE", "S2", "exit code 23"),
|
|
("WGS_BAD_QUICKCHECK_SAMPLE", "S1", "quickcheck"),
|
|
("WGS_BAD_INDEX_SAMPLE", "S3", "index"),
|
|
("WGS_BAD_METRICS_SAMPLE", "S2", "metrics"),
|
|
("WGS_BAD_RG_SAMPLE", "S1", "read group"),
|
|
],
|
|
)
|
|
def test_any_sample_or_postflight_failure_publishes_no_artifacts(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
capsys: pytest.CaptureFixture[str],
|
|
environment: str,
|
|
sample: str,
|
|
message: str,
|
|
) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
authority = _alignment_authority(workspace)
|
|
gatk, _ = _fake_tools(tmp_path, monkeypatch)
|
|
monkeypatch.setenv(environment, sample)
|
|
assert main(_argv(workspace, authority, gatk)) == 1
|
|
failure = json.loads(
|
|
(workspace / "final" / "Cohort.wgs_markdup.manifest.json").read_text()
|
|
)
|
|
assert failure["status"] == "failed"
|
|
assert failure["artifacts"] == {}
|
|
assert message in capsys.readouterr().err
|
|
|
|
|
|
def test_old_metrics_cannot_satisfy_failure_and_old_authority_survives(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
align = _alignment_authority(workspace)
|
|
final = workspace / "final"
|
|
old_bundle = final / ".wgs_markdup_runs" / "Cohort" / ("4" * 32)
|
|
old_bundle.mkdir(parents=True)
|
|
old_metrics = old_bundle / "Cohort.wgs_markdup.S2.markdup.metrics.txt"
|
|
old_metrics.write_text("valid old metrics must not count")
|
|
authority = final / "Cohort.wgs_markdup.manifest.json"
|
|
authority.write_text(
|
|
json.dumps(
|
|
{
|
|
"stage": "wgs_markdup",
|
|
"status": "success",
|
|
"start_time": "2026-01-01T00:00:00+00:00",
|
|
"artifacts": {"old": {"full_path": str(old_metrics)}},
|
|
}
|
|
)
|
|
)
|
|
authority_hash = hashlib.sha256(authority.read_bytes()).hexdigest()
|
|
metrics_hash = hashlib.sha256(old_metrics.read_bytes()).hexdigest()
|
|
gatk, _ = _fake_tools(tmp_path, monkeypatch)
|
|
monkeypatch.setenv("WGS_BAD_METRICS_SAMPLE", "S2")
|
|
assert main(_argv(workspace, align, gatk)) == 1
|
|
assert hashlib.sha256(authority.read_bytes()).hexdigest() == authority_hash
|
|
assert hashlib.sha256(old_metrics.read_bytes()).hexdigest() == metrics_hash
|
|
sidecars = list(final.glob("Cohort.wgs_markdup.failed.*.manifest.json"))
|
|
assert len(sidecars) == 1
|
|
assert json.loads(sidecars[0].read_text())["artifacts"] == {}
|
|
|
|
|
|
def test_concurrent_attempts_leave_one_complete_authority_bundle(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
align = _alignment_authority(workspace)
|
|
gatk, _ = _fake_tools(tmp_path, monkeypatch)
|
|
argv = _argv(workspace, align, gatk)
|
|
with ThreadPoolExecutor(max_workers=2) as pool:
|
|
results = [future.result() for future in (pool.submit(main, argv), pool.submit(main, argv))]
|
|
assert 0 in results
|
|
manifest = json.loads(
|
|
(workspace / "final" / "Cohort.wgs_markdup.manifest.json").read_text()
|
|
)
|
|
assert manifest["status"] == "success"
|
|
assert len(manifest["artifacts"]) == 12
|
|
assert {
|
|
Path(item["full_path"]).parent for item in manifest["artifacts"].values()
|
|
} == {
|
|
workspace
|
|
/ "final"
|
|
/ ".wgs_markdup_runs"
|
|
/ "Cohort"
|
|
/ manifest["run_id"]
|
|
}
|
|
|
|
|
|
def test_failing_attempt_holds_identity_lock_while_deciding_failure_destination(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
align = _alignment_authority(workspace)
|
|
gatk, _ = _fake_tools(tmp_path, monkeypatch)
|
|
authority = workspace / "final" / "Cohort.wgs_markdup.manifest.json"
|
|
failure_ready = Event()
|
|
release_failure = Event()
|
|
success_done = Event()
|
|
original_atomic = run_wgs_markdup._atomic_json
|
|
|
|
def delayed_failure_write(path: Path, payload: dict) -> None:
|
|
if payload.get("status") == "failed" and path == authority:
|
|
failure_ready.set()
|
|
assert release_failure.wait(5)
|
|
original_atomic(path, payload)
|
|
|
|
monkeypatch.setattr(run_wgs_markdup, "_atomic_json", delayed_failure_write)
|
|
monkeypatch.setenv("WGS_FAIL_SAMPLE", "S1")
|
|
argv = _argv(workspace, align, gatk)
|
|
|
|
def successful_attempt() -> int:
|
|
try:
|
|
return main(argv)
|
|
finally:
|
|
success_done.set()
|
|
|
|
with ThreadPoolExecutor(max_workers=2) as pool:
|
|
failing = pool.submit(main, argv)
|
|
assert failure_ready.wait(5)
|
|
monkeypatch.delenv("WGS_FAIL_SAMPLE")
|
|
successful = pool.submit(successful_attempt)
|
|
success_was_blocked = not success_done.wait(0.5)
|
|
release_failure.set()
|
|
assert failing.result() == 1
|
|
assert successful.result() == 0
|
|
assert success_was_blocked
|
|
assert json.loads(authority.read_text())["status"] == "success"
|
|
|
|
|
|
def test_authority_analysis_swap_between_header_and_resolver_fails_closed(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys
|
|
) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
align = _alignment_authority(workspace)
|
|
gatk, _ = _fake_tools(tmp_path, monkeypatch)
|
|
original_header = run_wgs_markdup._authority_header
|
|
|
|
def swap_after_header(*args, **kwargs):
|
|
header = original_header(*args, **kwargs)
|
|
swapped = json.loads(align.read_text())
|
|
swapped["wgs_analysis_id"] = "wgs-analysis:" + "d" * 64
|
|
align.write_text(json.dumps(swapped))
|
|
return header
|
|
|
|
monkeypatch.setattr(run_wgs_markdup, "_authority_header", swap_after_header)
|
|
assert main(_argv(workspace, align, gatk)) == 1
|
|
assert "analysis identity mismatch" in capsys.readouterr().err
|