797 lines
25 KiB
Python
797 lines
25 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
|
|
JOINT_ANALYSIS = "wgs-analysis:" + "c" * 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_joint_vcf(path: Path, *, include_indel: bool = True) -> None:
|
|
with gzip.open(path, "wt", encoding="utf-8") as handle:
|
|
handle.write("##fileformat=VCFv4.2\n")
|
|
handle.write("##contig=<ID=chr1,length=1000>\n")
|
|
handle.write(
|
|
"#CHROM\tPOS\tID\tREF\tALT\tQUAL\tFILTER\tINFO\tFORMAT"
|
|
"\tS1\tS2\tS3\n"
|
|
)
|
|
records = [
|
|
"chr1\t1\t.\tA\tC\t50\tPASS\t.\tGT\t0/0\t0/1\t1/1\n",
|
|
"chr1\t3\t.\tAT\tGC\t50\tPASS\t.\tGT\t0/0\t0/1\t1/1\n",
|
|
]
|
|
if include_indel:
|
|
records.insert(
|
|
1,
|
|
"chr1\t2\t.\tA\tAT\t50\tPASS\t.\tGT\t0/0\t0/1\t1/1\n",
|
|
)
|
|
for record in records:
|
|
handle.write(record)
|
|
|
|
|
|
def _joint_authority(
|
|
workspace: Path, *, include_indel: bool = True
|
|
) -> Path:
|
|
bundle = (
|
|
workspace
|
|
/ "final"
|
|
/ ".wgs_joint_genotyping_runs"
|
|
/ "Cohort"
|
|
/ ("8" * 32)
|
|
)
|
|
bundle.mkdir(parents=True)
|
|
vcf = bundle / "Cohort.wgs_joint_genotyping.cohort.joint.vcf.gz"
|
|
tbi = Path(f"{vcf}.tbi")
|
|
summary = bundle / "Cohort.wgs_joint_genotyping.cohort.summary.tsv"
|
|
log = bundle / "Cohort.wgs_joint_genotyping.cohort.joint.log"
|
|
_write_joint_vcf(vcf, include_indel=include_indel)
|
|
tbi.write_bytes(b"TBI:joint")
|
|
summary.write_text(
|
|
"CHROM\tPOS\tREF\tALT\tQUAL\nchr1\t1\tA\tC\t50\n",
|
|
encoding="utf-8",
|
|
)
|
|
log.write_text("GenotypeGVCFs exit=0\n", encoding="utf-8")
|
|
authority = (
|
|
workspace / "final" / "Cohort.wgs_joint_genotyping.manifest.json"
|
|
)
|
|
authority.write_text(
|
|
json.dumps(
|
|
{
|
|
"manifest_version": "1.0",
|
|
"stage": "wgs_joint_genotyping",
|
|
"status": "success",
|
|
"prefix": "Cohort",
|
|
"job_id": "job-1",
|
|
"run_id": "8" * 32,
|
|
"workspace": str(workspace.resolve()),
|
|
"wgs_sample_set_id": SAMPLES,
|
|
"wgs_reference_id": REFERENCE,
|
|
"wgs_analysis_id": JOINT_ANALYSIS,
|
|
"metrics": {
|
|
"input_sample_ids": ["S1", "S2", "S3"],
|
|
"output_sample_ids": ["S1", "S2", "S3"],
|
|
},
|
|
"artifacts": {
|
|
"joint_vcf": _record(workspace, vcf, "joint_vcf"),
|
|
"joint_vcf_index": _record(
|
|
workspace, tbi, "joint_vcf_index"
|
|
),
|
|
"joint_summary": _record(
|
|
workspace, summary, "joint_summary"
|
|
),
|
|
"joint_log": _record(workspace, log, "joint_log"),
|
|
},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
return authority
|
|
|
|
|
|
def _tool(tmp_path: Path, name: str) -> Path:
|
|
tool = tmp_path / name
|
|
tool.write_text(
|
|
"""#!/usr/bin/env python3
|
|
import gzip, json, os, pathlib, sys
|
|
name = pathlib.Path(sys.argv[0]).name
|
|
args = sys.argv[1:]
|
|
if args in (['--version'], ['version']):
|
|
print(('The Genome Analysis Toolkit (GATK) v4.6.0.0'
|
|
if name == 'gatk' else 'bcftools 1.20'))
|
|
raise SystemExit(0)
|
|
calls = os.environ.get('WGS_SELECT_CALLS')
|
|
if calls:
|
|
with open(calls, 'a', encoding='utf-8') as handle:
|
|
handle.write(json.dumps([name] + args) + '\\n')
|
|
def value(flag):
|
|
return args[args.index(flag) + 1]
|
|
def records(path):
|
|
with gzip.open(path, 'rt', encoding='utf-8') as handle:
|
|
return list(handle)
|
|
def select(source, output, kind):
|
|
lines = records(source)
|
|
with gzip.open(output, 'wt', encoding='utf-8') as handle:
|
|
for line in lines:
|
|
if line.startswith('#'):
|
|
handle.write(line)
|
|
continue
|
|
fields = line.rstrip().split('\\t')
|
|
ref = fields[3]
|
|
alts = fields[4].split(',')
|
|
is_snp = len(ref) == 1 and all(len(alt) == 1 for alt in alts)
|
|
is_indel = any(len(alt) != len(ref) for alt in alts)
|
|
if (kind == 'SNP' and is_snp) or (kind == 'INDEL' and is_indel):
|
|
handle.write(line)
|
|
fail = os.environ.get('FAIL_POINT', '')
|
|
if name == 'gatk':
|
|
command = args[0]
|
|
if command == 'SelectVariants':
|
|
kind = value('-select-type-to-include')
|
|
if fail == 'gatk:select:' + kind:
|
|
raise SystemExit(23)
|
|
select(pathlib.Path(value('-V')), pathlib.Path(value('-O')), kind)
|
|
raise SystemExit(0)
|
|
if command == 'IndexFeatureFile':
|
|
path = pathlib.Path(value('-I'))
|
|
kind = 'SNP' if '.snp.' in path.name else 'INDEL'
|
|
if fail == 'gatk:index:' + kind:
|
|
raise SystemExit(23)
|
|
marker = (
|
|
b'CORRUPT:gatk'
|
|
if fail == 'gatk:corrupt-index:' + kind
|
|
else b'TBI:gatk'
|
|
)
|
|
pathlib.Path(str(path) + '.tbi').write_bytes(marker)
|
|
raise SystemExit(0)
|
|
if command == 'ValidateVariants':
|
|
path = pathlib.Path(value('-V'))
|
|
kind = ('input' if '.joint.' in path.name else
|
|
('SNP' if '.snp.' in path.name else 'INDEL'))
|
|
if fail == 'gatk:validate:' + kind:
|
|
raise SystemExit(23)
|
|
index = pathlib.Path(str(path) + '.tbi')
|
|
if ('-L' in args and (
|
|
not index.is_file()
|
|
or index.read_bytes().startswith(b'CORRUPT')
|
|
)):
|
|
raise SystemExit(24)
|
|
raise SystemExit(0)
|
|
raise SystemExit(2)
|
|
if args[0] == 'view':
|
|
if '-v' in args:
|
|
raw = value('-v')
|
|
kind = 'SNP' if raw == 'snps' else 'INDEL'
|
|
if fail == 'bcftools:select:' + kind:
|
|
raise SystemExit(23)
|
|
select(pathlib.Path(args[-1]), pathlib.Path(value('-o')), kind)
|
|
raise SystemExit(0)
|
|
path = pathlib.Path(args[-1])
|
|
label = 'input' if '.joint.' in path.name else (
|
|
'SNP' if '.snp.' in path.name else 'INDEL'
|
|
)
|
|
if fail == 'bcftools:validate:' + label:
|
|
raise SystemExit(23)
|
|
index = pathlib.Path(str(path) + '.tbi')
|
|
if not index.is_file() or index.read_bytes().startswith(b'CORRUPT'):
|
|
raise SystemExit(24)
|
|
for line in records(path):
|
|
if not line.startswith('#'):
|
|
sys.stdout.write(line)
|
|
raise SystemExit(0)
|
|
if args[0] == 'index':
|
|
path = pathlib.Path(args[-1])
|
|
kind = 'SNP' if '.snp.' in path.name else 'INDEL'
|
|
if '-n' in args:
|
|
if fail == 'bcftools:validate:' + (
|
|
'input' if '.joint.' in path.name else kind
|
|
):
|
|
raise SystemExit(23)
|
|
index = pathlib.Path(str(path) + '.tbi')
|
|
if not index.is_file() or index.read_bytes().startswith(b'CORRUPT'):
|
|
raise SystemExit(24)
|
|
print('3')
|
|
raise SystemExit(0)
|
|
if fail == 'bcftools:index:' + kind:
|
|
raise SystemExit(23)
|
|
pathlib.Path(str(path) + '.tbi').write_bytes(b'TBI:bcftools')
|
|
raise SystemExit(0)
|
|
if args[0] == 'stats':
|
|
path = pathlib.Path(args[-1])
|
|
kind = 'SNP' if '.snp.' in path.name else 'INDEL'
|
|
if fail == 'bcftools:stats:' + kind:
|
|
raise SystemExit(23)
|
|
print('SN\\t0\\tnumber of records:\\t1')
|
|
raise SystemExit(0)
|
|
raise SystemExit(2)
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
tool.chmod(0o755)
|
|
return tool
|
|
|
|
|
|
def _argv(
|
|
workspace: Path,
|
|
joint: Path,
|
|
gatk: Path,
|
|
bcftools: Path,
|
|
method: str = "gatk",
|
|
) -> list[str]:
|
|
return [
|
|
"--workspace",
|
|
str(workspace),
|
|
"--job-id",
|
|
"job-1",
|
|
"--prefix",
|
|
"Cohort",
|
|
"--joint-manifest",
|
|
str(joint),
|
|
"--method",
|
|
method,
|
|
"--gatk",
|
|
str(gatk),
|
|
"--bcftools",
|
|
str(bcftools),
|
|
]
|
|
|
|
|
|
def _variant_kinds(path: Path) -> set[str]:
|
|
observed = set()
|
|
with gzip.open(path, "rt", encoding="utf-8") as handle:
|
|
for line in handle:
|
|
if line.startswith("#"):
|
|
continue
|
|
fields = line.rstrip().split("\t")
|
|
ref, alts = fields[3], fields[4].split(",")
|
|
if len(ref) == 1 and all(len(alt) == 1 for alt in alts):
|
|
observed.add("SNP")
|
|
elif any(len(alt) != len(ref) for alt in alts):
|
|
observed.add("INDEL")
|
|
else:
|
|
observed.add("OTHER")
|
|
return observed
|
|
|
|
|
|
def test_wgs_variant_select_exposes_import_safe_cli() -> None:
|
|
from apps.cli.run_wgs_variant_select import build_parser, main
|
|
|
|
assert callable(build_parser)
|
|
assert callable(main)
|
|
|
|
|
|
def test_parser_defaults_to_gatk_and_accepts_only_two_methods() -> None:
|
|
from apps.cli.run_wgs_variant_select import build_parser
|
|
|
|
parser = build_parser()
|
|
base = [
|
|
"--workspace", "w", "--job-id", "j", "--prefix", "P",
|
|
"--joint-manifest", "joint.json",
|
|
]
|
|
assert parser.parse_args(base).method == "gatk"
|
|
assert parser.parse_args(base + ["--method", "bcftools"]).method == "bcftools"
|
|
with pytest.raises(SystemExit):
|
|
parser.parse_args(base + ["--method", "auto"])
|
|
|
|
|
|
def test_both_methods_publish_same_roles_and_names_but_distinct_authorities(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
from apps.cli.run_wgs_variant_select import main
|
|
from gfa_core.wgs_upstream_manifest import (
|
|
WgsUpstreamManifestError,
|
|
resolve_wgs_artifacts,
|
|
)
|
|
|
|
workspace = tmp_path / "workspace"
|
|
joint = _joint_authority(workspace)
|
|
gatk = _tool(tmp_path, "gatk")
|
|
bcftools = _tool(tmp_path, "bcftools")
|
|
manifests = {}
|
|
for method in ("gatk", "bcftools"):
|
|
assert main(_argv(workspace, joint, gatk, bcftools, method)) == 0
|
|
authority = (
|
|
workspace
|
|
/ "final"
|
|
/ f"Cohort.wgs_variant_select.{method}.manifest.json"
|
|
)
|
|
manifests[method] = json.loads(authority.read_text())
|
|
resolved = resolve_wgs_artifacts(
|
|
authority,
|
|
"wgs_variant_select",
|
|
("selected_snp_vcf", "selected_indel_vcf"),
|
|
expected_sample_set_id=SAMPLES,
|
|
expected_reference_id=REFERENCE,
|
|
expected_analysis_id=manifests[method]["wgs_analysis_id"],
|
|
expected_method=method,
|
|
)
|
|
assert set(resolved) == {"selected_snp_vcf", "selected_indel_vcf"}
|
|
other_method = "bcftools" if method == "gatk" else "gatk"
|
|
with pytest.raises(WgsUpstreamManifestError, match="method mismatch"):
|
|
resolve_wgs_artifacts(
|
|
authority,
|
|
"wgs_variant_select",
|
|
("selected_snp_vcf",),
|
|
expected_method=other_method,
|
|
)
|
|
expected_roles = {
|
|
"selected_snp_vcf",
|
|
"selected_snp_vcf_index",
|
|
"selected_indel_vcf",
|
|
"selected_indel_vcf_index",
|
|
"variant_select_report",
|
|
}
|
|
assert set(manifests["gatk"]["artifacts"]) == expected_roles
|
|
assert set(manifests["bcftools"]["artifacts"]) == expected_roles
|
|
assert manifests["gatk"]["parameters"]["method"] == "gatk"
|
|
assert manifests["bcftools"]["parameters"]["method"] == "bcftools"
|
|
assert (
|
|
manifests["gatk"]["wgs_analysis_id"]
|
|
!= manifests["bcftools"]["wgs_analysis_id"]
|
|
)
|
|
names = {}
|
|
for method, manifest in manifests.items():
|
|
names[method] = {
|
|
role: Path(record["full_path"]).name
|
|
for role, record in manifest["artifacts"].items()
|
|
}
|
|
snp = Path(manifest["artifacts"]["selected_snp_vcf"]["full_path"])
|
|
indel = Path(manifest["artifacts"]["selected_indel_vcf"]["full_path"])
|
|
assert _variant_kinds(snp) == {"SNP"}
|
|
assert _variant_kinds(indel) == {"INDEL"}
|
|
assert Path(
|
|
manifest["artifacts"]["selected_snp_vcf_index"]["full_path"]
|
|
) == Path(f"{snp}.tbi")
|
|
assert Path(
|
|
manifest["artifacts"]["selected_indel_vcf_index"]["full_path"]
|
|
) == Path(f"{indel}.tbi")
|
|
assert manifest["metrics"]["input_sample_ids"] == ["S1", "S2", "S3"]
|
|
assert manifest["metrics"]["output_sample_ids"] == ["S1", "S2", "S3"]
|
|
report = Path(
|
|
manifest["artifacts"]["variant_select_report"]["full_path"]
|
|
).read_text()
|
|
assert f"method: {method}" in report
|
|
assert names["gatk"] == names["bcftools"]
|
|
|
|
|
|
@pytest.mark.parametrize("method", ("gatk", "bcftools"))
|
|
def test_zero_record_variant_class_still_publishes_valid_indexed_output(
|
|
tmp_path: Path, method: str
|
|
) -> None:
|
|
from apps.cli.run_wgs_variant_select import main
|
|
|
|
workspace = tmp_path / "workspace"
|
|
joint = _joint_authority(workspace, include_indel=False)
|
|
assert main(
|
|
_argv(
|
|
workspace,
|
|
joint,
|
|
_tool(tmp_path, "gatk"),
|
|
_tool(tmp_path, "bcftools"),
|
|
method,
|
|
)
|
|
) == 0
|
|
manifest = json.loads(
|
|
(
|
|
workspace
|
|
/ "final"
|
|
/ f"Cohort.wgs_variant_select.{method}.manifest.json"
|
|
).read_text()
|
|
)
|
|
assert manifest["metrics"]["indel_records"] == 0
|
|
indel = Path(
|
|
manifest["artifacts"]["selected_indel_vcf"]["full_path"]
|
|
)
|
|
assert indel.is_file()
|
|
assert Path(f"{indel}.tbi").is_file()
|
|
|
|
|
|
def test_zero_record_gatk_output_still_queries_and_rejects_corrupt_tbi(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
from apps.cli.run_wgs_variant_select import main
|
|
|
|
workspace = tmp_path / "workspace"
|
|
joint = _joint_authority(workspace, include_indel=False)
|
|
monkeypatch.setenv("FAIL_POINT", "gatk:corrupt-index:INDEL")
|
|
assert main(
|
|
_argv(
|
|
workspace,
|
|
joint,
|
|
_tool(tmp_path, "gatk"),
|
|
Path("/bin/false"),
|
|
"gatk",
|
|
)
|
|
) == 1
|
|
|
|
|
|
def test_stale_publication_failure_never_advertises_fallback(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
from apps.cli import run_wgs_variant_select as stage
|
|
|
|
workspace = tmp_path / "workspace"
|
|
joint = _joint_authority(workspace)
|
|
monkeypatch.setattr(stage, "_publish", lambda *args, **kwargs: False)
|
|
assert stage.main(
|
|
_argv(
|
|
workspace,
|
|
joint,
|
|
_tool(tmp_path, "gatk"),
|
|
Path("/bin/false"),
|
|
"gatk",
|
|
)
|
|
) == 1
|
|
failure = json.loads(
|
|
(
|
|
workspace
|
|
/ "final"
|
|
/ "Cohort.wgs_variant_select.gatk.manifest.json"
|
|
).read_text()
|
|
)
|
|
assert failure["artifacts"] == {}
|
|
assert "fallback_method" not in failure
|
|
|
|
|
|
@pytest.mark.parametrize("method", ("gatk", "bcftools"))
|
|
def test_all_input_and_output_indices_are_queried_by_actual_method(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, method: str
|
|
) -> None:
|
|
from apps.cli.run_wgs_variant_select import main
|
|
|
|
workspace = tmp_path / "workspace"
|
|
joint = _joint_authority(workspace)
|
|
calls = tmp_path / "calls.jsonl"
|
|
monkeypatch.setenv("WGS_SELECT_CALLS", str(calls))
|
|
gatk = _tool(tmp_path, "gatk")
|
|
bcftools = _tool(tmp_path, "bcftools")
|
|
assert main(_argv(workspace, joint, gatk, bcftools, method)) == 0
|
|
observed = [json.loads(line) for line in calls.read_text().splitlines()]
|
|
if method == "gatk":
|
|
assert {call[0] for call in observed} == {"gatk"}
|
|
validations = [
|
|
call for call in observed if len(call) > 1
|
|
and call[1] == "ValidateVariants"
|
|
]
|
|
assert validations
|
|
assert all("REF" in call for call in validations)
|
|
assert sum(
|
|
call[1] == "ValidateVariants" for call in observed
|
|
) == 3
|
|
else:
|
|
assert {call[0] for call in observed} == {"bcftools"}
|
|
assert sum(
|
|
call[1] == "view" and "-R" in call for call in observed
|
|
) == 3
|
|
assert sum(call[1] == "stats" for call in observed) == 2
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("method", "inactive_tool"),
|
|
(("gatk", "bcftools"), ("bcftools", "gatk")),
|
|
)
|
|
def test_method_does_not_execute_or_require_inactive_fallback_tool(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
method: str,
|
|
inactive_tool: str,
|
|
) -> None:
|
|
from apps.cli.run_wgs_variant_select import main
|
|
|
|
workspace = tmp_path / "workspace"
|
|
joint = _joint_authority(workspace)
|
|
calls = tmp_path / "calls.jsonl"
|
|
monkeypatch.setenv("WGS_SELECT_CALLS", str(calls))
|
|
gatk = Path("/bin/false") if inactive_tool == "gatk" else _tool(
|
|
tmp_path, "gatk"
|
|
)
|
|
bcftools = (
|
|
Path("/bin/false")
|
|
if inactive_tool == "bcftools"
|
|
else _tool(tmp_path, "bcftools")
|
|
)
|
|
assert main(_argv(workspace, joint, gatk, bcftools, method)) == 0
|
|
observed = [json.loads(line) for line in calls.read_text().splitlines()]
|
|
assert {call[0] for call in observed} == {method}
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("method", "failure"),
|
|
(
|
|
("gatk", "gatk:select:SNP"),
|
|
("gatk", "gatk:index:INDEL"),
|
|
("gatk", "gatk:validate:SNP"),
|
|
("gatk", "gatk:validate:input"),
|
|
("bcftools", "bcftools:select:SNP"),
|
|
("bcftools", "bcftools:index:INDEL"),
|
|
("bcftools", "bcftools:validate:SNP"),
|
|
("bcftools", "bcftools:stats:INDEL"),
|
|
),
|
|
)
|
|
def test_each_select_index_validate_and_stat_failure_preserves_method_authority(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
method: str,
|
|
failure: str,
|
|
) -> None:
|
|
from apps.cli.run_wgs_variant_select import main
|
|
|
|
workspace = tmp_path / "workspace"
|
|
joint = _joint_authority(workspace)
|
|
authority = (
|
|
workspace
|
|
/ "final"
|
|
/ f"Cohort.wgs_variant_select.{method}.manifest.json"
|
|
)
|
|
authority.write_text(
|
|
json.dumps(
|
|
{
|
|
"stage": "wgs_variant_select",
|
|
"status": "success",
|
|
"method": method,
|
|
"sentinel": True,
|
|
}
|
|
)
|
|
)
|
|
before = hashlib.sha256(authority.read_bytes()).hexdigest()
|
|
monkeypatch.setenv("FAIL_POINT", failure)
|
|
assert main(
|
|
_argv(
|
|
workspace,
|
|
joint,
|
|
_tool(tmp_path, "gatk"),
|
|
_tool(tmp_path, "bcftools"),
|
|
method,
|
|
)
|
|
) == 1
|
|
assert hashlib.sha256(authority.read_bytes()).hexdigest() == before
|
|
sidecars = list(
|
|
(workspace / "final").glob(
|
|
f"Cohort.wgs_variant_select.{method}.failed.*.manifest.json"
|
|
)
|
|
)
|
|
assert len(sidecars) == 1
|
|
failure_manifest = json.loads(sidecars[0].read_text())
|
|
assert failure_manifest["artifacts"] == {}
|
|
assert failure_manifest["method"] == method
|
|
|
|
|
|
def test_gatk_failure_does_not_automatically_run_bcftools_selection(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
from apps.cli.run_wgs_variant_select import main
|
|
|
|
workspace = tmp_path / "workspace"
|
|
joint = _joint_authority(workspace)
|
|
calls = tmp_path / "calls.jsonl"
|
|
monkeypatch.setenv("WGS_SELECT_CALLS", str(calls))
|
|
monkeypatch.setenv("FAIL_POINT", "gatk:select:SNP")
|
|
assert main(
|
|
_argv(
|
|
workspace,
|
|
joint,
|
|
_tool(tmp_path, "gatk"),
|
|
_tool(tmp_path, "bcftools"),
|
|
"gatk",
|
|
)
|
|
) == 1
|
|
observed = [json.loads(line) for line in calls.read_text().splitlines()]
|
|
assert not any(
|
|
call[0] == "bcftools"
|
|
and call[1] == "view"
|
|
and "-v" in call
|
|
for call in observed
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"mutation", ("analysis", "hash", "path", "symlink", "roles", "sidecar")
|
|
)
|
|
def test_joint_authority_is_strictly_bound(
|
|
tmp_path: Path, mutation: str
|
|
) -> None:
|
|
from apps.cli.run_wgs_variant_select import main
|
|
|
|
workspace = tmp_path / "workspace"
|
|
joint = _joint_authority(workspace)
|
|
payload = json.loads(joint.read_text())
|
|
if mutation == "analysis":
|
|
payload.pop("wgs_analysis_id")
|
|
elif mutation == "hash":
|
|
payload["artifacts"]["joint_vcf"]["sha256"] = "f" * 64
|
|
elif mutation == "path":
|
|
payload["artifacts"]["joint_vcf"]["path"] = "../escape.vcf.gz"
|
|
elif mutation == "symlink":
|
|
record = payload["artifacts"]["joint_vcf"]
|
|
original = Path(record["full_path"])
|
|
link = original.with_name("linked.vcf.gz")
|
|
link.symlink_to(original)
|
|
record["path"] = str(link.relative_to(workspace))
|
|
record["full_path"] = str(link)
|
|
elif mutation == "roles":
|
|
payload["artifacts"].pop("joint_vcf_index")
|
|
else:
|
|
record = payload["artifacts"]["joint_vcf_index"]
|
|
unrelated = Path(record["full_path"]).with_name("unrelated.tbi")
|
|
unrelated.write_bytes(b"TBI:unrelated")
|
|
record.update(_record(workspace, unrelated, "joint_vcf_index"))
|
|
joint.write_text(json.dumps(payload))
|
|
assert main(
|
|
_argv(
|
|
workspace,
|
|
joint,
|
|
_tool(tmp_path, "gatk"),
|
|
_tool(tmp_path, "bcftools"),
|
|
)
|
|
) == 1
|
|
|
|
|
|
def test_corrupt_nonempty_joint_tbi_is_rejected_by_real_query_boundary(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
from apps.cli.run_wgs_variant_select import main
|
|
|
|
workspace = tmp_path / "workspace"
|
|
joint = _joint_authority(workspace)
|
|
payload = json.loads(joint.read_text())
|
|
record = payload["artifacts"]["joint_vcf_index"]
|
|
tbi = Path(record["full_path"])
|
|
tbi.write_bytes(b"CORRUPT:input")
|
|
record.update(_record(workspace, tbi, "joint_vcf_index"))
|
|
joint.write_text(json.dumps(payload))
|
|
assert main(
|
|
_argv(
|
|
workspace,
|
|
joint,
|
|
_tool(tmp_path, "gatk"),
|
|
_tool(tmp_path, "bcftools"),
|
|
)
|
|
) == 1
|
|
|
|
|
|
def test_method_publish_lock_rejects_equal_or_older_attempt(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
from apps.cli.run_wgs_variant_select import _publish
|
|
|
|
authority = tmp_path / "final" / "Cohort.wgs_variant_select.gatk.manifest.json"
|
|
authority.parent.mkdir()
|
|
timestamp = "2026-07-23T12:00:00Z"
|
|
authority.write_text(
|
|
json.dumps({"status": "success", "start_time": timestamp})
|
|
)
|
|
for number, started_at in enumerate(
|
|
("2026-07-23T11:00:00Z", timestamp)
|
|
):
|
|
bundle = tmp_path / f"bundle-{number}"
|
|
bundle.mkdir()
|
|
manifest = tmp_path / f"manifest-{number}.json"
|
|
manifest.write_text("{}")
|
|
assert not _publish(
|
|
bundle,
|
|
tmp_path / "runs" / str(number),
|
|
manifest,
|
|
authority,
|
|
started_at,
|
|
)
|
|
|
|
|
|
def test_concurrent_publish_keeps_newest_complete_method_authority(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
from apps.cli.run_wgs_variant_select import _publish
|
|
|
|
authority = tmp_path / "final" / "Cohort.wgs_variant_select.gatk.manifest.json"
|
|
authority.parent.mkdir()
|
|
barrier = Barrier(2)
|
|
|
|
def worker(name: str, started_at: str) -> None:
|
|
bundle = tmp_path / f"bundle-{name}"
|
|
bundle.mkdir()
|
|
manifest = tmp_path / f"manifest-{name}.json"
|
|
manifest.write_text(
|
|
json.dumps(
|
|
{"status": "success", "start_time": started_at, "winner": name}
|
|
)
|
|
)
|
|
barrier.wait()
|
|
_publish(
|
|
bundle,
|
|
tmp_path / "runs" / name,
|
|
manifest,
|
|
authority,
|
|
started_at,
|
|
)
|
|
|
|
threads = [
|
|
Thread(target=worker, args=("a", "2026-07-23T11:00:00Z")),
|
|
Thread(target=worker, args=("b", "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"
|
|
|
|
|
|
def test_status_and_results_select_exact_method_authority_and_show_method(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
import asyncio
|
|
|
|
from apps.cli.run_wgs_variant_select 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"
|
|
joint = _joint_authority(workspace)
|
|
assert main(
|
|
_argv(
|
|
workspace,
|
|
joint,
|
|
_tool(tmp_path, "gatk"),
|
|
Path("/bin/false"),
|
|
"gatk",
|
|
)
|
|
) == 0
|
|
authority = (
|
|
workspace
|
|
/ "final"
|
|
/ "Cohort.wgs_variant_select.gatk.manifest.json"
|
|
)
|
|
manifest = json.loads(authority.read_text())
|
|
analysis_id = manifest["wgs_analysis_id"]
|
|
status = get_stage_status(
|
|
"job-1",
|
|
"Cohort",
|
|
"wgs_variant_select",
|
|
workspace_root,
|
|
wgs_sample_set_id=SAMPLES,
|
|
wgs_reference_id=REFERENCE,
|
|
wgs_analysis_id=analysis_id,
|
|
)
|
|
assert status.completed
|
|
assert status.manifest_path == authority
|
|
assert status.manifest["method"] == "gatk"
|
|
assert not get_stage_status(
|
|
"job-1",
|
|
"Cohort",
|
|
"wgs_variant_select",
|
|
workspace_root,
|
|
wgs_sample_set_id=SAMPLES,
|
|
wgs_reference_id=REFERENCE,
|
|
wgs_analysis_id="wgs-analysis:" + "f" * 64,
|
|
).completed
|
|
monkeypatch.setattr(web_main, "WORKSPACE_ROOT", workspace_root)
|
|
results = asyncio.run(
|
|
web_main.get_results(
|
|
job_id="job-1",
|
|
stage="wgs_variant_select",
|
|
wgs_sample_set_id=SAMPLES,
|
|
wgs_reference_id=REFERENCE,
|
|
wgs_analysis_id=analysis_id,
|
|
)
|
|
)
|
|
serialized = json.dumps(results)
|
|
assert '"method": "gatk"' in serialized
|
|
assert "selected_snp_vcf" in serialized
|
|
assert "selected_indel_vcf" in serialized
|