102 lines
3.5 KiB
Python
102 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
import inspect
|
|
|
|
from apps.web import pipeline_backend
|
|
from apps.web.vue3.main import ExecuteRequest
|
|
from gfa_core.project_paths import (
|
|
get_conda_env_bin,
|
|
get_conda_env_rscript,
|
|
get_configured_path,
|
|
)
|
|
from gfa_core import project_paths
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
SCOPED_FILES = (
|
|
ROOT / "gfa_core/mas/tools/stage_tools.py",
|
|
ROOT / "apps/web/pipeline_backend.py",
|
|
ROOT / "apps/web/vue3/main.py",
|
|
ROOT / "apps/web/vue3/index.html",
|
|
)
|
|
FORBIDDEN_TOOL_PATHS = (
|
|
"/data/miniconda3/envs/ASAW",
|
|
"/home/xzq/miniconda3/envs/ASAW",
|
|
"/home/xia-lab/miniconda3/envs/ASAW",
|
|
"/usr/bin/emmax",
|
|
)
|
|
|
|
|
|
def test_asaw_paths_follow_bbav5_conda_base(monkeypatch, tmp_path: Path) -> None:
|
|
conda_base = tmp_path / "conda"
|
|
monkeypatch.setenv("BBAV5_CONDA_BASE", str(conda_base))
|
|
|
|
assert get_conda_env_bin("ASAW") == conda_base / "envs/ASAW/bin"
|
|
assert get_conda_env_rscript("ASAW") == conda_base / "envs/ASAW/bin/Rscript"
|
|
|
|
|
|
def test_configured_tool_path_comes_from_environment(monkeypatch) -> None:
|
|
monkeypatch.setenv("GWAS_EMMAX_BIN", "/configured/emmax")
|
|
|
|
assert get_configured_path("GWAS_EMMAX_BIN", "emmax") == "/configured/emmax"
|
|
assert get_configured_path("UNSET_TOOL", "fallback") == "fallback"
|
|
|
|
|
|
def test_wgs_tool_defaults_are_resolved_server_side() -> None:
|
|
request = ExecuteRequest(stage="wgs_qc")
|
|
asaw_bin = get_conda_env_bin("ASAW")
|
|
parameters = inspect.signature(pipeline_backend.run_stage).parameters
|
|
|
|
assert request.wgsFastp == ""
|
|
assert request.wgsSamtools == ""
|
|
assert request.wgsBwaMem2 == ""
|
|
assert request.wgsGatk == ""
|
|
assert request.wgsBcftools == ""
|
|
assert request.wgsJava == ""
|
|
assert parameters["wgs_fastp"].default == str(asaw_bin / "fastp")
|
|
assert parameters["wgs_samtools"].default == str(asaw_bin / "samtools")
|
|
assert parameters["wgs_bwa_mem2"].default == str(asaw_bin / "bwa-mem2")
|
|
assert parameters["wgs_gatk"].default == str(asaw_bin / "gatk")
|
|
assert hasattr(project_paths, "get_wgs_bcftools")
|
|
assert parameters["wgs_bcftools"].default == project_paths.get_wgs_bcftools()
|
|
assert hasattr(project_paths, "get_wgs_java")
|
|
assert parameters["wgs_java"].default == project_paths.get_wgs_java()
|
|
|
|
|
|
def test_wgs_java_uses_environment_then_test4py311(
|
|
monkeypatch, tmp_path: Path
|
|
) -> None:
|
|
assert hasattr(project_paths, "get_wgs_java")
|
|
get_wgs_java = project_paths.get_wgs_java
|
|
monkeypatch.setenv("WGS_JAVA", "/configured/java")
|
|
assert get_wgs_java() == "/configured/java"
|
|
assert get_wgs_java("/requested/java") == "/requested/java"
|
|
|
|
monkeypatch.delenv("WGS_JAVA")
|
|
monkeypatch.setenv("BBAV5_CONDA_BASE", str(tmp_path / "conda"))
|
|
assert get_wgs_java() == str(
|
|
tmp_path / "conda/envs/test4py311/bin/java"
|
|
)
|
|
|
|
|
|
def test_wgs_bcftools_uses_environment_then_compatible_default(
|
|
monkeypatch,
|
|
) -> None:
|
|
assert hasattr(project_paths, "get_wgs_bcftools")
|
|
get_wgs_bcftools = project_paths.get_wgs_bcftools
|
|
monkeypatch.setenv("WGS_BCFTOOLS", "/configured/bcftools")
|
|
assert get_wgs_bcftools() == "/configured/bcftools"
|
|
assert get_wgs_bcftools("/requested/bcftools") == "/requested/bcftools"
|
|
|
|
|
|
def test_gwas_population_wgs_defaults_have_no_machine_tool_paths() -> None:
|
|
violations = []
|
|
for path in SCOPED_FILES:
|
|
text = path.read_text(encoding="utf-8")
|
|
for forbidden in FORBIDDEN_TOOL_PATHS:
|
|
if forbidden in text:
|
|
violations.append(f"{path.relative_to(ROOT)}: {forbidden}")
|
|
|
|
assert violations == []
|