224 lines
9.4 KiB
Python
224 lines
9.4 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import FrozenInstanceError
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from gfa_core import wgs_upstream_manifest as wgs_manifest
|
|
from gfa_core.wgs_upstream_manifest import (
|
|
ResolvedWgsArtifact,
|
|
WgsUpstreamManifestError,
|
|
resolve_wgs_artifacts,
|
|
)
|
|
|
|
|
|
RUN_ID = "a" * 32
|
|
SAMPLE_SET_ID = "wgs-samples:" + "b" * 64
|
|
REFERENCE_ID = "wgs-reference:" + "c" * 64
|
|
|
|
|
|
def _artifact(workspace: Path, bundle: Path, key: str, role: str, *, sample_id: str = "", component: str = "") -> dict:
|
|
path = bundle / key
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(key.encode())
|
|
record = {
|
|
"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:
|
|
record["sample_id"] = sample_id
|
|
if component:
|
|
record["component"] = component
|
|
return record
|
|
|
|
|
|
def _qc_manifest(workspace: Path) -> tuple[Path, dict]:
|
|
bundle = workspace / "final" / ".wgs_qc_runs" / "Cohort" / RUN_ID
|
|
artifacts = {}
|
|
for sample_id in ("S1", "S2", "S3"):
|
|
for role, suffix in (("clean_reads_r1", "R1.fastq.gz"), ("clean_reads_r2", "R2.fastq.gz")):
|
|
key = f"{role}.{sample_id}"
|
|
artifacts[key] = _artifact(workspace, bundle, f"Cohort.wgs_qc.{sample_id}.{suffix}", role, sample_id=sample_id)
|
|
manifest = {
|
|
"manifest_version": "1.0", "stage": "wgs_qc", "status": "success",
|
|
"prefix": "Cohort", "job_id": "job-1", "run_id": RUN_ID, "workspace": str(workspace.resolve()),
|
|
"wgs_sample_set_id": SAMPLE_SET_ID, "wgs_analysis_id": "wgs-analysis:" + "d" * 64,
|
|
"artifacts": artifacts,
|
|
}
|
|
path = workspace / "final" / "Cohort.wgs_qc.manifest.json"
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(manifest), encoding="utf-8")
|
|
return path, manifest
|
|
|
|
|
|
def _resolve(workspace: Path, manifest_path: Path, **overrides):
|
|
args = {
|
|
"manifest_path": manifest_path, "expected_stage": "wgs_qc",
|
|
"required_roles": ("clean_reads_r1", "clean_reads_r2"),
|
|
"expected_sample_set_id": SAMPLE_SET_ID,
|
|
}
|
|
args.update(overrides)
|
|
return resolve_wgs_artifacts(**args)
|
|
|
|
|
|
def test_resolver_returns_frozen_artifacts_grouped_in_immutable_tuples(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
manifest_path, _ = _qc_manifest(workspace)
|
|
resolved = _resolve(workspace, manifest_path)
|
|
assert set(resolved) == {"clean_reads_r1", "clean_reads_r2"}
|
|
assert all(isinstance(group, tuple) and len(group) == 3 for group in resolved.values())
|
|
first = resolved["clean_reads_r1"][0]
|
|
assert isinstance(first, ResolvedWgsArtifact)
|
|
assert (first.key, first.role, first.sample_id, first.component) == (
|
|
"clean_reads_r1.S1", "clean_reads_r1", "S1", "",
|
|
)
|
|
with pytest.raises(FrozenInstanceError):
|
|
first.role = "changed" # type: ignore[misc]
|
|
|
|
|
|
def test_workspace_identity_accepts_mount_alias_for_same_job_only() -> None:
|
|
assert wgs_manifest.wgs_workspace_identity_matches(
|
|
"/tmp/custom-job-root/job-1",
|
|
"/tmp/custom-job-root/job-1",
|
|
)
|
|
assert wgs_manifest.wgs_workspace_identity_matches(
|
|
"/home/xzq/project/workspace/job-1",
|
|
"/home/xia-lab/remote-projects/project/workspace/job-1",
|
|
)
|
|
assert wgs_manifest.wgs_workspace_identity_matches(
|
|
"workspace/job-1",
|
|
"/home/xia-lab/remote-projects/project/workspace/job-1",
|
|
)
|
|
assert not wgs_manifest.wgs_workspace_identity_matches(
|
|
"/home/xzq/project/workspace/job-2",
|
|
"/home/xia-lab/remote-projects/project/workspace/job-1",
|
|
)
|
|
|
|
|
|
def test_resolver_rebases_relocated_mount_paths_to_current_workspace(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
workspace = tmp_path / "current" / "workspace" / "job-1"
|
|
manifest_path, manifest = _qc_manifest(workspace)
|
|
old_workspace = Path("/old/mount/project/workspace/job-1")
|
|
manifest["workspace"] = str(old_workspace)
|
|
for artifact in manifest["artifacts"].values():
|
|
artifact["full_path"] = str(old_workspace / artifact["path"])
|
|
manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
|
|
|
|
resolved = _resolve(workspace, manifest_path)
|
|
|
|
assert all(
|
|
item.path.is_relative_to(workspace.resolve())
|
|
for group in resolved.values()
|
|
for item in group
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("field", "value", "message"),
|
|
[("stage", "wgs_index", "producer"), ("status", "failed", "successful"),
|
|
("prefix", "Other", "prefix"), ("workspace", "/tmp/not-authority", "workspace"),
|
|
("wgs_sample_set_id", "wgs-samples:" + "0" * 64, "identity")],
|
|
)
|
|
def test_resolver_rejects_wrong_producer_status_scope_and_identity(tmp_path: Path, field: str, value: str, message: str) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
path, manifest = _qc_manifest(workspace)
|
|
manifest[field] = value
|
|
path.write_text(json.dumps(manifest), encoding="utf-8")
|
|
with pytest.raises(WgsUpstreamManifestError, match=message):
|
|
_resolve(workspace, path)
|
|
|
|
|
|
@pytest.mark.parametrize("mutation", ["escape", "symlink", "missing", "size", "hash", "full_path"])
|
|
def test_resolver_rejects_untrusted_artifact_evidence(tmp_path: Path, mutation: str) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
path, manifest = _qc_manifest(workspace)
|
|
item = manifest["artifacts"]["clean_reads_r1.S1"]
|
|
artifact_path = workspace / item["path"]
|
|
if mutation == "escape":
|
|
outside = tmp_path / "outside.fastq.gz"; outside.write_bytes(b"outside")
|
|
item.update(path="../outside.fastq.gz", full_path=str(outside), size=7, size_bytes=7,
|
|
sha256=hashlib.sha256(b"outside").hexdigest())
|
|
elif mutation == "symlink":
|
|
target = tmp_path / "target.fastq.gz"; target.write_bytes(artifact_path.read_bytes())
|
|
artifact_path.unlink(); artifact_path.symlink_to(target)
|
|
elif mutation == "missing":
|
|
artifact_path.unlink()
|
|
elif mutation == "size":
|
|
item["size"] += 1
|
|
elif mutation == "hash":
|
|
item["sha256"] = "0" * 64
|
|
else:
|
|
item["full_path"] = str(tmp_path / "wrong.fastq.gz")
|
|
path.write_text(json.dumps(manifest), encoding="utf-8")
|
|
with pytest.raises(WgsUpstreamManifestError):
|
|
_resolve(workspace, path)
|
|
|
|
|
|
def test_resolver_rejects_missing_role_and_duplicate_sample_role(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
path, manifest = _qc_manifest(workspace)
|
|
manifest["artifacts"] = {
|
|
key: value for key, value in manifest["artifacts"].items() if value["role"] != "clean_reads_r2"
|
|
}
|
|
path.write_text(json.dumps(manifest), encoding="utf-8")
|
|
with pytest.raises(WgsUpstreamManifestError, match="missing required role"):
|
|
_resolve(workspace, path)
|
|
|
|
path, manifest = _qc_manifest(workspace)
|
|
duplicate = dict(manifest["artifacts"]["clean_reads_r1.S1"])
|
|
manifest["artifacts"]["clean_reads_r1.S1.duplicate"] = duplicate
|
|
path.write_text(json.dumps(manifest), encoding="utf-8")
|
|
with pytest.raises(WgsUpstreamManifestError, match="duplicate"):
|
|
_resolve(workspace, path)
|
|
|
|
|
|
def test_resolver_rejects_duplicate_index_component(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
bundle = workspace / "final" / ".wgs_index_runs" / "Cohort" / RUN_ID
|
|
artifacts = {
|
|
"component.1": _artifact(workspace, bundle, "reference.0123", "bwa_mem2_index_component", component=".0123"),
|
|
"component.2": _artifact(workspace, bundle, "reference.copy.0123", "bwa_mem2_index_component", component=".0123"),
|
|
}
|
|
manifest = {"manifest_version": "1.0", "stage": "wgs_index", "status": "success", "prefix": "Cohort",
|
|
"job_id": "job-1", "run_id": RUN_ID, "workspace": str(workspace.resolve()),
|
|
"wgs_reference_id": REFERENCE_ID, "artifacts": artifacts}
|
|
path = workspace / "final" / "Cohort.wgs_index.manifest.json"
|
|
path.write_text(json.dumps(manifest), encoding="utf-8")
|
|
with pytest.raises(WgsUpstreamManifestError, match="duplicate"):
|
|
resolve_wgs_artifacts(path, "wgs_index", ("bwa_mem2_index_component",), expected_reference_id=REFERENCE_ID)
|
|
|
|
|
|
def test_resolver_binds_every_bwa_component_to_the_declared_reference_basename(tmp_path: Path) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
bundle = workspace / "final" / ".wgs_index_runs" / "Cohort" / RUN_ID
|
|
reference = _artifact(workspace, bundle, "reference.fa", "reference_fasta")
|
|
artifacts = {"reference_fasta": reference}
|
|
for component in (".0123", ".amb", ".ann", ".bwt.2bit.64", ".pac"):
|
|
artifacts[f"component.{component}"] = _artifact(
|
|
workspace, bundle, f"unrelated.fa{component}",
|
|
"bwa_mem2_index_component", component=component,
|
|
)
|
|
manifest = {
|
|
"manifest_version": "1.0", "stage": "wgs_index", "status": "success", "prefix": "Cohort",
|
|
"job_id": "job-1", "run_id": RUN_ID, "workspace": str(workspace.resolve()),
|
|
"wgs_reference_id": REFERENCE_ID, "artifacts": artifacts,
|
|
}
|
|
path = workspace / "final" / "Cohort.wgs_index.manifest.json"
|
|
path.write_text(json.dumps(manifest), encoding="utf-8")
|
|
|
|
with pytest.raises(WgsUpstreamManifestError, match="reference basename"):
|
|
resolve_wgs_artifacts(
|
|
path, "wgs_index", ("reference_fasta", "bwa_mem2_index_component"),
|
|
expected_reference_id=REFERENCE_ID,
|
|
)
|