88 lines
3.5 KiB
Python
88 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
|
|
|
|
def _stage18_helpers():
|
|
path = Path(__file__).with_name("test_gwas_fmt_stage18.py")
|
|
spec = importlib.util.spec_from_file_location("gwas_fmt_stage18_helpers", path)
|
|
assert spec and spec.loader
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def _assert_core_manifest(manifest_path: Path) -> dict[str, object]:
|
|
validator = importlib.import_module("apps.cli.validate_manifest")
|
|
schema = Path(__file__).resolve().parents[2] / "gfa_core/schemas/core/manifest_v1.schema.json"
|
|
assert validator.validate_schema(manifest_path, schema) is validator.HAS_JSONSCHEMA
|
|
assert validator.validate_file_existence(manifest_path) is True
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
required = set(json.loads(schema.read_text(encoding="utf-8"))["required"])
|
|
assert required <= set(manifest)
|
|
assert manifest["manifest_version"] == "1.0"
|
|
assert manifest["cmd"][:3] == ["python", "-m", "apps.cli.run_gwas_tassel"]
|
|
assert Path(manifest["workspace"]) == Path(manifest["workspace_root"]) / "final"
|
|
assert manifest["start_time"]
|
|
assert manifest["end_time"]
|
|
assert manifest["elapsed_sec"] >= 0
|
|
return manifest
|
|
|
|
|
|
def test_gwas_fmt_success_manifest_conforms_to_core_v1(tmp_path: Path) -> None:
|
|
helpers = _stage18_helpers()
|
|
cli = importlib.import_module("apps.cli.run_gwas_tassel")
|
|
vcf = tmp_path / "input.vcf"
|
|
tassel = tmp_path / "run_pipeline.pl"
|
|
workspace = tmp_path / "workspace"
|
|
helpers._write_vcf(vcf)
|
|
helpers._write_fake_tassel(tassel)
|
|
|
|
assert cli.main([
|
|
"--workspace", str(workspace), "--job-id", "job", "--prefix", "cohort",
|
|
"--vcf", str(vcf), "--tassel-pipeline", str(tassel),
|
|
]) == 0
|
|
|
|
manifest = _assert_core_manifest(workspace / "final/cohort.gwas_fmt.manifest.json")
|
|
assert manifest["status"] == "success"
|
|
assert manifest["genotype_scope_id"].startswith("genotype:")
|
|
assert manifest["inputs"][0]["sha256"]
|
|
assert manifest["tool_versions"]["tassel"] == "5.2.96"
|
|
assert manifest["metrics"]["strict_genotype_conservation"] is True
|
|
assert set(manifest["artifacts"]) == {"sorted_vcf", "tassel_hapmap", "environment", "report"}
|
|
for name, artifact in manifest["artifacts"].items():
|
|
assert artifact["artifact"] == name
|
|
assert {"path", "full_path", "type", "size", "sha256"} <= artifact.keys()
|
|
|
|
|
|
def test_gwas_fmt_failure_manifest_conforms_to_core_v1(tmp_path: Path) -> None:
|
|
cli = importlib.import_module("apps.cli.run_gwas_tassel")
|
|
workspace = tmp_path / "workspace"
|
|
|
|
assert cli.main([
|
|
"--workspace", str(workspace), "--job-id", "job", "--prefix", "cohort",
|
|
"--vcf", str(tmp_path / "missing.vcf"),
|
|
]) == 1
|
|
|
|
manifest = _assert_core_manifest(workspace / "final/cohort.gwas_fmt.manifest.json")
|
|
assert manifest["status"] == "failed"
|
|
assert manifest["artifacts"] == {}
|
|
assert manifest["metrics"] == {}
|
|
|
|
|
|
def test_manifest_validator_fails_closed_when_jsonschema_is_unavailable(
|
|
monkeypatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
validator = importlib.import_module("apps.cli.validate_manifest")
|
|
manifest = tmp_path / "manifest.json"
|
|
schema = tmp_path / "schema.json"
|
|
manifest.write_text("{}", encoding="utf-8")
|
|
schema.write_text("{}", encoding="utf-8")
|
|
monkeypatch.setattr(validator, "HAS_JSONSCHEMA", False)
|
|
|
|
assert validator.validate_schema(manifest, schema) is False
|