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, )