first add
This commit is contained in:
90
scripts/test_bttoxin_digger.py
Normal file
90
scripts/test_bttoxin_digger.py
Normal file
@@ -0,0 +1,90 @@
|
||||
#!/usr/bin/env python3
|
||||
"""BtToxin_Digger 容器最小可行测试
|
||||
|
||||
使用 tests/test_data 下的 97-27.fna 与 C15.fna 作为输入,验证 nucl 类型分析,
|
||||
检查是否生成关键输出文件。
|
||||
"""
|
||||
from pathlib import Path
|
||||
import sys
|
||||
import logging
|
||||
|
||||
sys.path.append(str(Path(__file__).resolve().parents[1] / "backend"))
|
||||
from app.utils.docker_client import DockerContainerManager # noqa: E402
|
||||
|
||||
|
||||
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def ensure_test_layout() -> tuple[Path, Path, Path]:
|
||||
base = Path(__file__).resolve().parents[1] / "tests"
|
||||
input_dir = base / "test_data"
|
||||
output_dir = base / "output"
|
||||
log_dir = base / "logs"
|
||||
for d in (input_dir, output_dir, log_dir):
|
||||
d.mkdir(parents=True, exist_ok=True)
|
||||
return input_dir, output_dir, log_dir
|
||||
|
||||
|
||||
def check_input_files(input_dir: Path) -> bool:
|
||||
files = ["97-27.fna", "C15.fna"]
|
||||
missing = [f for f in files if not (input_dir / f).exists()]
|
||||
if missing:
|
||||
logger.error(f"缺少测试输入文件: {missing}")
|
||||
logger.info("请将 97-27.fna 与 C15.fna 放入 tests/test_data 目录")
|
||||
return False
|
||||
logger.info(f"发现测试文件: {files}")
|
||||
return True
|
||||
|
||||
|
||||
def run_test() -> int:
|
||||
input_dir, output_dir, log_dir = ensure_test_layout()
|
||||
if not check_input_files(input_dir):
|
||||
return 1
|
||||
|
||||
mgr = DockerContainerManager(
|
||||
image="quay.io/biocontainers/bttoxin_digger:1.0.10--hdfd78af_0",
|
||||
platform="linux/amd64",
|
||||
)
|
||||
|
||||
logger.info("开始 BtToxin_Digger 分析 (nucl)")
|
||||
result = mgr.run_bttoxin_digger(
|
||||
input_dir=input_dir,
|
||||
output_dir=output_dir,
|
||||
log_dir=log_dir,
|
||||
sequence_type="nucl",
|
||||
scaf_suffix=".fna",
|
||||
threads=4,
|
||||
)
|
||||
|
||||
if not result.get("success"):
|
||||
logger.error(f"分析失败: {result.get('error')} (exit={result.get('exit_code')})")
|
||||
logs = result.get("logs")
|
||||
if logs:
|
||||
logger.error(f"容器日志(节选):\n{logs[:1000]}")
|
||||
return 2
|
||||
|
||||
toxins_dir = output_dir / "Results" / "Toxins"
|
||||
expected = [
|
||||
toxins_dir / "97-27.list",
|
||||
toxins_dir / "97-27.gbk",
|
||||
toxins_dir / "C15.list",
|
||||
toxins_dir / "C15.gbk",
|
||||
toxins_dir / "Bt_all_genes.table",
|
||||
toxins_dir / "All_Toxins.txt",
|
||||
]
|
||||
missing = [p.name for p in expected if not p.exists()]
|
||||
if missing:
|
||||
logger.error(f"缺少预期输出文件: {missing}")
|
||||
return 3
|
||||
|
||||
logger.info("✓ 测试通过,关键输出文件存在")
|
||||
logger.info(f"输出目录: {toxins_dir}")
|
||||
logger.info(f"日志文件: {log_dir / 'digger_execution.log'}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(run_test())
|
||||
|
||||
|
||||
Reference in New Issue
Block a user