203 lines
7.5 KiB
Python
203 lines
7.5 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
from types import SimpleNamespace
|
||
|
||
import pytest
|
||
|
||
from app.config.ai_prompts import build_default_role_prompt_bundle
|
||
from app.services.ai_service_core import AIService
|
||
from app.services.sstc import charter_prefill_utils
|
||
from app.services.signal.draft_generator_service import DraftGeneratorService
|
||
|
||
|
||
def test_default_role_prompt_bundle_is_pm_reviewer() -> None:
|
||
bundle = build_default_role_prompt_bundle()
|
||
assert bundle["active_role_key"] == "pm_dispatcher"
|
||
role = next((item for item in bundle["roles"] if item.get("role_key") == "pm_dispatcher"), {})
|
||
assert "立项评审" in str(role.get("role_name") or "")
|
||
assert "同名或相似项目" in str(role.get("system_prompt") or "")
|
||
assert "MVP" in str(role.get("system_prompt") or "")
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_generate_and_render_draft_includes_pm_sections(monkeypatch) -> None:
|
||
service = DraftGeneratorService(db=object(), tenant_id=1)
|
||
captured: dict[str, str] = {}
|
||
|
||
async def fake_load_members(*, world_model):
|
||
_ = world_model
|
||
return [
|
||
{"id": 7, "name": "Alice", "workload_percent": 25.0},
|
||
{"id": 8, "name": "Bob", "workload_percent": 60.0},
|
||
]
|
||
|
||
async def fake_generate_draft_payload(*, prompt):
|
||
captured["prompt"] = prompt
|
||
return {}
|
||
|
||
monkeypatch.setattr(service, "_load_members", fake_load_members)
|
||
monkeypatch.setattr(service, "_generate_draft_payload", fake_generate_draft_payload)
|
||
|
||
work_item = SimpleNamespace(
|
||
draft_version=0,
|
||
draft_content={},
|
||
draft_expires_at=None,
|
||
confirm_message_id=None,
|
||
execution_result={},
|
||
)
|
||
world_model = {
|
||
"project": {"title": "企业知识库问答平台", "description": "面向内部知识库的问答能力"},
|
||
"active_projects": [
|
||
{
|
||
"id": 51,
|
||
"name": "AI劝退助手",
|
||
"title": "AI劝退助手",
|
||
"description": "项目进展与风险提醒",
|
||
"status": "active",
|
||
"owner_id": 10,
|
||
},
|
||
{
|
||
"id": 52,
|
||
"name": "企业知识库问答平台 V2",
|
||
"title": "企业知识库问答平台 V2",
|
||
"description": "旧版知识库问答平台",
|
||
"status": "active",
|
||
"owner_id": 11,
|
||
},
|
||
],
|
||
"recent_signals": [
|
||
{"id": 91, "title": "文档检索性能优化", "signal_type": "新需求", "status": "open"},
|
||
],
|
||
"capacity": [],
|
||
}
|
||
|
||
generated = await service.generate_and_persist(
|
||
work_item=work_item,
|
||
intent_name="project.create",
|
||
message_text="我想做一个企业知识库问答项目，先帮我评估一下。",
|
||
world_model=world_model,
|
||
)
|
||
|
||
prompt = captured["prompt"]
|
||
assert "现有活跃项目（JSON）" in prompt
|
||
assert "候选相似项目（JSON）" in prompt
|
||
assert "同名/相似项目列表" in prompt
|
||
assert generated["draft"]["similar_projects"]
|
||
assert "🔎 同名 / 相似项目：" in generated["draft_text"]
|
||
assert "🚀 MVP 建议：" in generated["draft_text"]
|
||
assert "📈 前景判断：" in generated["draft_text"]
|
||
assert "🧩 需求清单：" in generated["draft_text"]
|
||
assert "❓ 还缺的信息：" in generated["draft_text"]
|
||
assert generated["draft"]["recommended_assignee_name"] == "Alice"
|
||
assert generated["draft"]["recommended_assignee_workload"] == 25.0
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_generate_charter_seed_uses_project_context(monkeypatch) -> None:
|
||
captured: dict[str, str] = {}
|
||
|
||
async def fake_load_project_context(*, db, tenant_id, project_id):
|
||
_ = db, tenant_id, project_id
|
||
return {
|
||
"id": 7,
|
||
"name": "kb-v2",
|
||
"title": "企业知识库问答平台",
|
||
"description": "内部知识库问答升级",
|
||
"status": "active",
|
||
"owner_id": 10,
|
||
}
|
||
|
||
async def fake_load_active_projects_context(*, db, tenant_id, project_id, limit=10):
|
||
_ = db, tenant_id, project_id, limit
|
||
return [
|
||
{
|
||
"id": 51,
|
||
"name": "ai-warn",
|
||
"title": "AI劝退助手",
|
||
"description": "项目进展与风险提醒",
|
||
"status": "active",
|
||
"owner_id": 11,
|
||
}
|
||
]
|
||
|
||
async def fake_generate(prompt, **kwargs):
|
||
captured["prompt"] = prompt
|
||
return json.dumps(
|
||
{
|
||
"title": "内部知识库问答升级",
|
||
"summary": "围绕内部知识库问答能力升级",
|
||
"background": "背景说明",
|
||
"objectives": ["提升检索质量"],
|
||
"scope": "范围说明",
|
||
"success_metrics": ["命中率提升"],
|
||
"risks": ["重复建设风险"],
|
||
"dependencies": ["知识库索引"],
|
||
"primary_model": "gpt",
|
||
"fallback_model": "gpt-mini",
|
||
},
|
||
ensure_ascii=False,
|
||
)
|
||
|
||
monkeypatch.setattr(charter_prefill_utils, "_load_charter_project_context", fake_load_project_context)
|
||
monkeypatch.setattr(charter_prefill_utils, "_load_active_projects_context", fake_load_active_projects_context)
|
||
monkeypatch.setattr(charter_prefill_utils.llm_service, "generate", fake_generate)
|
||
|
||
result = await charter_prefill_utils._generate_charter_seed(
|
||
db=object(),
|
||
tenant_id=1,
|
||
project_id=7,
|
||
signal_context="AI 劝退诉求：请升级内部知识库问答能力。",
|
||
)
|
||
|
||
prompt = captured["prompt"]
|
||
assert "当前项目上下文" in prompt
|
||
assert "企业知识库问答平台" in prompt
|
||
assert "现有活跃项目（JSON）" in prompt
|
||
assert "AI劝退助手" in prompt
|
||
assert result["summary"] == "围绕内部知识库问答能力升级"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_generate_charter_from_signal_includes_project_context(monkeypatch) -> None:
|
||
service = AIService()
|
||
captured: dict[str, object] = {}
|
||
|
||
async def fake_invoke(self, prompt, **kwargs):
|
||
_ = self
|
||
captured["prompt"] = prompt
|
||
captured["kwargs"] = kwargs
|
||
return json.dumps(
|
||
{
|
||
"title": "内部知识库问答升级",
|
||
"summary": "围绕内部知识库问答能力升级",
|
||
"background": "背景说明",
|
||
"objectives": ["提升检索质量"],
|
||
"scope": "范围说明",
|
||
"success_metrics": ["命中率提升"],
|
||
"risks": ["重复建设风险"],
|
||
"dependencies": ["知识库索引"],
|
||
"primary_model": "gpt",
|
||
"fallback_model": "gpt-mini",
|
||
},
|
||
ensure_ascii=False,
|
||
)
|
||
|
||
monkeypatch.setattr(AIService, "_invoke_with_retry", fake_invoke)
|
||
|
||
result = await service.generate_charter_from_signal(
|
||
"AI 劝退诉求：请升级内部知识库问答能力。",
|
||
tenant_id=1,
|
||
db=object(),
|
||
project_id=7,
|
||
project_context="当前项目上下文：企业知识库问答平台",
|
||
active_projects_json='[{"id":51,"title":"AI劝退助手"}]',
|
||
)
|
||
|
||
prompt = str(captured["prompt"] or "")
|
||
kwargs = captured["kwargs"] if isinstance(captured.get("kwargs"), dict) else {}
|
||
assert "当前项目上下文：企业知识库问答平台" in prompt
|
||
assert "现有活跃项目（JSON）" in prompt
|
||
assert kwargs.get("project_id") == 7
|
||
assert result["summary"] == "围绕内部知识库问答能力升级"
|