688 lines
24 KiB
Python
688 lines
24 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
|
||
|
|
COMBINE_ANALYSIS = "wgs-analysis:" + "c" * 64
|
||
|
|
INDEX_ANALYSIS = "wgs-analysis:" + "d" * 64
|
||
|
|
|
||
|
|
|
||
|
|
def _record(workspace: Path, path: Path, role: 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(),
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def _write_vcf(path: Path, samples: list[str], *, gvcf: bool = True) -> None:
|
||
|
|
with gzip.open(path, "wt", encoding="utf-8") as handle:
|
||
|
|
handle.write("##fileformat=VCFv4.2\n")
|
||
|
|
if gvcf:
|
||
|
|
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"
|
||
|
|
)
|
||
|
|
alt = "<NON_REF>" if gvcf else "C"
|
||
|
|
handle.write(
|
||
|
|
f"chr1\t1\t.\tA\t{alt}\t50\tPASS\t.\tGT\t"
|
||
|
|
+ "\t".join("0/0" for _ in samples)
|
||
|
|
+ "\n"
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def _authorities(workspace: Path) -> tuple[Path, Path]:
|
||
|
|
final = workspace / "final"
|
||
|
|
combine_bundle = (
|
||
|
|
final / ".wgs_gvcf_combine_runs" / "Cohort" / ("7" * 32)
|
||
|
|
)
|
||
|
|
combine_bundle.mkdir(parents=True)
|
||
|
|
combined = (
|
||
|
|
combine_bundle / "Cohort.wgs_gvcf_combine.cohort.combined.g.vcf.gz"
|
||
|
|
)
|
||
|
|
combined_index = Path(f"{combined}.tbi")
|
||
|
|
combine_log = combine_bundle / "Cohort.wgs_gvcf_combine.cohort.combine.log"
|
||
|
|
_write_vcf(combined, ["S1", "S2", "S3"])
|
||
|
|
combined_index.write_bytes(b"TBI:combined")
|
||
|
|
combine_log.write_text("CombineGVCFs exit=0\n", encoding="utf-8")
|
||
|
|
combine = final / "Cohort.wgs_gvcf_combine.manifest.json"
|
||
|
|
combine.write_text(
|
||
|
|
json.dumps(
|
||
|
|
{
|
||
|
|
"manifest_version": "1.0",
|
||
|
|
"stage": "wgs_gvcf_combine",
|
||
|
|
"status": "success",
|
||
|
|
"prefix": "Cohort",
|
||
|
|
"job_id": "job-1",
|
||
|
|
"run_id": "7" * 32,
|
||
|
|
"workspace": str(workspace.resolve()),
|
||
|
|
"wgs_sample_set_id": SAMPLES,
|
||
|
|
"wgs_reference_id": REFERENCE,
|
||
|
|
"wgs_analysis_id": COMBINE_ANALYSIS,
|
||
|
|
"upstream_wgs_analysis_ids": {
|
||
|
|
"wgs_haplotype_calling": "wgs-analysis:" + "e" * 64,
|
||
|
|
"wgs_index": INDEX_ANALYSIS,
|
||
|
|
},
|
||
|
|
"metrics": {
|
||
|
|
"input_sample_ids": ["S1", "S2", "S3"],
|
||
|
|
"output_sample_ids": ["S1", "S2", "S3"],
|
||
|
|
},
|
||
|
|
"artifacts": {
|
||
|
|
"combined_gvcf": _record(
|
||
|
|
workspace, combined, "combined_gvcf"
|
||
|
|
),
|
||
|
|
"combined_gvcf_index": _record(
|
||
|
|
workspace, combined_index, "combined_gvcf_index"
|
||
|
|
),
|
||
|
|
"combine_log": _record(
|
||
|
|
workspace, combine_log, "combine_log"
|
||
|
|
),
|
||
|
|
},
|
||
|
|
}
|
||
|
|
),
|
||
|
|
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 combine, index
|
||
|
|
|
||
|
|
|
||
|
|
def test_stage08_rejects_producer_from_different_index_generation_at_all_gates(
|
||
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||
|
|
) -> None:
|
||
|
|
from apps.cli.run_wgs_joint_genotyping 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"
|
||
|
|
combine, index = _authorities(workspace)
|
||
|
|
gatk = _gatk(tmp_path)
|
||
|
|
assert main(_argv(workspace, combine, index, gatk)) == 0
|
||
|
|
output = workspace / "final" / "Cohort.wgs_joint_genotyping.manifest.json"
|
||
|
|
output_analysis = json.loads(output.read_text())["wgs_analysis_id"]
|
||
|
|
|
||
|
|
producer = json.loads(combine.read_text())
|
||
|
|
producer["upstream_wgs_analysis_ids"]["wgs_index"] = (
|
||
|
|
"wgs-analysis:" + "0" * 64
|
||
|
|
)
|
||
|
|
combine.write_text(json.dumps(producer))
|
||
|
|
assert main(_argv(workspace, combine, 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_joint_genotyping",
|
||
|
|
job=pipeline_backend.JobInfo(
|
||
|
|
"job-1", "", str(workspace), str(workspace / "logs"), "now"
|
||
|
|
),
|
||
|
|
prefix="Cohort",
|
||
|
|
wgs_combine_manifest=str(combine),
|
||
|
|
wgs_index_manifest=str(index),
|
||
|
|
wgs_sample_set_id=SAMPLES,
|
||
|
|
wgs_reference_id=REFERENCE,
|
||
|
|
wgs_combine_analysis_id=COMBINE_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_joint_genotyping", 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 == 'GenotypeGVCFs':
|
||
|
|
source = pathlib.Path(value('-V'))
|
||
|
|
with gzip.open(source, 'rt') as handle:
|
||
|
|
header = next(line for line in handle if line.startswith('#CHROM'))
|
||
|
|
samples = 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('#CHROM\\tPOS\\tID\\tREF\\tALT\\tQUAL\\tFILTER\\tINFO'
|
||
|
|
'\\tFORMAT\\t' + '\\t'.join(samples) + '\\n')
|
||
|
|
handle.write('chr1\\t1\\t.\\tA\\tC\\t50\\tPASS\\t.\\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:joint'
|
||
|
|
pathlib.Path(str(path) + '.tbi').write_bytes(content)
|
||
|
|
raise SystemExit(0)
|
||
|
|
if command == 'VariantsToTable':
|
||
|
|
pathlib.Path(value('-O')).write_text(
|
||
|
|
'CHROM\\tPOS\\tREF\\tALT\\tQUAL\\nchr1\\t1\\tA\\tC\\t50\\n',
|
||
|
|
encoding='utf-8',
|
||
|
|
)
|
||
|
|
raise SystemExit(0)
|
||
|
|
raise SystemExit(2)
|
||
|
|
""",
|
||
|
|
encoding="utf-8",
|
||
|
|
)
|
||
|
|
tool.chmod(0o755)
|
||
|
|
return tool
|
||
|
|
|
||
|
|
|
||
|
|
def _argv(
|
||
|
|
workspace: Path, combine: Path, index: Path, gatk: Path
|
||
|
|
) -> list[str]:
|
||
|
|
return [
|
||
|
|
"--workspace",
|
||
|
|
str(workspace),
|
||
|
|
"--job-id",
|
||
|
|
"job-1",
|
||
|
|
"--prefix",
|
||
|
|
"Cohort",
|
||
|
|
"--combine-manifest",
|
||
|
|
str(combine),
|
||
|
|
"--index-manifest",
|
||
|
|
str(index),
|
||
|
|
"--gatk",
|
||
|
|
str(gatk),
|
||
|
|
]
|
||
|
|
|
||
|
|
|
||
|
|
def test_wgs_joint_genotyping_exposes_import_safe_cli() -> None:
|
||
|
|
from apps.cli.run_wgs_joint_genotyping import build_parser, main
|
||
|
|
|
||
|
|
assert callable(build_parser)
|
||
|
|
assert callable(main)
|
||
|
|
|
||
|
|
|
||
|
|
def test_parser_locks_scientific_defaults() -> None:
|
||
|
|
from apps.cli.run_wgs_joint_genotyping import build_parser
|
||
|
|
|
||
|
|
parsed = build_parser().parse_args(
|
||
|
|
[
|
||
|
|
"--workspace", "w", "--job-id", "j", "--prefix", "P",
|
||
|
|
"--combine-manifest", "c", "--index-manifest", "i",
|
||
|
|
]
|
||
|
|
)
|
||
|
|
assert parsed.max_alternate_alleles == 6
|
||
|
|
assert parsed.calling_threshold == 10
|
||
|
|
assert parsed.include_non_variant_sites is False
|
||
|
|
|
||
|
|
|
||
|
|
def test_three_samples_produce_only_canonical_outputs_and_parameter_snapshot(
|
||
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||
|
|
) -> None:
|
||
|
|
from apps.cli.run_wgs_joint_genotyping import main
|
||
|
|
|
||
|
|
workspace = tmp_path / "workspace"
|
||
|
|
combine, index = _authorities(workspace)
|
||
|
|
calls = tmp_path / "gatk.calls"
|
||
|
|
monkeypatch.setenv("GATK_CALLS", str(calls))
|
||
|
|
assert main(_argv(workspace, combine, index, _gatk(tmp_path))) == 0
|
||
|
|
authority = workspace / "final" / "Cohort.wgs_joint_genotyping.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_gvcf_combine": COMBINE_ANALYSIS,
|
||
|
|
"wgs_index": INDEX_ANALYSIS,
|
||
|
|
}
|
||
|
|
assert manifest["parameters"] == {
|
||
|
|
"gatk": str(_gatk(tmp_path)),
|
||
|
|
"operation": "GenotypeGVCFs",
|
||
|
|
"max_alternate_alleles": 6,
|
||
|
|
"calling_threshold": 10,
|
||
|
|
"include_non_variant_sites": False,
|
||
|
|
}
|
||
|
|
assert set(manifest["artifacts"]) == {
|
||
|
|
"joint_vcf", "joint_vcf_index", "joint_summary", "joint_log"
|
||
|
|
}
|
||
|
|
paths = {
|
||
|
|
item["role"]: Path(item["full_path"])
|
||
|
|
for item in manifest["artifacts"].values()
|
||
|
|
}
|
||
|
|
assert paths["joint_vcf"].name == (
|
||
|
|
"Cohort.wgs_joint_genotyping.cohort.joint.vcf.gz"
|
||
|
|
)
|
||
|
|
assert paths["joint_vcf_index"] == Path(f"{paths['joint_vcf']}.tbi")
|
||
|
|
assert paths["joint_summary"].read_text().splitlines() == [
|
||
|
|
"CHROM\tPOS\tREF\tALT\tQUAL", "chr1\t1\tA\tC\t50"
|
||
|
|
]
|
||
|
|
with gzip.open(paths["joint_vcf"], "rt") as handle:
|
||
|
|
header = next(line for line in handle if line.startswith("#CHROM"))
|
||
|
|
assert header.rstrip().split("\t")[9:] == ["S1", "S2", "S3"]
|
||
|
|
invocations = [json.loads(line) for line in calls.read_text().splitlines()]
|
||
|
|
genotype = next(call for call in invocations if call[0] == "GenotypeGVCFs")
|
||
|
|
assert genotype[genotype.index("--max-alternate-alleles") + 1] == "6"
|
||
|
|
assert genotype[genotype.index("--standard-min-confidence-threshold-for-calling") + 1] == "10"
|
||
|
|
assert genotype[genotype.index("--include-non-variant-sites") + 1] == "false"
|
||
|
|
assert any(call[0] == "ValidateVariants" for call in invocations)
|
||
|
|
|
||
|
|
|
||
|
|
def test_corrupt_nonempty_input_tbi_is_rejected_by_indexed_query(
|
||
|
|
tmp_path: Path
|
||
|
|
) -> None:
|
||
|
|
from apps.cli.run_wgs_joint_genotyping import main
|
||
|
|
|
||
|
|
workspace = tmp_path / "workspace"
|
||
|
|
combine, index = _authorities(workspace)
|
||
|
|
payload = json.loads(combine.read_text())
|
||
|
|
record = payload["artifacts"]["combined_gvcf_index"]
|
||
|
|
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()
|
||
|
|
combine.write_text(json.dumps(payload))
|
||
|
|
assert main(_argv(workspace, combine, index, _gatk(tmp_path))) == 1
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize(
|
||
|
|
("flag", "value"),
|
||
|
|
(
|
||
|
|
("--max-alternate-alleles", "7"),
|
||
|
|
("--calling-threshold", "11"),
|
||
|
|
("--include-non-variant-sites", "true"),
|
||
|
|
),
|
||
|
|
)
|
||
|
|
def test_scientific_parameters_are_locked(
|
||
|
|
tmp_path: Path, flag: str, value: str
|
||
|
|
) -> None:
|
||
|
|
from apps.cli.run_wgs_joint_genotyping import main
|
||
|
|
|
||
|
|
workspace = tmp_path / "workspace"
|
||
|
|
combine, index = _authorities(workspace)
|
||
|
|
argv = _argv(workspace, combine, index, _gatk(tmp_path))
|
||
|
|
argv.extend([flag, value])
|
||
|
|
assert main(argv) == 1
|
||
|
|
|
||
|
|
|
||
|
|
def test_declared_index_must_be_the_vcf_sidecar(tmp_path: Path) -> None:
|
||
|
|
from apps.cli.run_wgs_joint_genotyping import main
|
||
|
|
|
||
|
|
workspace = tmp_path / "workspace"
|
||
|
|
combine, index = _authorities(workspace)
|
||
|
|
payload = json.loads(combine.read_text())
|
||
|
|
record = payload["artifacts"]["combined_gvcf_index"]
|
||
|
|
unrelated = Path(record["full_path"]).with_name("unrelated.tbi")
|
||
|
|
unrelated.write_bytes(b"TBI:unrelated")
|
||
|
|
record.update(
|
||
|
|
{
|
||
|
|
"path": str(unrelated.relative_to(workspace)),
|
||
|
|
"full_path": str(unrelated),
|
||
|
|
"size": unrelated.stat().st_size,
|
||
|
|
"size_bytes": unrelated.stat().st_size,
|
||
|
|
"sha256": hashlib.sha256(unrelated.read_bytes()).hexdigest(),
|
||
|
|
}
|
||
|
|
)
|
||
|
|
combine.write_text(json.dumps(payload))
|
||
|
|
assert main(_argv(workspace, combine, index, _gatk(tmp_path))) == 1
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize(
|
||
|
|
"failure", ("GenotypeGVCFs", "IndexFeatureFile", "VariantsToTable")
|
||
|
|
)
|
||
|
|
def test_each_tool_return_code_fails_and_preserves_old_authority(
|
||
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, failure: str
|
||
|
|
) -> None:
|
||
|
|
from apps.cli.run_wgs_joint_genotyping import main
|
||
|
|
|
||
|
|
workspace = tmp_path / "workspace"
|
||
|
|
combine, index = _authorities(workspace)
|
||
|
|
authority = workspace / "final" / "Cohort.wgs_joint_genotyping.manifest.json"
|
||
|
|
authority.write_text(
|
||
|
|
json.dumps({"stage": "wgs_joint_genotyping", "status": "success", "sentinel": True})
|
||
|
|
)
|
||
|
|
before = hashlib.sha256(authority.read_bytes()).hexdigest()
|
||
|
|
monkeypatch.setenv("FAIL_COMMAND", failure)
|
||
|
|
assert main(_argv(workspace, combine, index, _gatk(tmp_path))) == 1
|
||
|
|
assert hashlib.sha256(authority.read_bytes()).hexdigest() == before
|
||
|
|
sidecars = list(
|
||
|
|
(workspace / "final").glob(
|
||
|
|
"Cohort.wgs_joint_genotyping.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_content_or_index_failure_is_not_published(
|
||
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, environment: str
|
||
|
|
) -> None:
|
||
|
|
from apps.cli.run_wgs_joint_genotyping import main
|
||
|
|
|
||
|
|
workspace = tmp_path / "workspace"
|
||
|
|
combine, index = _authorities(workspace)
|
||
|
|
monkeypatch.setenv(environment, "1")
|
||
|
|
assert main(_argv(workspace, combine, index, _gatk(tmp_path))) == 1
|
||
|
|
failure = json.loads(
|
||
|
|
(workspace / "final" / "Cohort.wgs_joint_genotyping.manifest.json").read_text()
|
||
|
|
)
|
||
|
|
assert failure["status"] == "failed"
|
||
|
|
assert failure["artifacts"] == {}
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize(
|
||
|
|
"mutation", ("combine_analysis", "reference", "path", "symlink", "roles")
|
||
|
|
)
|
||
|
|
def test_two_upstream_authorities_are_strictly_bound(
|
||
|
|
tmp_path: Path, mutation: str
|
||
|
|
) -> None:
|
||
|
|
from apps.cli.run_wgs_joint_genotyping import main
|
||
|
|
|
||
|
|
workspace = tmp_path / "workspace"
|
||
|
|
combine, index = _authorities(workspace)
|
||
|
|
if mutation == "combine_analysis":
|
||
|
|
payload = json.loads(combine.read_text())
|
||
|
|
payload.pop("wgs_analysis_id")
|
||
|
|
combine.write_text(json.dumps(payload))
|
||
|
|
elif mutation == "reference":
|
||
|
|
payload = json.loads(index.read_text())
|
||
|
|
payload["wgs_reference_id"] = "wgs-reference:" + "f" * 64
|
||
|
|
index.write_text(json.dumps(payload))
|
||
|
|
elif mutation == "path":
|
||
|
|
payload = json.loads(combine.read_text())
|
||
|
|
payload["artifacts"]["combined_gvcf"]["path"] = "../escape.vcf.gz"
|
||
|
|
combine.write_text(json.dumps(payload))
|
||
|
|
elif mutation == "symlink":
|
||
|
|
payload = json.loads(combine.read_text())
|
||
|
|
record = payload["artifacts"]["combined_gvcf"]
|
||
|
|
original = Path(record["full_path"])
|
||
|
|
link = original.with_name("linked.g.vcf.gz")
|
||
|
|
link.symlink_to(original)
|
||
|
|
record["path"] = str(link.relative_to(workspace))
|
||
|
|
record["full_path"] = str(link)
|
||
|
|
combine.write_text(json.dumps(payload))
|
||
|
|
else:
|
||
|
|
payload = json.loads(combine.read_text())
|
||
|
|
payload["artifacts"].pop("combined_gvcf_index")
|
||
|
|
combine.write_text(json.dumps(payload))
|
||
|
|
assert main(_argv(workspace, combine, 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_joint_genotyping 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"
|
||
|
|
combine, index = _authorities(workspace)
|
||
|
|
assert main(_argv(workspace, combine, index, _gatk(tmp_path))) == 0
|
||
|
|
manifest_path = (
|
||
|
|
workspace / "final" / "Cohort.wgs_joint_genotyping.manifest.json"
|
||
|
|
)
|
||
|
|
manifest = json.loads(manifest_path.read_text())
|
||
|
|
analysis_id = manifest["wgs_analysis_id"]
|
||
|
|
status = get_stage_status(
|
||
|
|
"job-1",
|
||
|
|
"Cohort",
|
||
|
|
"wgs_joint_genotyping",
|
||
|
|
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_joint_genotyping",
|
||
|
|
wgs_sample_set_id=SAMPLES,
|
||
|
|
wgs_reference_id=REFERENCE,
|
||
|
|
wgs_analysis_id=analysis_id,
|
||
|
|
)
|
||
|
|
)
|
||
|
|
serialized = json.dumps(results)
|
||
|
|
assert "joint_vcf" in serialized
|
||
|
|
assert "joint_summary" in serialized
|
||
|
|
manifest["artifacts"].pop("joint_vcf_index")
|
||
|
|
manifest_path.write_text(json.dumps(manifest))
|
||
|
|
assert not get_stage_status(
|
||
|
|
"job-1",
|
||
|
|
"Cohort",
|
||
|
|
"wgs_joint_genotyping",
|
||
|
|
workspace_root,
|
||
|
|
wgs_sample_set_id=SAMPLES,
|
||
|
|
wgs_reference_id=REFERENCE,
|
||
|
|
wgs_analysis_id=analysis_id,
|
||
|
|
).completed
|
||
|
|
|
||
|
|
|
||
|
|
def test_backend_builds_stage08_command_and_accepts_only_fresh_manifest(
|
||
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||
|
|
) -> None:
|
||
|
|
from apps.cli.run_wgs_joint_genotyping import main
|
||
|
|
from apps.web import pipeline_backend
|
||
|
|
|
||
|
|
workspace = tmp_path / "workspace"
|
||
|
|
combine, 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_joint_genotyping",
|
||
|
|
job=pipeline_backend.JobInfo(
|
||
|
|
"job-1", "", str(workspace), str(workspace / "logs"), "now"
|
||
|
|
),
|
||
|
|
prefix="Cohort",
|
||
|
|
wgs_combine_manifest=str(combine),
|
||
|
|
wgs_index_manifest=str(index),
|
||
|
|
wgs_sample_set_id=SAMPLES,
|
||
|
|
wgs_reference_id=REFERENCE,
|
||
|
|
wgs_combine_analysis_id=COMBINE_ANALYSIS,
|
||
|
|
wgs_index_analysis_id=INDEX_ANALYSIS,
|
||
|
|
wgs_gatk=str(gatk),
|
||
|
|
wgs_max_alternate_alleles=6,
|
||
|
|
wgs_calling_threshold=10,
|
||
|
|
wgs_include_non_variant_sites=False,
|
||
|
|
)
|
||
|
|
assert "completed" in message.lower()
|
||
|
|
assert output == (
|
||
|
|
workspace / "final" / "Cohort.wgs_joint_genotyping.manifest.json"
|
||
|
|
)
|
||
|
|
command = calls[0]
|
||
|
|
assert command[:3] == [
|
||
|
|
pipeline_backend.sys.executable,
|
||
|
|
"-m",
|
||
|
|
"apps.cli.run_wgs_joint_genotyping",
|
||
|
|
]
|
||
|
|
assert command[command.index("--combine-manifest") + 1] == str(combine)
|
||
|
|
assert command[command.index("--max-alternate-alleles") + 1] == "6"
|
||
|
|
assert command[command.index("--calling-threshold") + 1] == "10"
|
||
|
|
assert command[command.index("--include-non-variant-sites") + 1] == "false"
|
||
|
|
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_joint_genotyping",
|
||
|
|
job=pipeline_backend.JobInfo(
|
||
|
|
"job-1", "", str(workspace), str(workspace / "logs"), "now"
|
||
|
|
),
|
||
|
|
prefix="Cohort",
|
||
|
|
wgs_combine_manifest=str(combine),
|
||
|
|
wgs_index_manifest=str(index),
|
||
|
|
wgs_sample_set_id=SAMPLES,
|
||
|
|
wgs_reference_id=REFERENCE,
|
||
|
|
wgs_combine_analysis_id=COMBINE_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_joint_genotyping import _publish
|
||
|
|
|
||
|
|
workspace = tmp_path / "workspace"
|
||
|
|
authority = workspace / "final" / "Cohort.wgs_joint_genotyping.manifest.json"
|
||
|
|
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,
|
||
|
|
workspace / "final" / ".wgs_joint_genotyping_runs" / "Cohort" / ("9" * 32),
|
||
|
|
manifest_work,
|
||
|
|
authority,
|
||
|
|
"2026-07-23T11:00:00Z",
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def test_publish_lock_rejects_equal_timestamp_attempt(tmp_path: Path) -> None:
|
||
|
|
from apps.cli.run_wgs_joint_genotyping import _publish
|
||
|
|
|
||
|
|
workspace = tmp_path / "workspace"
|
||
|
|
authority = workspace / "final" / "Cohort.wgs_joint_genotyping.manifest.json"
|
||
|
|
authority.parent.mkdir(parents=True)
|
||
|
|
timestamp = "2026-07-23T12:00:00Z"
|
||
|
|
authority.write_text(
|
||
|
|
json.dumps({"status": "success", "start_time": timestamp})
|
||
|
|
)
|
||
|
|
bundle = tmp_path / "bundle"
|
||
|
|
bundle.mkdir()
|
||
|
|
(bundle / "artifact").write_text("equal")
|
||
|
|
manifest_work = tmp_path / "manifest.json"
|
||
|
|
manifest_work.write_text("{}")
|
||
|
|
assert not _publish(
|
||
|
|
bundle,
|
||
|
|
workspace / "final" / ".wgs_joint_genotyping_runs" / "Cohort" / ("8" * 32),
|
||
|
|
manifest_work,
|
||
|
|
authority,
|
||
|
|
timestamp,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def test_concurrent_publish_has_one_complete_authority(tmp_path: Path) -> None:
|
||
|
|
from apps.cli.run_wgs_joint_genotyping import _publish
|
||
|
|
|
||
|
|
workspace = tmp_path / "workspace"
|
||
|
|
authority = workspace / "final" / "Cohort.wgs_joint_genotyping.manifest.json"
|
||
|
|
authority.parent.mkdir(parents=True)
|
||
|
|
barrier = Barrier(2)
|
||
|
|
|
||
|
|
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})
|
||
|
|
)
|
||
|
|
barrier.wait()
|
||
|
|
_publish(
|
||
|
|
bundle,
|
||
|
|
workspace / "final" / ".wgs_joint_genotyping_runs" / "Cohort" / name,
|
||
|
|
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()
|
||
|
|
assert json.loads(authority.read_text())["winner"] == "b" * 32
|