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_and_render_draft_includes_rich_mvp_template(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}] async def fake_generate_draft_payload(*, prompt): captured["prompt"] = prompt return { "title": "跨境 DTC 品牌 AI 运营平台", "summary": "帮助中国 DTC 品牌用 AI 快速完成海外市场的内容本地化、社媒运营、用户洞察。", "core_pain_points": [ { "pain": "内容本地化", "status": "翻译公司贵、慢、不地道", "opportunity": "AI 快速生成 + 人工微调", }, { "pain": "社媒运营", "status": "雇海外运营成本高", "opportunity": "AI 自动生成 + 排期发布", }, ], "mvp_items": [ "多语言内容本地化", "社媒内容生成器", "用户反馈分析", ], "technical_architecture": [ "前端：React + TypeScript + Tailwind", "后端：FastAPI + PostgreSQL + Redis", ], "business_model": [ {"plan": "Starter", "price": "$49/月", "features": "1 个品牌，1000 次 AI 调用/月"}, ], "validation_metrics": { "product": ["注册转化率 > 15%", "7 日留存 > 40%"], "market": ["找到 5 个愿意付费的种子用户"], }, "quick_start_suggestions": { "week_1_2": ["做 10 个目标用户的深度访谈"], "week_3_4": ["如果反馈正面，开始写代码"], }, "team_setup": [ {"role": "1 个产品", "notes": "兼职 CEO"}, {"role": "2 个后端", "notes": "AI 集成 + 数据处理"}, ], "similar_projects": [], "missing_info": [], "followup_plan": [ {"when": "明天", "action": "确认触发时机与验收口径"}, ], } 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": "跨境 DTC 品牌 AI 运营平台", "description": "出海运营项目"}, "active_projects": [], "recent_signals": [], "capacity": [], } generated = await service.generate_and_persist( work_item=work_item, intent_name="project.create", message_text="帮我起一个出海 DTC 品牌项目 MVP", world_model=world_model, ) prompt = generated["draft_text"] assert "一、定位" in prompt assert "二、核心痛点" in prompt assert "三、MVP 功能范围" in prompt assert "四、技术架构" in prompt assert "五、商业模式" in prompt assert "六、验证指标" in prompt assert "七、快速启动建议" in prompt assert "八、团队配置" in prompt assert "内容本地化" in prompt assert "Starter" in prompt assert "1 个产品" in prompt assert "positioning_one_liner" in captured["prompt"] assert "core_pain_points" in captured["prompt"] assert "mvp_phase_1" in captured["prompt"] assert "technical_architecture" in captured["prompt"] assert "business_model" in captured["prompt"] assert "validation_metrics" in captured["prompt"] assert "quick_start_suggestions" in captured["prompt"] assert "team_setup" in captured["prompt"] assert any(item in generated["draft"]["requirement_items"][0]["title"] for item in ("内容本地化", "多语言内容本地化")) @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"] == "围绕内部知识库问答能力升级"