adminandClaude Fable 5
4866b0408a
服务端生命周期部分（契约 §1/§2/§4/§5/§6；Dev-A）：
- extension_spec.py: spec_v2 白名单 12 占位符 + {output_dir} 必含 + 白名单外拒绝
- extensions.py: 生命周期状态机(契约 §2 枚举) + {extensions}/{ext_id} 文件存储
+ ExtensionExecutor 接口占位(Dev-B 实现 extension_exec.py, inject_executor 注入)
+ extension_* 事件帧经 EventBus 发布(events.py 注册, 不改 job_* 语义)
- extension_llm.py: build/repair AI 转换(DOCGEN_LLM_* 优先, BIOFLOW_EXTENSION_* 别名)
LLM 不可用/坏输出一律回退规则版(_detect_requirements/_command_to_template 移植), 绝不 500
- routers/extension.py: 契约 §6 端点全集; trial/run 执行器未接入返回 503
- intent.py: 契约 §5 两层判定纯函数(A 创建词库 × B 已发布扩展打分 + 歧义消解 + 漏判安全)
chat/stream 接线留给 Dev-C 主分支整合
- config/schemas/main: upload_ext_allow 加 .zip/.py/.r/.sh; extensions_dir +
extension_max_repair_rounds(20) + LLM 键注释; 模型; router 注册
- tests: 5 个测试文件 71 用例; 全仓 286 用例通过
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
111 lines
4.4 KiB
Python
111 lines
4.4 KiB
Python
"""tests/test_extension_spec.py —— 扩展模板白名单契约（契约 §2 spec_v2 移植）。
|
|
|
|
覆盖：占位符白名单 12 个 / output_dir 必含 / 白名单外拒绝 / param_* 放行 /
|
|
analysis_type 归一 / 规则版 analysis_type 探测 / template_parameters 派生补齐。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from app.services.bioflow import extension_spec as spec
|
|
|
|
pytestmark = pytest.mark.bioflow_unit
|
|
|
|
|
|
# ---- 白名单 ----
|
|
|
|
def test_placeholders_whitelist_has_12():
|
|
assert len(spec.PLACEHOLDERS) == 12
|
|
assert spec.PLACEHOLDERS == {
|
|
"code_dir", "input_dir", "input_file", "input_hmm", "input_fasta",
|
|
"input_query_fasta", "input_reference_fasta", "input_pep_fasta",
|
|
"input_cds_fasta", "input_id_file", "output_dir", "work_dir",
|
|
}
|
|
|
|
|
|
def test_output_dir_required():
|
|
with pytest.raises(spec.ExtensionSpecError, match="output_dir"):
|
|
spec.validate_command_template("python {code_dir}/a.py --in {input_file}")
|
|
|
|
|
|
def test_empty_template_rejected():
|
|
with pytest.raises(spec.ExtensionSpecError):
|
|
spec.validate_command_template("")
|
|
|
|
|
|
def test_unknown_placeholder_rejected():
|
|
with pytest.raises(spec.ExtensionSpecError, match="白名单外"):
|
|
spec.validate_command_template("python {code_dir}/a.py --out {output_dir} --x {input_bam}")
|
|
|
|
|
|
def test_whitelist_all_accepted():
|
|
tpl = (
|
|
"python {code_dir}/run.py --dir {input_dir} --file {input_file} "
|
|
"--hmm {input_hmm} --fa {input_fasta} --q {input_query_fasta} "
|
|
"--ref {input_reference_fasta} --pep {input_pep_fasta} "
|
|
"--cds {input_cds_fasta} --ids {input_id_file} --out {output_dir} "
|
|
"--work {work_dir}"
|
|
)
|
|
spec.validate_command_template(tpl) # 不抛即通过
|
|
|
|
|
|
def test_param_placeholders_allowed():
|
|
tpl = "python {code_dir}/a.py --in {input_file} --out {output_dir} --cpu {param_cpu} --mem {param_memory}"
|
|
spec.validate_command_template(tpl)
|
|
assert spec.param_placeholder_names(tpl) == {"cpu", "memory"}
|
|
|
|
|
|
def test_unsupported_detection():
|
|
tpl = "python {code_dir}/a.py --out {output_dir} --db {database}"
|
|
assert spec.unsupported_placeholders(tpl) == {"database"}
|
|
assert spec.unsupported_placeholders(
|
|
"python {code_dir}/a.py --out {output_dir} --cpu {param_cpu}"
|
|
) == set()
|
|
|
|
|
|
# ---- analysis_type ----
|
|
|
|
def test_normalize_analysis_type():
|
|
assert spec.normalize_analysis_type("table_filter") == "table_filter"
|
|
assert spec.normalize_analysis_type("") == "generic_cli"
|
|
assert spec.normalize_analysis_type("bogus") == "generic_cli"
|
|
assert spec.normalize_analysis_type(None) == "generic_cli"
|
|
|
|
|
|
def test_infer_analysis_type_rules():
|
|
assert spec.infer_analysis_type(template="blastp -query {input_fasta}", display_name="x") == "blast_search"
|
|
assert spec.infer_analysis_type(template="hmmsearch {input_hmm} {input_fasta}", display_name="x") == "hmmer_search"
|
|
assert spec.infer_analysis_type(template="awk -F'\\t' '{print $1}' {input_file}", display_name="x") == "table_filter"
|
|
assert spec.infer_analysis_type(template="python {code_dir}/a.py", display_name="id 清洗 clean_id") == "fasta_header_cleaning"
|
|
assert spec.infer_analysis_type(template="python {code_dir}/a.py", display_name="画图工具") == "plot_render"
|
|
assert spec.infer_analysis_type(template="python {code_dir}/a.py", display_name="随便") == "generic_cli"
|
|
|
|
|
|
# ---- template_parameters ----
|
|
|
|
def test_template_parameters_derived():
|
|
tpl = "python {code_dir}/a.py --cpu {param_cpu} --out {output_dir}"
|
|
params = spec.template_parameters(tpl)
|
|
names = {p["name"] for p in params}
|
|
assert names == {"cpu"}
|
|
assert params[0]["type"] == "string"
|
|
|
|
|
|
def test_template_parameters_keeps_existing():
|
|
tpl = "python {code_dir}/a.py --cpu {param_cpu} --mem {param_mem} --out {output_dir}"
|
|
params = spec.template_parameters(tpl, [
|
|
{"name": "cpu", "type": "int", "required": True, "description": "核数", "default": 4},
|
|
])
|
|
by_name = {p["name"]: p for p in params}
|
|
assert set(by_name) == {"cpu", "mem"}
|
|
assert by_name["cpu"]["type"] == "int"
|
|
assert by_name["cpu"]["default"] == 4
|
|
assert by_name["mem"]["required"] is False
|
|
|
|
|
|
def test_safe_parameter_name():
|
|
assert spec.safe_parameter_name("fastq 文件") == "fastq"
|
|
assert spec.safe_parameter_name("2x") == "p_2x"
|
|
assert spec.safe_parameter_name("param_cpu") == "cpu"
|
|
assert spec.safe_parameter_name("") == ""
|