52 lines
1.8 KiB
Python
52 lines
1.8 KiB
Python
import pytest
|
|
|
|
from backend.app.services.feishu_signal_pipeline import FeishuSignalPipelineService
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_classify_level_fallback_when_llm_empty(monkeypatch):
|
|
service = FeishuSignalPipelineService(db=None, tenant_id=1)
|
|
|
|
async def _fake_generate(*args, **kwargs):
|
|
return ""
|
|
|
|
monkeypatch.setattr(service.llm, "generate", _fake_generate)
|
|
result = await service._classify_level(message="我们需要新增导出按钮", projects=[])
|
|
|
|
assert result["level"] == "signal"
|
|
assert 0.0 <= float(result["project_score"]) <= 1.0
|
|
assert str(result["level_reasoning"]).strip()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_signal_fastpath_for_simple_request(monkeypatch):
|
|
service = FeishuSignalPipelineService(db=None, tenant_id=1)
|
|
|
|
async def _should_not_be_called(*args, **kwargs):
|
|
raise AssertionError("LLM should not be called on simple signal request fastpath")
|
|
|
|
monkeypatch.setattr(service.llm, "generate", _should_not_be_called)
|
|
result = await service._extract_signal(
|
|
"新增一个导出按钮，支持飞书同步",
|
|
projects=[{"id": 3, "name": "飞书协同平台"}],
|
|
)
|
|
|
|
assert result["type"] == "新需求"
|
|
assert result["priority"] in {"P1", "P2", "P3"}
|
|
assert str(result["summary"]).strip()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_project_prefill_fallback_when_llm_empty(monkeypatch):
|
|
service = FeishuSignalPipelineService(db=None, tenant_id=1)
|
|
|
|
async def _fake_generate(*args, **kwargs):
|
|
return ""
|
|
|
|
monkeypatch.setattr(service.llm, "generate", _fake_generate)
|
|
result = await service._extract_project_prefill("我们要做一个飞书需求同步项目")
|
|
|
|
assert str(result["suggested_name"]).strip()
|
|
assert str(result["suggested_description"]).strip()
|
|
assert str(result["suggested_owner"]).strip()
|