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 == []