398 lines
14 KiB
Python
398 lines
14 KiB
Python
|
import hashlib
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
import yaml
|
||
|
|
|
||
|
|
from apps.web import pipeline_backend
|
||
|
|
|
||
|
|
|
||
|
|
pytestmark = pytest.mark.wgs_tools
|
||
|
|
|
||
|
|
FIXTURE = (
|
||
|
|
Path(__file__).resolve().parents[1]
|
||
|
|
/ "fixtures"
|
||
|
|
/ "wgs_three_sample"
|
||
|
|
/ "wgs_samples.yaml"
|
||
|
|
)
|
||
|
|
REFERENCE = Path("/tmp/wgs-impute-real-legal.rHngBE/reference.9.1m.fasta")
|
||
|
|
EXPECTED_ROLES = {
|
||
|
|
"wgs_qc": {
|
||
|
|
"wgs_sample_config", "clean_reads_r1", "clean_reads_r2",
|
||
|
|
"fastp_html", "fastp_json",
|
||
|
|
},
|
||
|
|
"wgs_index": {
|
||
|
|
"reference_fasta", "reference_fai", "reference_dict",
|
||
|
|
"bwa_mem2_index_component",
|
||
|
|
},
|
||
|
|
"wgs_align": {"sorted_bam", "sorted_bam_index", "align_log"},
|
||
|
|
"wgs_markdup": {
|
||
|
|
"markdup_bam", "markdup_bam_index", "markdup_metrics", "markdup_log",
|
||
|
|
},
|
||
|
|
"wgs_bam_qc": {
|
||
|
|
"bam_flagstat", "bam_depth_summary", "bam_qc_summary", "bam_qc_report",
|
||
|
|
},
|
||
|
|
"wgs_haplotype_calling": {"gvcf", "gvcf_index", "haplotype_log"},
|
||
|
|
"wgs_gvcf_combine": {
|
||
|
|
"combined_gvcf", "combined_gvcf_index", "combine_log",
|
||
|
|
},
|
||
|
|
"wgs_joint_genotyping": {
|
||
|
|
"joint_vcf", "joint_vcf_index", "joint_summary", "joint_log",
|
||
|
|
},
|
||
|
|
"wgs_variant_select": {
|
||
|
|
"selected_snp_vcf", "selected_snp_vcf_index",
|
||
|
|
"selected_indel_vcf", "selected_indel_vcf_index",
|
||
|
|
"variant_select_report",
|
||
|
|
},
|
||
|
|
"wgs_variant_filter": {
|
||
|
|
"filtered_snp_vcf", "filtered_snp_vcf_index",
|
||
|
|
"pass_snp_vcf", "pass_snp_vcf_index",
|
||
|
|
"filtered_indel_vcf", "filtered_indel_vcf_index",
|
||
|
|
"pass_indel_vcf", "pass_indel_vcf_index",
|
||
|
|
"variant_filter_report",
|
||
|
|
},
|
||
|
|
"wgs_impute": {
|
||
|
|
"imputed_snp_vcf", "imputed_snp_vcf_index",
|
||
|
|
"impute_report", "impute_log",
|
||
|
|
},
|
||
|
|
"wgs_maf_missing_filter": {
|
||
|
|
"final_snp_vcf", "final_snp_vcf_index",
|
||
|
|
"maf_missing_report", "maf_missing_stats",
|
||
|
|
},
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def test_wgs_three_sample_fixture_is_explicit_and_required() -> None:
|
||
|
|
document = yaml.safe_load(FIXTURE.read_text(encoding="utf-8"))
|
||
|
|
assert sorted(document) == ["samples"]
|
||
|
|
assert sorted(document["samples"]) == ["102", "108", "112"]
|
||
|
|
assert "include" not in FIXTURE.read_text(encoding="utf-8")
|
||
|
|
assert "exclude" not in FIXTURE.read_text(encoding="utf-8")
|
||
|
|
reference_header = REFERENCE.open(encoding="utf-8").readline().strip()
|
||
|
|
assert reference_header == ">9"
|
||
|
|
assert ":" not in reference_header
|
||
|
|
|
||
|
|
|
||
|
|
def _sha256(path: Path) -> str:
|
||
|
|
digest = hashlib.sha256()
|
||
|
|
with path.open("rb") as handle:
|
||
|
|
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
||
|
|
digest.update(chunk)
|
||
|
|
return digest.hexdigest()
|
||
|
|
|
||
|
|
|
||
|
|
def _assert_artifacts(workspace: Path, manifest: dict) -> None:
|
||
|
|
assert manifest["artifacts"]
|
||
|
|
for record in manifest["artifacts"].values():
|
||
|
|
path = (workspace / record["path"]).resolve()
|
||
|
|
path.relative_to(workspace.resolve())
|
||
|
|
assert path.is_file()
|
||
|
|
assert path.stat().st_size == record["size"]
|
||
|
|
assert _sha256(path) == record["sha256"]
|
||
|
|
|
||
|
|
|
||
|
|
def _run(
|
||
|
|
stage: str,
|
||
|
|
job: pipeline_backend.JobInfo,
|
||
|
|
workspace: Path,
|
||
|
|
**kwargs: object,
|
||
|
|
) -> tuple[Path, dict]:
|
||
|
|
message, manifest_path = pipeline_backend.run_stage(
|
||
|
|
stage=stage,
|
||
|
|
job=job,
|
||
|
|
prefix="ThreeSample",
|
||
|
|
**kwargs,
|
||
|
|
)
|
||
|
|
assert "completed" in message.lower(), message
|
||
|
|
assert manifest_path is not None
|
||
|
|
expected = workspace / "final" / manifest_path.name
|
||
|
|
assert manifest_path == expected
|
||
|
|
manifest = json.loads(expected.read_text(encoding="utf-8"))
|
||
|
|
assert manifest["stage"] == stage
|
||
|
|
assert manifest["status"] == "success"
|
||
|
|
assert manifest["job_id"] == job.job_id
|
||
|
|
assert manifest["prefix"] == "ThreeSample"
|
||
|
|
assert manifest["wgs_analysis_id"].startswith("wgs-analysis:")
|
||
|
|
assert {record["role"] for record in manifest["artifacts"].values()} == (
|
||
|
|
EXPECTED_ROLES[stage]
|
||
|
|
)
|
||
|
|
_assert_artifacts(workspace, manifest)
|
||
|
|
return manifest_path, manifest
|
||
|
|
|
||
|
|
|
||
|
|
def test_wgs_three_sample_stage01_to_stage12_production_chain(
|
||
|
|
tmp_path: Path,
|
||
|
|
monkeypatch: pytest.MonkeyPatch,
|
||
|
|
) -> None:
|
||
|
|
monkeypatch.setenv(
|
||
|
|
"PATH",
|
||
|
|
"/home/xia-lab/miniconda3/envs/rnaseq/bin:"
|
||
|
|
"/home/xia-lab/miniconda3/envs/ASAW/bin:"
|
||
|
|
f"{os.environ['PATH']}",
|
||
|
|
)
|
||
|
|
workspace = tmp_path / "workspace"
|
||
|
|
job = pipeline_backend.JobInfo(
|
||
|
|
"wgs-three-sample",
|
||
|
|
str(FIXTURE.parent),
|
||
|
|
str(workspace),
|
||
|
|
str(workspace / "logs"),
|
||
|
|
"now",
|
||
|
|
)
|
||
|
|
manifests: dict[str, Path] = {}
|
||
|
|
data: dict[str, dict] = {}
|
||
|
|
common_tools = {
|
||
|
|
"wgs_fastp": "/usr/bin/fastp",
|
||
|
|
"wgs_samtools": "/home/xia-lab/miniconda3/envs/rnaseq/bin/samtools",
|
||
|
|
"wgs_bwa_mem2": "/home/xia-lab/miniconda3/envs/ASAW/bin/bwa-mem2",
|
||
|
|
"wgs_gatk": "/tmp/wgs-index-gatk-wrapper.sh",
|
||
|
|
"wgs_bcftools": "/tmp/wgs-task11-bcftools-wrapper.sh",
|
||
|
|
"wgs_java": "/home/xia-lab/miniconda3/envs/ASAW/bin/java",
|
||
|
|
"wgs_beagle_jar": "/tmp/beagle.29Oct24.c8e.jar",
|
||
|
|
"wgs_threads": 2,
|
||
|
|
"wgs_max_jobs": 3,
|
||
|
|
}
|
||
|
|
|
||
|
|
manifests["wgs_qc"], data["wgs_qc"] = _run(
|
||
|
|
"wgs_qc",
|
||
|
|
job,
|
||
|
|
workspace,
|
||
|
|
wgs_sample_config=str(FIXTURE),
|
||
|
|
**common_tools,
|
||
|
|
)
|
||
|
|
assert data["wgs_qc"]["metrics"]["output_sample_ids"] == [
|
||
|
|
"102",
|
||
|
|
"108",
|
||
|
|
"112",
|
||
|
|
]
|
||
|
|
sample_set_id = data["wgs_qc"]["wgs_sample_set_id"]
|
||
|
|
assert "wgs_reference_id" not in data["wgs_qc"]
|
||
|
|
|
||
|
|
manifests["wgs_index"], data["wgs_index"] = _run(
|
||
|
|
"wgs_index",
|
||
|
|
job,
|
||
|
|
workspace,
|
||
|
|
wgs_reference_fasta=str(REFERENCE),
|
||
|
|
**common_tools,
|
||
|
|
)
|
||
|
|
reference_id = data["wgs_index"]["wgs_reference_id"]
|
||
|
|
assert "wgs_sample_set_id" not in data["wgs_index"]
|
||
|
|
analysis_ids = {
|
||
|
|
stage: manifest["wgs_analysis_id"] for stage, manifest in data.items()
|
||
|
|
}
|
||
|
|
|
||
|
|
manifests["wgs_align"], data["wgs_align"] = _run(
|
||
|
|
"wgs_align",
|
||
|
|
job,
|
||
|
|
workspace,
|
||
|
|
wgs_qc_manifest=str(manifests["wgs_qc"]),
|
||
|
|
wgs_index_manifest=str(manifests["wgs_index"]),
|
||
|
|
wgs_sample_set_id=sample_set_id,
|
||
|
|
wgs_reference_id=reference_id,
|
||
|
|
**common_tools,
|
||
|
|
)
|
||
|
|
analysis_ids["wgs_align"] = data["wgs_align"]["wgs_analysis_id"]
|
||
|
|
|
||
|
|
manifests["wgs_markdup"], data["wgs_markdup"] = _run(
|
||
|
|
"wgs_markdup",
|
||
|
|
job,
|
||
|
|
workspace,
|
||
|
|
wgs_align_manifest=str(manifests["wgs_align"]),
|
||
|
|
wgs_sample_set_id=sample_set_id,
|
||
|
|
wgs_reference_id=reference_id,
|
||
|
|
wgs_analysis_id=analysis_ids["wgs_align"],
|
||
|
|
**common_tools,
|
||
|
|
)
|
||
|
|
assert (
|
||
|
|
data["wgs_markdup"]["upstream_wgs_analysis_id"]
|
||
|
|
== analysis_ids["wgs_align"]
|
||
|
|
)
|
||
|
|
analysis_ids["wgs_markdup"] = data["wgs_markdup"]["wgs_analysis_id"]
|
||
|
|
|
||
|
|
manifests["wgs_bam_qc"], data["wgs_bam_qc"] = _run(
|
||
|
|
"wgs_bam_qc",
|
||
|
|
job,
|
||
|
|
workspace,
|
||
|
|
wgs_markdup_manifest=str(manifests["wgs_markdup"]),
|
||
|
|
wgs_index_manifest=str(manifests["wgs_index"]),
|
||
|
|
wgs_sample_set_id=sample_set_id,
|
||
|
|
wgs_reference_id=reference_id,
|
||
|
|
wgs_analysis_id=analysis_ids["wgs_markdup"],
|
||
|
|
wgs_index_analysis_id=analysis_ids["wgs_index"],
|
||
|
|
**common_tools,
|
||
|
|
)
|
||
|
|
analysis_ids["wgs_bam_qc"] = data["wgs_bam_qc"]["wgs_analysis_id"]
|
||
|
|
assert data["wgs_bam_qc"]["metrics"]["input_sample_ids"] == [
|
||
|
|
"102",
|
||
|
|
"108",
|
||
|
|
"112",
|
||
|
|
]
|
||
|
|
assert data["wgs_bam_qc"]["metrics"]["output_sample_ids"] == [
|
||
|
|
"102",
|
||
|
|
"108",
|
||
|
|
"112",
|
||
|
|
]
|
||
|
|
|
||
|
|
manifests["wgs_haplotype_calling"], data["wgs_haplotype_calling"] = _run(
|
||
|
|
"wgs_haplotype_calling",
|
||
|
|
job,
|
||
|
|
workspace,
|
||
|
|
wgs_markdup_manifest=str(manifests["wgs_markdup"]),
|
||
|
|
wgs_bam_qc_manifest=str(manifests["wgs_bam_qc"]),
|
||
|
|
wgs_index_manifest=str(manifests["wgs_index"]),
|
||
|
|
wgs_sample_set_id=sample_set_id,
|
||
|
|
wgs_reference_id=reference_id,
|
||
|
|
wgs_markdup_analysis_id=analysis_ids["wgs_markdup"],
|
||
|
|
wgs_bam_qc_analysis_id=analysis_ids["wgs_bam_qc"],
|
||
|
|
wgs_index_analysis_id=analysis_ids["wgs_index"],
|
||
|
|
**common_tools,
|
||
|
|
)
|
||
|
|
analysis_ids["wgs_haplotype_calling"] = data[
|
||
|
|
"wgs_haplotype_calling"
|
||
|
|
]["wgs_analysis_id"]
|
||
|
|
|
||
|
|
manifests["wgs_gvcf_combine"], data["wgs_gvcf_combine"] = _run(
|
||
|
|
"wgs_gvcf_combine",
|
||
|
|
job,
|
||
|
|
workspace,
|
||
|
|
wgs_haplotype_manifest=str(manifests["wgs_haplotype_calling"]),
|
||
|
|
wgs_index_manifest=str(manifests["wgs_index"]),
|
||
|
|
wgs_sample_set_id=sample_set_id,
|
||
|
|
wgs_reference_id=reference_id,
|
||
|
|
wgs_haplotype_analysis_id=analysis_ids["wgs_haplotype_calling"],
|
||
|
|
wgs_index_analysis_id=analysis_ids["wgs_index"],
|
||
|
|
**common_tools,
|
||
|
|
)
|
||
|
|
analysis_ids["wgs_gvcf_combine"] = data["wgs_gvcf_combine"][
|
||
|
|
"wgs_analysis_id"
|
||
|
|
]
|
||
|
|
|
||
|
|
manifests["wgs_joint_genotyping"], data["wgs_joint_genotyping"] = _run(
|
||
|
|
"wgs_joint_genotyping",
|
||
|
|
job,
|
||
|
|
workspace,
|
||
|
|
wgs_combine_manifest=str(manifests["wgs_gvcf_combine"]),
|
||
|
|
wgs_index_manifest=str(manifests["wgs_index"]),
|
||
|
|
wgs_sample_set_id=sample_set_id,
|
||
|
|
wgs_reference_id=reference_id,
|
||
|
|
wgs_combine_analysis_id=analysis_ids["wgs_gvcf_combine"],
|
||
|
|
wgs_index_analysis_id=analysis_ids["wgs_index"],
|
||
|
|
**common_tools,
|
||
|
|
)
|
||
|
|
analysis_ids["wgs_joint_genotyping"] = data["wgs_joint_genotyping"][
|
||
|
|
"wgs_analysis_id"
|
||
|
|
]
|
||
|
|
|
||
|
|
manifests["wgs_variant_select"], data["wgs_variant_select"] = _run(
|
||
|
|
"wgs_variant_select",
|
||
|
|
job,
|
||
|
|
workspace,
|
||
|
|
wgs_joint_manifest=str(manifests["wgs_joint_genotyping"]),
|
||
|
|
wgs_sample_set_id=sample_set_id,
|
||
|
|
wgs_reference_id=reference_id,
|
||
|
|
wgs_joint_analysis_id=analysis_ids["wgs_joint_genotyping"],
|
||
|
|
wgs_method="gatk",
|
||
|
|
**common_tools,
|
||
|
|
)
|
||
|
|
assert data["wgs_variant_select"]["method"] == "gatk"
|
||
|
|
assert data["wgs_variant_select"]["parameters"]["method"] == "gatk"
|
||
|
|
analysis_ids["wgs_variant_select"] = data["wgs_variant_select"][
|
||
|
|
"wgs_analysis_id"
|
||
|
|
]
|
||
|
|
|
||
|
|
manifests["wgs_variant_filter"], data["wgs_variant_filter"] = _run(
|
||
|
|
"wgs_variant_filter",
|
||
|
|
job,
|
||
|
|
workspace,
|
||
|
|
wgs_select_manifest=str(manifests["wgs_variant_select"]),
|
||
|
|
wgs_sample_set_id=sample_set_id,
|
||
|
|
wgs_reference_id=reference_id,
|
||
|
|
wgs_select_analysis_id=analysis_ids["wgs_variant_select"],
|
||
|
|
wgs_method="gatk",
|
||
|
|
**common_tools,
|
||
|
|
)
|
||
|
|
assert data["wgs_variant_filter"]["method"] == "gatk"
|
||
|
|
assert data["wgs_variant_filter"]["parameters"]["method"] == "gatk"
|
||
|
|
analysis_ids["wgs_variant_filter"] = data["wgs_variant_filter"][
|
||
|
|
"wgs_analysis_id"
|
||
|
|
]
|
||
|
|
|
||
|
|
manifests["wgs_impute"], data["wgs_impute"] = _run(
|
||
|
|
"wgs_impute",
|
||
|
|
job,
|
||
|
|
workspace,
|
||
|
|
wgs_filter_manifest=str(manifests["wgs_variant_filter"]),
|
||
|
|
wgs_sample_set_id=sample_set_id,
|
||
|
|
wgs_reference_id=reference_id,
|
||
|
|
wgs_filter_analysis_id=analysis_ids["wgs_variant_filter"],
|
||
|
|
**common_tools,
|
||
|
|
)
|
||
|
|
analysis_ids["wgs_impute"] = data["wgs_impute"]["wgs_analysis_id"]
|
||
|
|
site_keys = data["wgs_impute"]["metrics"]["output_site_keys"]
|
||
|
|
assert site_keys
|
||
|
|
assert site_keys == data["wgs_impute"]["metrics"]["input_site_keys"]
|
||
|
|
assert [key[1] for key in site_keys] == sorted(key[1] for key in site_keys)
|
||
|
|
|
||
|
|
manifests["wgs_maf_missing_filter"], data[
|
||
|
|
"wgs_maf_missing_filter"
|
||
|
|
] = _run(
|
||
|
|
"wgs_maf_missing_filter",
|
||
|
|
job,
|
||
|
|
workspace,
|
||
|
|
wgs_impute_manifest=str(manifests["wgs_impute"]),
|
||
|
|
wgs_sample_set_id=sample_set_id,
|
||
|
|
wgs_reference_id=reference_id,
|
||
|
|
wgs_impute_analysis_id=analysis_ids["wgs_impute"],
|
||
|
|
**common_tools,
|
||
|
|
)
|
||
|
|
analysis_ids["wgs_maf_missing_filter"] = data[
|
||
|
|
"wgs_maf_missing_filter"
|
||
|
|
]["wgs_analysis_id"]
|
||
|
|
assert data["wgs_maf_missing_filter"]["metrics"]["retained_sites"] >= 1
|
||
|
|
assert data["wgs_maf_missing_filter"]["method"] == "bcftools"
|
||
|
|
|
||
|
|
assert len(set(analysis_ids.values())) == len(analysis_ids)
|
||
|
|
assert data["wgs_bam_qc"]["upstream_wgs_analysis_ids"] == {
|
||
|
|
"wgs_markdup": analysis_ids["wgs_markdup"],
|
||
|
|
"wgs_index": analysis_ids["wgs_index"],
|
||
|
|
}
|
||
|
|
assert data["wgs_haplotype_calling"]["upstream_wgs_analysis_ids"] == {
|
||
|
|
"wgs_markdup": analysis_ids["wgs_markdup"],
|
||
|
|
"wgs_bam_qc": analysis_ids["wgs_bam_qc"],
|
||
|
|
"wgs_index": analysis_ids["wgs_index"],
|
||
|
|
}
|
||
|
|
assert data["wgs_gvcf_combine"]["upstream_wgs_analysis_ids"] == {
|
||
|
|
"wgs_haplotype_calling": analysis_ids["wgs_haplotype_calling"],
|
||
|
|
"wgs_index": analysis_ids["wgs_index"],
|
||
|
|
}
|
||
|
|
assert data["wgs_joint_genotyping"]["upstream_wgs_analysis_ids"] == {
|
||
|
|
"wgs_gvcf_combine": analysis_ids["wgs_gvcf_combine"],
|
||
|
|
"wgs_index": analysis_ids["wgs_index"],
|
||
|
|
}
|
||
|
|
for consumer, producer in (
|
||
|
|
("wgs_variant_select", "wgs_joint_genotyping"),
|
||
|
|
("wgs_variant_filter", "wgs_variant_select"),
|
||
|
|
("wgs_impute", "wgs_variant_filter"),
|
||
|
|
("wgs_maf_missing_filter", "wgs_impute"),
|
||
|
|
):
|
||
|
|
assert data[consumer]["upstream_wgs_analysis_ids"] == {
|
||
|
|
producer: analysis_ids[producer]
|
||
|
|
}
|
||
|
|
for stage, manifest in data.items():
|
||
|
|
if stage != "wgs_index":
|
||
|
|
assert manifest["wgs_sample_set_id"] == sample_set_id
|
||
|
|
if stage != "wgs_qc":
|
||
|
|
assert manifest["wgs_reference_id"] == reference_id
|
||
|
|
if stage not in {"wgs_qc", "wgs_index"}:
|
||
|
|
assert manifest["metrics"]["input_sample_ids"] == [
|
||
|
|
"102",
|
||
|
|
"108",
|
||
|
|
"112",
|
||
|
|
]
|
||
|
|
assert manifest["metrics"]["output_sample_ids"] == [
|
||
|
|
"102",
|
||
|
|
"108",
|
||
|
|
"112",
|
||
|
|
]
|