666 lines
22 KiB
Python
666 lines
22 KiB
Python
from __future__ import annotations
|
|
|
|
import gzip
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
from threading import Barrier, Thread
|
|
|
|
import pytest
|
|
|
|
|
|
SAMPLES = "wgs-samples:" + "a" * 64
|
|
REFERENCE = "wgs-reference:" + "b" * 64
|
|
HAPLOTYPE_ANALYSIS = "wgs-analysis:" + "c" * 64
|
|
INDEX_ANALYSIS = "wgs-analysis:" + "d" * 64
|
|
|
|
|
|
def _record(
|
|
workspace: Path, path: Path, role: str, sample_id: str = ""
|
|
) -> dict:
|
|
item = {
|
|
"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_id:
|
|
item["sample_id"] = sample_id
|
|
return item
|
|
|
|
|
|
def _write_gvcf(path: Path, samples: list[str]) -> None:
|
|
with gzip.open(path, "wt", encoding="utf-8") as handle:
|
|
handle.write("##fileformat=VCFv4.2\n")
|
|
handle.write(
|
|
'##ALT=<ID=NON_REF,Description="Represents any possible '
|
|
'alternative allele">\n'
|
|
)
|
|
handle.write(
|
|
"#CHROM\tPOS\tID\tREF\tALT\tQUAL\tFILTER\tINFO\tFORMAT\t"
|
|
+ "\t".join(samples)
|
|
+ "\n"
|
|
)
|
|
handle.write(
|
|
"chr1\t1\t.\tA\t<NON_REF>\t.\t.\tEND=10\tGT\t"
|
|
+ "\t".join("0/0" for _ in samples)
|
|
+ "\n"
|
|
)
|
|
|
|
|
|
def _authorities(workspace: Path) -> tuple[Path, Path]:
|
|
final = workspace / "final"
|
|
haplotype_bundle = (
|
|
final / ".wgs_haplotype_calling_runs" / "Cohort" / ("6" * 32)
|
|
)
|
|
haplotype_bundle.mkdir(parents=True)
|
|
artifacts = {}
|
|
for sample in ("S1", "S2", "S3"):
|
|
gvcf = haplotype_bundle / (
|
|
f"Cohort.wgs_haplotype_calling.{sample}.g.vcf.gz"
|
|
)
|
|
tbi = Path(f"{gvcf}.tbi")
|
|
log = haplotype_bundle / (
|
|
f"Cohort.wgs_haplotype_calling.{sample}.haplotype.log"
|
|
)
|
|
_write_gvcf(gvcf, [sample])
|
|
tbi.write_bytes(f"TBI:{sample}".encode())
|
|
log.write_text(f"sample={sample}\n", encoding="utf-8")
|
|
artifacts[f"gvcf.{sample}"] = _record(
|
|
workspace, gvcf, "gvcf", sample
|
|
)
|
|
artifacts[f"gvcf_index.{sample}"] = _record(
|
|
workspace, tbi, "gvcf_index", sample
|
|
)
|
|
artifacts[f"haplotype_log.{sample}"] = _record(
|
|
workspace, log, "haplotype_log", sample
|
|
)
|
|
haplotype = final / "Cohort.wgs_haplotype_calling.manifest.json"
|
|
haplotype.write_text(
|
|
json.dumps(
|
|
{
|
|
"manifest_version": "1.0",
|
|
"stage": "wgs_haplotype_calling",
|
|
"status": "success",
|
|
"prefix": "Cohort",
|
|
"job_id": "job-1",
|
|
"run_id": "6" * 32,
|
|
"workspace": str(workspace.resolve()),
|
|
"wgs_sample_set_id": SAMPLES,
|
|
"wgs_reference_id": REFERENCE,
|
|
"wgs_analysis_id": HAPLOTYPE_ANALYSIS,
|
|
"upstream_wgs_analysis_ids": {
|
|
"wgs_markdup": "wgs-analysis:" + "e" * 64,
|
|
"wgs_bam_qc": "wgs-analysis:" + "f" * 64,
|
|
"wgs_index": INDEX_ANALYSIS,
|
|
},
|
|
"metrics": {
|
|
"input_sample_ids": ["S1", "S2", "S3"],
|
|
"output_sample_ids": ["S1", "S2", "S3"],
|
|
},
|
|
"artifacts": artifacts,
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
index_bundle = final / ".wgs_index_runs" / "Cohort" / ("2" * 32)
|
|
index_bundle.mkdir(parents=True)
|
|
reference = index_bundle / "Cohort.wgs_index.reference.fasta"
|
|
fai = Path(f"{reference}.fai")
|
|
dictionary = reference.with_suffix(".dict")
|
|
reference.write_text(">chr1\nAAAAAAAAAA\n", encoding="utf-8")
|
|
fai.write_text("chr1\t10\t6\t10\t11\n", encoding="utf-8")
|
|
dictionary.write_text("@HD\tVN:1.6\n@SQ\tSN:chr1\tLN:10\n", encoding="utf-8")
|
|
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_fasta": _record(
|
|
workspace, reference, "reference_fasta"
|
|
),
|
|
"reference_fai": _record(
|
|
workspace, fai, "reference_fai"
|
|
),
|
|
"reference_dict": _record(
|
|
workspace, dictionary, "reference_dict"
|
|
),
|
|
},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
return haplotype, index
|
|
|
|
|
|
def test_stage07_rejects_producer_from_different_index_generation_at_all_gates(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
from apps.cli.run_wgs_gvcf_combine import main
|
|
from apps.web import pipeline_backend
|
|
from gfa_core.mas.tools.stage_status import get_stage_status
|
|
|
|
workspace_root = tmp_path / "workspaces"
|
|
workspace = workspace_root / "job-1"
|
|
haplotype, index = _authorities(workspace)
|
|
gatk = _gatk(tmp_path)
|
|
assert main(_argv(workspace, haplotype, index, gatk)) == 0
|
|
output = workspace / "final" / "Cohort.wgs_gvcf_combine.manifest.json"
|
|
output_analysis = json.loads(output.read_text())["wgs_analysis_id"]
|
|
|
|
producer = json.loads(haplotype.read_text())
|
|
producer["upstream_wgs_analysis_ids"]["wgs_index"] = (
|
|
"wgs-analysis:" + "0" * 64
|
|
)
|
|
haplotype.write_text(json.dumps(producer))
|
|
assert main(_argv(workspace, haplotype, index, gatk)) == 1
|
|
|
|
called = {"value": False}
|
|
def forbidden_run(*_args, **_kwargs):
|
|
called["value"] = True
|
|
return 0
|
|
monkeypatch.setattr(pipeline_backend, "run_cmd_logged", forbidden_run)
|
|
message, _ = pipeline_backend.run_stage(
|
|
stage="wgs_gvcf_combine",
|
|
job=pipeline_backend.JobInfo(
|
|
"job-1", "", str(workspace), str(workspace / "logs"), "now"
|
|
),
|
|
prefix="Cohort",
|
|
wgs_haplotype_manifest=str(haplotype),
|
|
wgs_index_manifest=str(index),
|
|
wgs_sample_set_id=SAMPLES,
|
|
wgs_reference_id=REFERENCE,
|
|
wgs_haplotype_analysis_id=HAPLOTYPE_ANALYSIS,
|
|
wgs_index_analysis_id=INDEX_ANALYSIS,
|
|
wgs_gatk=str(gatk),
|
|
)
|
|
assert "lineage" in message.lower()
|
|
assert not called["value"]
|
|
assert not get_stage_status(
|
|
"job-1", "Cohort", "wgs_gvcf_combine", workspace_root,
|
|
wgs_sample_set_id=SAMPLES,
|
|
wgs_reference_id=REFERENCE,
|
|
wgs_analysis_id=output_analysis,
|
|
).completed
|
|
|
|
|
|
def _gatk(tmp_path: Path) -> Path:
|
|
tool = tmp_path / "gatk"
|
|
tool.write_text(
|
|
"""#!/usr/bin/env python3
|
|
import gzip, json, os, pathlib, sys
|
|
args = sys.argv[1:]
|
|
if args == ['--version']:
|
|
print('The Genome Analysis Toolkit (GATK) v4.6.0.0')
|
|
raise SystemExit(0)
|
|
command = args[0]
|
|
def value(flag):
|
|
return args[args.index(flag) + 1]
|
|
calls = os.environ.get('GATK_CALLS')
|
|
if calls:
|
|
with open(calls, 'a', encoding='utf-8') as handle:
|
|
handle.write(json.dumps(args) + '\\n')
|
|
if os.environ.get('FAIL_COMMAND') == command:
|
|
raise SystemExit(23)
|
|
if command == 'ValidateVariants':
|
|
path = pathlib.Path(value('-V'))
|
|
index = pathlib.Path(str(path) + '.tbi')
|
|
if not path.is_file() or not index.is_file():
|
|
raise SystemExit(2)
|
|
if index.read_bytes().startswith(b'CORRUPT'):
|
|
raise SystemExit(24)
|
|
raise SystemExit(0)
|
|
if command == 'CombineGVCFs':
|
|
inputs = [
|
|
pathlib.Path(args[pos + 1])
|
|
for pos, item in enumerate(args) if item == '-V'
|
|
]
|
|
if len(inputs) != 3:
|
|
raise SystemExit(25)
|
|
samples = []
|
|
for source in inputs:
|
|
with gzip.open(source, 'rt') as handle:
|
|
header = next(
|
|
line for line in handle if line.startswith('#CHROM')
|
|
)
|
|
samples.extend(header.rstrip().split('\\t')[9:])
|
|
if os.environ.get('BAD_OUTPUT_HEADER'):
|
|
samples[-1] = 'WRONG'
|
|
output = pathlib.Path(value('-O'))
|
|
with gzip.open(output, 'wt') as handle:
|
|
handle.write('##fileformat=VCFv4.2\\n')
|
|
handle.write('##ALT=<ID=NON_REF,Description="NON_REF">\\n')
|
|
handle.write('#CHROM\\tPOS\\tID\\tREF\\tALT\\tQUAL\\tFILTER\\tINFO'
|
|
'\\tFORMAT\\t' + '\\t'.join(samples) + '\\n')
|
|
handle.write('chr1\\t1\\t.\\tA\\t<NON_REF>\\t.\\t.\\tEND=10'
|
|
'\\tGT\\t' + '\\t'.join('0/0' for _ in samples) + '\\n')
|
|
raise SystemExit(0)
|
|
if command == 'IndexFeatureFile':
|
|
path = pathlib.Path(value('-I'))
|
|
content = (
|
|
b'CORRUPT:output' if os.environ.get('CORRUPT_OUTPUT_INDEX')
|
|
else b'TBI:combined'
|
|
)
|
|
pathlib.Path(str(path) + '.tbi').write_bytes(content)
|
|
raise SystemExit(0)
|
|
raise SystemExit(2)
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
tool.chmod(0o755)
|
|
return tool
|
|
|
|
|
|
def _argv(
|
|
workspace: Path, haplotype: Path, index: Path, gatk: Path
|
|
) -> list[str]:
|
|
return [
|
|
"--workspace",
|
|
str(workspace),
|
|
"--job-id",
|
|
"job-1",
|
|
"--prefix",
|
|
"Cohort",
|
|
"--haplotype-manifest",
|
|
str(haplotype),
|
|
"--index-manifest",
|
|
str(index),
|
|
"--gatk",
|
|
str(gatk),
|
|
]
|
|
|
|
|
|
def test_wgs_gvcf_combine_exposes_import_safe_cli() -> None:
|
|
from apps.cli.run_wgs_gvcf_combine import build_parser, main
|
|
|
|
assert callable(build_parser)
|
|
assert callable(main)
|
|
|
|
|
|
def test_three_samples_produce_singleton_canonical_outputs_and_exact_header(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
from apps.cli.run_wgs_gvcf_combine import main
|
|
|
|
workspace = tmp_path / "workspace"
|
|
haplotype, index = _authorities(workspace)
|
|
calls = tmp_path / "gatk.calls"
|
|
monkeypatch.setenv("GATK_CALLS", str(calls))
|
|
assert main(_argv(workspace, haplotype, index, _gatk(tmp_path))) == 0
|
|
authority = workspace / "final" / "Cohort.wgs_gvcf_combine.manifest.json"
|
|
manifest = json.loads(authority.read_text())
|
|
assert manifest["metrics"]["input_sample_ids"] == ["S1", "S2", "S3"]
|
|
assert manifest["metrics"]["output_sample_ids"] == ["S1", "S2", "S3"]
|
|
assert manifest["upstream_wgs_analysis_ids"] == {
|
|
"wgs_haplotype_calling": HAPLOTYPE_ANALYSIS,
|
|
"wgs_index": INDEX_ANALYSIS,
|
|
}
|
|
assert set(manifest["artifacts"]) == {
|
|
"combined_gvcf",
|
|
"combined_gvcf_index",
|
|
"combine_log",
|
|
}
|
|
records = {
|
|
item["role"]: Path(item["full_path"])
|
|
for item in manifest["artifacts"].values()
|
|
}
|
|
assert records["combined_gvcf"].name == (
|
|
"Cohort.wgs_gvcf_combine.cohort.combined.g.vcf.gz"
|
|
)
|
|
assert records["combined_gvcf_index"] == Path(
|
|
f"{records['combined_gvcf']}.tbi"
|
|
)
|
|
with gzip.open(records["combined_gvcf"], "rt") as handle:
|
|
header = next(line for line in handle if line.startswith("#CHROM"))
|
|
assert header.rstrip().split("\t")[9:] == ["S1", "S2", "S3"]
|
|
assert records["combined_gvcf_index"].read_bytes() == b"TBI:combined"
|
|
log = records["combine_log"].read_text()
|
|
assert "CombineGVCFs exit=0" in log
|
|
assert "IndexFeatureFile exit=0" in log
|
|
assert "ValidateVariants exit=0" in log
|
|
invocations = [json.loads(line) for line in calls.read_text().splitlines()]
|
|
validate_inputs = [
|
|
call[call.index("-V") + 1]
|
|
for call in invocations
|
|
if call[0] == "ValidateVariants" and "-L" in call
|
|
and ".wgs_haplotype_calling_runs" in call[call.index("-V") + 1]
|
|
]
|
|
assert len(validate_inputs) == 3
|
|
combine = next(call for call in invocations if call[0] == "CombineGVCFs")
|
|
assert combine.count("-V") == 3
|
|
|
|
|
|
@pytest.mark.parametrize("mutation", ("missing", "duplicate"))
|
|
def test_missing_or_duplicate_declared_gvcf_is_rejected(
|
|
tmp_path: Path, mutation: str
|
|
) -> None:
|
|
from apps.cli.run_wgs_gvcf_combine import main
|
|
|
|
workspace = tmp_path / "workspace"
|
|
haplotype, index = _authorities(workspace)
|
|
payload = json.loads(haplotype.read_text())
|
|
if mutation == "missing":
|
|
payload["artifacts"].pop("gvcf.S2")
|
|
else:
|
|
payload["artifacts"]["gvcf.duplicate"] = dict(
|
|
payload["artifacts"]["gvcf.S2"]
|
|
)
|
|
haplotype.write_text(json.dumps(payload))
|
|
assert main(_argv(workspace, haplotype, index, _gatk(tmp_path))) == 1
|
|
|
|
|
|
def test_corrupt_nonempty_input_tbi_is_rejected_by_real_query(
|
|
tmp_path: Path
|
|
) -> None:
|
|
from apps.cli.run_wgs_gvcf_combine import main
|
|
|
|
workspace = tmp_path / "workspace"
|
|
haplotype, index = _authorities(workspace)
|
|
payload = json.loads(haplotype.read_text())
|
|
record = payload["artifacts"]["gvcf_index.S2"]
|
|
tbi = Path(record["full_path"])
|
|
tbi.write_bytes(b"CORRUPT:input")
|
|
record["size"] = record["size_bytes"] = tbi.stat().st_size
|
|
record["sha256"] = hashlib.sha256(tbi.read_bytes()).hexdigest()
|
|
haplotype.write_text(json.dumps(payload))
|
|
assert main(_argv(workspace, haplotype, index, _gatk(tmp_path))) == 1
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"failure", ("CombineGVCFs", "IndexFeatureFile", "ValidateVariants")
|
|
)
|
|
def test_each_gatk_return_code_fails_and_preserves_old_authority(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
failure: str,
|
|
) -> None:
|
|
from apps.cli.run_wgs_gvcf_combine import main
|
|
|
|
workspace = tmp_path / "workspace"
|
|
haplotype, index = _authorities(workspace)
|
|
authority = workspace / "final" / "Cohort.wgs_gvcf_combine.manifest.json"
|
|
old = workspace / "final" / "old.combined.g.vcf.gz"
|
|
old.write_bytes(b"old-success")
|
|
authority.write_text(
|
|
json.dumps(
|
|
{
|
|
"stage": "wgs_gvcf_combine",
|
|
"status": "success",
|
|
"sentinel": True,
|
|
}
|
|
)
|
|
)
|
|
before = (
|
|
hashlib.sha256(authority.read_bytes()).hexdigest(),
|
|
hashlib.sha256(old.read_bytes()).hexdigest(),
|
|
)
|
|
monkeypatch.setenv("FAIL_COMMAND", failure)
|
|
assert main(_argv(workspace, haplotype, index, _gatk(tmp_path))) == 1
|
|
after = (
|
|
hashlib.sha256(authority.read_bytes()).hexdigest(),
|
|
hashlib.sha256(old.read_bytes()).hexdigest(),
|
|
)
|
|
assert after == before
|
|
sidecars = list(
|
|
(workspace / "final").glob(
|
|
"Cohort.wgs_gvcf_combine.failed.*.manifest.json"
|
|
)
|
|
)
|
|
assert len(sidecars) == 1
|
|
assert json.loads(sidecars[0].read_text())["artifacts"] == {}
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"environment", ("BAD_OUTPUT_HEADER", "CORRUPT_OUTPUT_INDEX")
|
|
)
|
|
def test_output_header_or_index_postflight_failure_is_not_published(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
environment: str,
|
|
) -> None:
|
|
from apps.cli.run_wgs_gvcf_combine import main
|
|
|
|
workspace = tmp_path / "workspace"
|
|
haplotype, index = _authorities(workspace)
|
|
monkeypatch.setenv(environment, "1")
|
|
assert main(_argv(workspace, haplotype, index, _gatk(tmp_path))) == 1
|
|
failure = json.loads(
|
|
(
|
|
workspace
|
|
/ "final"
|
|
/ "Cohort.wgs_gvcf_combine.manifest.json"
|
|
).read_text()
|
|
)
|
|
assert failure["status"] == "failed"
|
|
assert failure["artifacts"] == {}
|
|
|
|
|
|
def test_dual_upstream_identity_is_mandatory(tmp_path: Path) -> None:
|
|
from apps.cli.run_wgs_gvcf_combine import main
|
|
|
|
workspace = tmp_path / "workspace"
|
|
haplotype, index = _authorities(workspace)
|
|
payload = json.loads(index.read_text())
|
|
payload["wgs_reference_id"] = "wgs-reference:" + "f" * 64
|
|
index.write_text(json.dumps(payload))
|
|
assert main(_argv(workspace, haplotype, index, _gatk(tmp_path))) == 1
|
|
|
|
|
|
def test_status_and_results_require_exact_identity_and_singleton_roles(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
import asyncio
|
|
|
|
from apps.cli.run_wgs_gvcf_combine import main
|
|
from apps.web.vue3 import main as web_main
|
|
from gfa_core.mas.tools.stage_status import get_stage_status
|
|
|
|
workspace_root = tmp_path / "workspaces"
|
|
workspace = workspace_root / "job-1"
|
|
haplotype, index = _authorities(workspace)
|
|
assert main(_argv(workspace, haplotype, index, _gatk(tmp_path))) == 0
|
|
manifest_path = (
|
|
workspace / "final" / "Cohort.wgs_gvcf_combine.manifest.json"
|
|
)
|
|
manifest = json.loads(manifest_path.read_text())
|
|
analysis_id = manifest["wgs_analysis_id"]
|
|
status = get_stage_status(
|
|
"job-1",
|
|
"Cohort",
|
|
"wgs_gvcf_combine",
|
|
workspace_root,
|
|
wgs_sample_set_id=SAMPLES,
|
|
wgs_reference_id=REFERENCE,
|
|
wgs_analysis_id=analysis_id,
|
|
)
|
|
assert status.completed
|
|
monkeypatch.setattr(web_main, "WORKSPACE_ROOT", workspace_root)
|
|
results = asyncio.run(
|
|
web_main.get_results(
|
|
job_id="job-1",
|
|
stage="wgs_gvcf_combine",
|
|
wgs_sample_set_id=SAMPLES,
|
|
wgs_reference_id=REFERENCE,
|
|
wgs_analysis_id=analysis_id,
|
|
)
|
|
)
|
|
serialized = json.dumps(results)
|
|
assert "combined_gvcf" in serialized
|
|
assert "combined_gvcf_index" in serialized
|
|
manifest["artifacts"].pop("combined_gvcf_index")
|
|
manifest_path.write_text(json.dumps(manifest))
|
|
broken = get_stage_status(
|
|
"job-1",
|
|
"Cohort",
|
|
"wgs_gvcf_combine",
|
|
workspace_root,
|
|
wgs_sample_set_id=SAMPLES,
|
|
wgs_reference_id=REFERENCE,
|
|
wgs_analysis_id=analysis_id,
|
|
)
|
|
assert not broken.completed
|
|
|
|
|
|
def test_backend_builds_stage07_command_and_accepts_only_fresh_manifest(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
from apps.cli.run_wgs_gvcf_combine import main
|
|
from apps.web import pipeline_backend
|
|
|
|
workspace = tmp_path / "workspace"
|
|
haplotype, index = _authorities(workspace)
|
|
gatk = _gatk(tmp_path)
|
|
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)
|
|
message, output = pipeline_backend.run_stage(
|
|
stage="wgs_gvcf_combine",
|
|
job=pipeline_backend.JobInfo(
|
|
"job-1", "", str(workspace), str(workspace / "logs"), "now"
|
|
),
|
|
prefix="Cohort",
|
|
wgs_haplotype_manifest=str(haplotype),
|
|
wgs_index_manifest=str(index),
|
|
wgs_sample_set_id=SAMPLES,
|
|
wgs_reference_id=REFERENCE,
|
|
wgs_haplotype_analysis_id=HAPLOTYPE_ANALYSIS,
|
|
wgs_index_analysis_id=INDEX_ANALYSIS,
|
|
wgs_gatk=str(gatk),
|
|
)
|
|
assert "completed" in message.lower()
|
|
assert output == (
|
|
workspace / "final" / "Cohort.wgs_gvcf_combine.manifest.json"
|
|
)
|
|
command = calls[0]
|
|
assert command[:3] == [
|
|
pipeline_backend.sys.executable,
|
|
"-m",
|
|
"apps.cli.run_wgs_gvcf_combine",
|
|
]
|
|
assert command[command.index("--haplotype-manifest") + 1] == str(
|
|
haplotype
|
|
)
|
|
before = output.read_bytes()
|
|
monkeypatch.setattr(
|
|
pipeline_backend, "run_cmd_logged", lambda *_args, **_kwargs: 0
|
|
)
|
|
stale_message, stale_output = pipeline_backend.run_stage(
|
|
stage="wgs_gvcf_combine",
|
|
job=pipeline_backend.JobInfo(
|
|
"job-1", "", str(workspace), str(workspace / "logs"), "now"
|
|
),
|
|
prefix="Cohort",
|
|
wgs_haplotype_manifest=str(haplotype),
|
|
wgs_index_manifest=str(index),
|
|
wgs_sample_set_id=SAMPLES,
|
|
wgs_reference_id=REFERENCE,
|
|
wgs_haplotype_analysis_id=HAPLOTYPE_ANALYSIS,
|
|
wgs_index_analysis_id=INDEX_ANALYSIS,
|
|
wgs_gatk=str(gatk),
|
|
)
|
|
assert "failed" in stale_message.lower()
|
|
assert stale_output == output
|
|
assert output.read_bytes() == before
|
|
|
|
|
|
def test_publish_lock_prevents_older_attempt_replacing_newer(
|
|
tmp_path: Path
|
|
) -> None:
|
|
from apps.cli.run_wgs_gvcf_combine import _publish
|
|
|
|
workspace = tmp_path / "workspace"
|
|
authority = workspace / "final" / "Cohort.wgs_gvcf_combine.manifest.json"
|
|
destination = (
|
|
workspace
|
|
/ "final"
|
|
/ ".wgs_gvcf_combine_runs"
|
|
/ "Cohort"
|
|
/ ("9" * 32)
|
|
)
|
|
authority.parent.mkdir(parents=True)
|
|
authority.write_text(
|
|
json.dumps({"status": "success", "start_time": "2026-07-23T12:00:00Z"})
|
|
)
|
|
bundle = tmp_path / "bundle"
|
|
bundle.mkdir()
|
|
(bundle / "artifact").write_text("older")
|
|
manifest_work = tmp_path / "manifest.json"
|
|
manifest_work.write_text("{}")
|
|
assert not _publish(
|
|
bundle,
|
|
destination,
|
|
manifest_work,
|
|
authority,
|
|
"2026-07-23T11:00:00Z",
|
|
)
|
|
assert json.loads(authority.read_text())["start_time"].startswith(
|
|
"2026-07-23T12"
|
|
)
|
|
|
|
|
|
def test_concurrent_publish_has_one_complete_authority(tmp_path: Path) -> None:
|
|
from apps.cli.run_wgs_gvcf_combine import _publish
|
|
|
|
workspace = tmp_path / "workspace"
|
|
authority = workspace / "final" / "Cohort.wgs_gvcf_combine.manifest.json"
|
|
authority.parent.mkdir(parents=True)
|
|
barrier = Barrier(2)
|
|
outcomes: list[bool] = []
|
|
|
|
def worker(name: str, started_at: str) -> None:
|
|
bundle = tmp_path / f"bundle-{name}"
|
|
bundle.mkdir()
|
|
(bundle / "artifact").write_text(name)
|
|
manifest = tmp_path / f"manifest-{name}.json"
|
|
manifest.write_text(
|
|
json.dumps(
|
|
{
|
|
"status": "success",
|
|
"start_time": started_at,
|
|
"winner": name,
|
|
}
|
|
)
|
|
)
|
|
destination = (
|
|
workspace
|
|
/ "final"
|
|
/ ".wgs_gvcf_combine_runs"
|
|
/ "Cohort"
|
|
/ name
|
|
)
|
|
barrier.wait()
|
|
outcomes.append(
|
|
_publish(
|
|
bundle, destination, manifest, authority, started_at
|
|
)
|
|
)
|
|
|
|
threads = [
|
|
Thread(target=worker, args=("a" * 32, "2026-07-23T11:00:00Z")),
|
|
Thread(target=worker, args=("b" * 32, "2026-07-23T12:00:00Z")),
|
|
]
|
|
for thread in threads:
|
|
thread.start()
|
|
for thread in threads:
|
|
thread.join()
|
|
manifest = json.loads(authority.read_text())
|
|
assert manifest["winner"] == "b" * 32
|
|
assert outcomes.count(True) in {1, 2}
|