141 lines
5.4 KiB
Python
141 lines
5.4 KiB
Python
import pytest
|
||
|
||
from backend.app.services.adaptive_rag_service import AdaptiveRAGService
|
||
from backend.app.services.feishu_signal_pipeline import FeishuSignalPipelineService
|
||
|
||
|
||
def test_requirement_like_question_detected():
|
||
assert FeishuSignalPipelineService._is_requirement_like_question("这个接口要不要支持飞书同步？")
|
||
assert FeishuSignalPipelineService._is_requirement_like_question("能不能新增一个导出按钮？")
|
||
assert FeishuSignalPipelineService._is_requirement_like_question("是否需要补充需求优先级字段？")
|
||
|
||
|
||
def test_non_requirement_question_not_detected():
|
||
assert not FeishuSignalPipelineService._is_requirement_like_question("今天几号？")
|
||
assert not FeishuSignalPipelineService._is_requirement_like_question("你是谁？")
|
||
assert not FeishuSignalPipelineService._is_requirement_like_question("这个现在怎么样？")
|
||
|
||
|
||
def test_question_to_statement_normalization():
|
||
normalized = FeishuSignalPipelineService._extract_requirement_statement_from_question("这个接口要不要支持飞书同步？")
|
||
assert "支持飞书同步" in normalized
|
||
assert "？" not in normalized
|
||
|
||
normalized2 = FeishuSignalPipelineService._extract_requirement_statement_from_question("能不能新增一个导出按钮？")
|
||
assert "新增一个导出按钮" in normalized2
|
||
assert "？" not in normalized2
|
||
|
||
|
||
def test_build_rag_conversation_key():
|
||
key = FeishuSignalPipelineService._build_rag_conversation_key(
|
||
source_chat_id="oc_123",
|
||
source_sender_open_id="ou_abc",
|
||
sender="张三",
|
||
)
|
||
assert key == "feishu:oc_123:ou_abc"
|
||
|
||
key2 = FeishuSignalPipelineService._build_rag_conversation_key(
|
||
source_chat_id="oc_123",
|
||
source_sender_open_id=None,
|
||
sender=" 张 三 ",
|
||
)
|
||
assert key2 == "feishu:oc_123:张三"
|
||
|
||
|
||
def test_openclaw_command_detection():
|
||
assert FeishuSignalPipelineService._is_openclaw_business_command("帮我创建一个项目，财务冲账协同")
|
||
assert FeishuSignalPipelineService._is_openclaw_business_command("把这个项目归档")
|
||
assert not FeishuSignalPipelineService._is_openclaw_business_command("今天几号？")
|
||
|
||
|
||
def test_build_openclaw_command_reply_for_project_create():
|
||
reply = FeishuSignalPipelineService._build_openclaw_command_reply(
|
||
{
|
||
"execution": {
|
||
"applied_count": 1,
|
||
"skipped_count": 0,
|
||
"failed_count": 0,
|
||
"results": [
|
||
{
|
||
"operation_code": "project.create",
|
||
"status": "applied",
|
||
"entity_type": "project",
|
||
"entity_id": "8",
|
||
"result": {
|
||
"project_id": 8,
|
||
"title": "财务冲账协同",
|
||
"status": "active",
|
||
},
|
||
}
|
||
],
|
||
}
|
||
}
|
||
)
|
||
|
||
assert reply == "已执行完成：项目#8「财务冲账协同」已创建，状态 active。"
|
||
|
||
|
||
def test_build_openclaw_command_reply_for_existing_project():
|
||
reply = FeishuSignalPipelineService._build_openclaw_command_reply(
|
||
{
|
||
"execution": {
|
||
"applied_count": 0,
|
||
"skipped_count": 1,
|
||
"failed_count": 0,
|
||
"results": [
|
||
{
|
||
"operation_code": "project.create",
|
||
"status": "skipped",
|
||
"message": "project_already_exists",
|
||
"entity_type": "project",
|
||
"entity_id": "8",
|
||
"result": {
|
||
"project_id": 8,
|
||
"title": "财务冲账协同",
|
||
"status": "active",
|
||
},
|
||
}
|
||
],
|
||
}
|
||
}
|
||
)
|
||
|
||
assert reply == "未重复执行：项目#8「财务冲账协同」已存在，状态 active。"
|
||
|
||
|
||
def test_adaptive_rag_intent_rule_detects_project_queries():
|
||
service = AdaptiveRAGService(db=None, tenant_id=1)
|
||
|
||
result = service._classify_intent_rule("现在有哪些项目")
|
||
assert result["intent"] == "question"
|
||
assert result["confidence"] >= 0.86
|
||
|
||
result2 = service._classify_intent_rule("一个 AI 客服工具，这个项目有吗")
|
||
assert result2["intent"] == "question"
|
||
assert result2["confidence"] >= 0.86
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_deterministic_project_lookup_answers_from_db(monkeypatch):
|
||
service = FeishuSignalPipelineService(db=None, tenant_id=1)
|
||
|
||
async def _fake_write_processing_log(**kwargs):
|
||
return None
|
||
|
||
monkeypatch.setattr(service, "_write_processing_log", _fake_write_processing_log)
|
||
result = await service._maybe_answer_project_lookup_question(
|
||
message="现在有哪些项目",
|
||
feishu_message_id="probe-lookup-1",
|
||
process_started_at=0.0,
|
||
projects=[
|
||
{"id": 8, "title": "财务冲账协同", "name": "财务冲账协同", "status": "active"},
|
||
{"id": 5, "title": "食谱需求", "name": "食谱需求", "status": "active"},
|
||
],
|
||
)
|
||
|
||
assert result is not None
|
||
assert result["status"] == "answered"
|
||
assert result["classification"] == "question"
|
||
assert result["created_project_id"] is None
|
||
assert "#8 财务冲账协同(active)" in result["direct_reply_text"]
|