PlantAI-rag-new/bioflow_service/tests/test_extension_intent.py

"""tests/test_extension_intent.py —— 扩展意图两层判定（契约 §5）。

覆盖 A 层创建（动作词×目标词）/ B 层运行打分（仅 published、display_name
模糊/nl_triggers/analysis_type/动作词加分）/ 歧义消解（目标词无动作词 →
有命中走运行，无命中回 create_ambiguous）/ 漏判安全（无命中且非创建 →
layer none，不影响现有 route() 返回路径）。
"""
from __future__ import annotations

import pytest

from app.services.bioflow.intent import (
    detect_extension_intent,
    extension_create_intent,
    match_extensions,
    route,
)

pytestmark = pytest.mark.bioflow_unit


def _pub_ext(ext_id="ext_1", name="测序质量统计", triggers=None, analysis_type="table_filter"):
    return {
        "ext_id": ext_id,
        "display_name": name,
        "status": "published",
        "nl_triggers": triggers or ["质量统计", "fastq 统计"],
        "analysis_type": analysis_type,
    }


def _draft_ext(ext_id="ext_9", name="未发布扩展"):
    return {"ext_id": ext_id, "display_name": name, "status": "draft",
            "nl_triggers": ["未发布"], "analysis_type": "generic_cli"}


# ---- A 层创建 ----

@pytest.mark.parametrize("text", [
    "帮我创建一个扩展：统计 fastq 质量",
    "建个分析模板",
    "新建一个分析脚本",
    "上传我的分析工具",
    "自定义一个扩展",
    "写一个分析模板",
    "弄个扩展统计 fastq",
])
def test_create_intent_hits(text):
    assert extension_create_intent(text) is True


@pytest.mark.parametrize("text", [
    "帮我跑水稻 SPL 家族鉴定",          # 无目标词
    "质量统计",                          # 目标词+运行词，无创建动作词（消歧走 B 层）
    "做拓展分析 统计 fastq 质量",        # 契约歧义消解句：无创建动作词 → 引导卡而非直接创建
    "建议新建文件夹",                    # 「建」后随「议」→ 不误命中
    "这是写作练习",                      # 「写」后随「作」→ 不误命中
    "扩展名为 fasta 的文件",             # 「扩展名」≠ 目标词语义（漏判安全，不误入创建）
    "今天天气不错",
])
def test_create_intent_misses(text):
    assert extension_create_intent(text) is False


# ---- B 层打分 ----

def test_match_only_published():
    exts = [_pub_ext(), _draft_ext()]
    hits = match_extensions("帮我做一下质量统计", exts)
    assert [h["ext_id"] for h in hits] == ["ext_1"]


def test_match_display_name_fuzzy():
    hits = match_extensions("测序质量统计", [_pub_ext()])
    assert len(hits) == 1
    assert hits[0]["ext_id"] == "ext_1"
    assert hits[0]["score"] >= 4.0


def test_match_nl_trigger_keyword():
    hits = match_extensions("帮我把这个 fastq 做一下 fastq 统计", [_pub_ext()])
    assert len(hits) == 1
    assert any(r.startswith("trigger:") for r in hits[0]["reasons"])


def test_match_analysis_type_keyword():
    ext = _pub_ext(ext_id="ext_2", name="表达量统计", triggers=["表达量"],
                   analysis_type="enrichment_or_stats")
    # 动作词 + analysis_type 关键词 = 1.0 + 1.5 = 2.5 > 阈值 → 命中
    hits = match_extensions("帮我做一下富集分析", [ext])
    assert len(hits) == 1
    assert hits[0]["ext_id"] == "ext_2"
    assert any(r == "analysis_type:富集" for r in hits[0]["reasons"])
    # 漏判安全：仅 analysis_type 关键词（无动作词/触发词）→ 1.5 ≤ 阈值 → 不命中
    assert match_extensions("富集分析", [ext]) == []


def test_match_action_verb_bonus():
    ext = _pub_ext(ext_id="ext_3", name="表格过滤工具", triggers=["过滤表格"],
                   analysis_type="table_filter")
    # 无动作词：nl_trigger 2.5 + analysis_type 1.5 = 4.0（过阈值 2.0）
    base = match_extensions("过滤表格", [ext])
    assert len(base) == 1
    # 动作词加分后更高
    boosted = match_extensions("帮我跑过滤表格", [ext])
    assert boosted[0]["score"] > base[0]["score"]


def test_match_no_hit_empty():
    assert match_extensions("帮我跑水稻 SPL 家族鉴定", [_pub_ext()]) == []
    assert match_extensions("随便说点什么", [_pub_ext()]) == []


# ---- 两层判定主入口 ----

def test_detect_create_layer():
    r = detect_extension_intent("帮我创建一个扩展：统计 fastq 质量", [_pub_ext()])
    assert r["layer"] == "create"
    assert r["create"] is True


def test_detect_run_layer_high_confidence():
    r = detect_extension_intent("帮我把这个 fastq 做一下质量统计", [_pub_ext()])
    assert r["layer"] == "run"
    assert len(r["candidates"]) == 1
    assert r["candidates"][0]["ext_id"] == "ext_1"


def test_detect_run_layer_multi_candidates():
    exts = [
        _pub_ext("ext_1", "测序质量统计", ["质量统计"]),
        _pub_ext("ext_2", "表达量统计", ["统计"]),
    ]
    r = detect_extension_intent("帮我做个统计分析", exts)
    assert r["layer"] == "run"
    assert len(r["candidates"]) >= 1


def test_detect_ambiguity_no_published():
    """目标词命中、无创建动作词、无 B 层命中 → create_ambiguous（引导卡）。

    契约歧义消解句「做拓展分析 质量统计」：无创建动作词 → 走消解而非直接创建。
    """
    r = detect_extension_intent("做拓展分析 质量统计", [])
    assert r["layer"] == "create_ambiguous"
    assert r["ambiguous"] is True


def test_detect_ambiguity_target_no_action_but_run_hit():
    """歧义消解：目标词命中但无创建动作词 → 有 B 层命中走运行。"""
    r = detect_extension_intent("做拓展分析 质量统计", [_pub_ext()])
    assert r["layer"] == "run"


def test_detect_none_fallback():
    """漏判安全：无命中且非创建 → layer none（回落正常生信阶段规划）。"""
    r = detect_extension_intent("帮我跑水稻 SPL 家族鉴定", [_pub_ext()])
    assert r["layer"] == "none"


# ---- 漏判安全：现有 route() 返回路径零改动 ----

def test_route_unchanged_by_extension_words():
    """含「扩展」等的句子不得改变现有意图分流结论（扩展判定是独立纯函数层）。"""
    assert route("帮我跑水稻 SPL 家族鉴定") == "bioflow"
    assert route("今天天气不错") == "qa"
    # 「建个分析模板」不命中生信任务动词表 → 仍 qa（创建流程由扩展层接管）
    assert route("建个分析模板") == "qa"