113 lines
3.9 KiB
Python
113 lines
3.9 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from gfa_core.mas.core.fast_intent import fast_intent_recognition
|
|
from gfa_core.mas.utils.intent_parser import parse_intent_from_query
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("query", "action"),
|
|
[
|
|
("生成结果报告", "generate"),
|
|
("生成基因家族结果报告", "generate"),
|
|
("打开基因家族结果报告", "open"),
|
|
],
|
|
)
|
|
def test_explicit_gene_family_report_intent_is_provider_independent(query: str, action: str) -> None:
|
|
for parsed in (parse_intent_from_query(query), fast_intent_recognition(query)):
|
|
assert parsed is not None
|
|
assert parsed["stages"] == ["gene_family_report"]
|
|
assert parsed["params"]["geneFamilyReportAction"] == action
|
|
assert parsed["params"]["reportAction"] == action
|
|
assert parsed["method"] == "gene_family_report_deterministic"
|
|
|
|
|
|
def test_rnaseq_report_intent_is_not_stolen() -> None:
|
|
for parsed in (
|
|
parse_intent_from_query("打开 RNA-seq 结果报告"),
|
|
fast_intent_recognition("打开 RNA-seq 结果报告"),
|
|
):
|
|
assert not parsed or parsed.get("stages") != ["gene_family_report"]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("query", "action"),
|
|
[
|
|
("生成结果报告", "generate"),
|
|
("生成基因家族结果报告", "generate"),
|
|
("打开基因家族结果报告", "open"),
|
|
],
|
|
)
|
|
def test_gene_family_report_chat_bypasses_router(
|
|
monkeypatch,
|
|
tmp_path,
|
|
query: str,
|
|
action: str,
|
|
) -> None:
|
|
from apps.web.vue3 import main as web_main
|
|
from gfa_core.mas.agents import router as router_module
|
|
|
|
final_dir = tmp_path / "workspace" / "job-report" / "final"
|
|
final_dir.mkdir(parents=True)
|
|
(final_dir / "Ft.SPL.gene_family_report.manifest.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"stage": "gene_family_report",
|
|
"status": "success",
|
|
"prefix": "Ft.SPL",
|
|
"parameters": {
|
|
"expected_stages": ["identify", "motif"],
|
|
"stage_statuses": {"identify": "success", "motif": "success"},
|
|
},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
async def forbidden_router(*_args, **_kwargs):
|
|
raise AssertionError("report intent must return before Router/LLM")
|
|
|
|
class ForbiddenOrchestrator:
|
|
async def plan(self, *_args, **_kwargs):
|
|
raise AssertionError("report plan must not call the LLM orchestrator")
|
|
|
|
monkeypatch.setattr(web_main, "PROJECT_ROOT", tmp_path)
|
|
monkeypatch.setattr(web_main, "WORKSPACE_ROOT", tmp_path / "workspace")
|
|
monkeypatch.setattr(web_main, "agent_conversation_store", None, raising=False)
|
|
monkeypatch.setattr(web_main, "llm_manager", object())
|
|
monkeypatch.setattr(web_main, "orchestrator", ForbiddenOrchestrator())
|
|
monkeypatch.setattr(web_main, "MAS_AVAILABLE", True)
|
|
monkeypatch.setattr(router_module.RouterAgent, "execute", forbidden_router)
|
|
|
|
result = asyncio.run(
|
|
web_main._run_agent_chat(
|
|
web_main.AgentChatRequest(message=query, job_id="job-report")
|
|
)
|
|
)
|
|
|
|
assert result["action"] == "propose_analysis"
|
|
assert result["model_used"] == "rule"
|
|
assert result["intent"]["stages"] == ["gene_family_report"]
|
|
assert result["intent"]["prefix"] == "Ft.SPL"
|
|
assert result["intent"]["params"]["reportAction"] == action
|
|
assert result["intent"]["params"]["expectedStages"] == ["identify", "motif"]
|
|
|
|
plan = asyncio.run(
|
|
web_main.agent_plan(
|
|
web_main.AgentPlanRequest(
|
|
message=query,
|
|
job_id="job-report",
|
|
prefix="Ft.SPL",
|
|
intent=result["intent"],
|
|
)
|
|
)
|
|
)
|
|
assert plan["success"] is True
|
|
assert plan["mode"] == "gene_family_report_deterministic"
|
|
assert plan["plan"]["stages"] == ["gene_family_report"]
|
|
assert plan["plan"]["params"]["expectedStages"] == ["identify", "motif"]
|