from types import SimpleNamespace import pytest from fastapi import FastAPI from httpx import ASGITransport, AsyncClient from app.api.router import api_router from app.api.ai_pmo_api import routes_core from app.api.rag_api import routes_query from app.core.database import get_db from app.dependencies.auth import get_current_user from app.dependencies.tenant import get_tenant_id def _build_test_app() -> FastAPI: app = FastAPI() app.include_router(routes_query.router) app.include_router(routes_core.router) async def _fake_get_db(): yield object() app.dependency_overrides[get_db] = _fake_get_db app.dependency_overrides[get_tenant_id] = lambda: 9 app.dependency_overrides[get_current_user] = lambda: SimpleNamespace( id=101, tenant_id=9, name="测试用户", username="tester", ) return app @pytest.mark.asyncio async def test_http_rag_ask_readonly_mainline_contract(monkeypatch) -> None: captured: dict[str, object] = {} class FakeKernel: def __init__(self, db, tenant_id: int): _ = db captured["tenant_id"] = tenant_id async def process_ingress(self, **kwargs): captured["ingress_kwargs"] = kwargs return { "handled": True, "status": "answered", "intent": "query.read_only", "direct_reply_required": True, "direct_reply_text": "当前主链状态正常。", "rag_result": { "answer": "当前主链状态正常。", "processing_stage": "qa", "route_decision": { "route_mode": "internal", "effective_source_policy": "internal", "decision_origin": "reply_anchor", }, "clarification_code": None, "clarification_candidates": [], "answer_mode": "fast_path", "grounding_status": "grounded", "policy_version": "retrieval-v2.1|routing:v3|prompt:deterministic", "citations": [], "sources_used": [], "evidence_pack": { "summary": { "conclusion": "当前主链状态正常。", "evidence": [], "uncertainties": [], }, "tool_trace": [], }, }, } monkeypatch.setattr(routes_query, "AIPMOKernelService", FakeKernel) app = _build_test_app() transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: response = await client.post( "/rag/ask", json={ "question": "当前主链状态？", "asked_via": "pmflow_manual", "action_mode": "read", }, ) assert response.status_code == 200 payload = response.json() assert payload["code"] == 0 assert "当前主链状态正常" in str((payload.get("data") or {}).get("answer") or "") assert payload["data"]["route_decision"]["effective_source_policy"] == "internal" assert payload["data"]["clarification_code"] is None assert payload["data"]["clarification_candidates"] == [] assert payload["data"]["answer_mode"] == "fast_path" assert payload["data"]["grounding_status"] == "grounded" assert payload["data"]["policy_version"] == "retrieval-v2.1|routing:v3|prompt:deterministic" assert payload["data"]["citations"] == [] assert payload["data"]["sources_used"] == [] assert payload["data"]["evidence_pack"]["summary"]["conclusion"] == "当前主链状态正常。" ingress_kwargs = captured["ingress_kwargs"] assert isinstance(ingress_kwargs, dict) assert ingress_kwargs["source_system"] == "pmflow_manual" rag_request_context = ingress_kwargs["ingress_options"]["rag_request_context"] assert rag_request_context["asked_via"] == "pmflow_manual" assert rag_request_context["action_mode"] == "read" assert rag_request_context["message_intent"] is None @pytest.mark.asyncio async def test_http_rag_ask_mixed_mainline_contract(monkeypatch) -> None: captured: dict[str, object] = {} class FakeKernel: def __init__(self, db, tenant_id: int): _ = db captured["tenant_id"] = tenant_id async def process_ingress(self, **kwargs): captured["ingress_kwargs"] = kwargs return { "handled": True, "status": "answered", "intent": "query.read_only", "direct_reply_required": True, "direct_reply_text": "[回答] 进展稳定\n[已收到需求] 新增任务确认流", "rag_result": {"answer": "[回答] 进展稳定\n[已收到需求] 新增任务确认流"}, } monkeypatch.setattr(routes_query, "AIPMOKernelService", FakeKernel) app = _build_test_app() transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: response = await client.post( "/rag/ask", json={ "question": "顺便问进展并新增任务确认流", "asked_via": "pmflow_manual", "action_mode": "read", "message_intent": "mixed", "mixed_signal_part": "新增任务确认流", }, ) assert response.status_code == 200 payload = response.json() assert payload["code"] == 0 assert "[回答]" in str((payload.get("data") or {}).get("answer") or "") ingress_kwargs = captured["ingress_kwargs"] assert isinstance(ingress_kwargs, dict) rag_request_context = ingress_kwargs["ingress_options"]["rag_request_context"] assert rag_request_context["message_intent"] == "mixed" assert rag_request_context["mixed_signal_part"] == "新增任务确认流" @pytest.mark.asyncio async def test_http_rag_ask_should_attach_project_operating_context_for_current_project(monkeypatch) -> None: captured: dict[str, object] = {} class FakeKernel: def __init__(self, db, tenant_id: int): _ = db captured["tenant_id"] = tenant_id async def process_ingress(self, **kwargs): captured["ingress_kwargs"] = kwargs return { "handled": True, "status": "answered", "intent": "query.read_only", "direct_reply_required": True, "direct_reply_text": "先给判断，再给下一步。", "rag_result": { "answer": "先给判断，再给下一步。", "route_decision": { "route_mode": "external", "effective_source_policy": "external", "decision_origin": "query_route_gateway", }, }, } monkeypatch.setattr(routes_query, "AIPMOKernelService", FakeKernel) app = _build_test_app() transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: response = await client.post( "/rag/ask", json={ "question": "OpenAI Docs 这种能力适不适合我们当前项目？", "asked_via": "feishu", "project_id": 12, "source_chat_id": "chat-12", "action_mode": "read", }, ) assert response.status_code == 200 ingress_options = captured["ingress_kwargs"]["ingress_options"] packet = ingress_options["project_operating_context"] assert packet["current_project_id"] == 12 assert packet["current_repo_scope"] assert packet["fit_scope"] == "current_project_and_repo" assert packet["decision_strategy"] == "answer_then_offer_plan" @pytest.mark.asyncio async def test_http_rag_ask_returns_offer_after_answer_handoff_metadata_for_current_project_fit_question( monkeypatch, ) -> None: class FakeKernel: def __init__(self, db, tenant_id: int): _ = (db, tenant_id) async def process_ingress(self, **kwargs): _ = kwargs return { "handled": True, "status": "answered", "intent": "query.read_only", "direct_reply_required": True, "direct_reply_text": "外部实时查询：目前证据不足，建议先明确要查融资还是招聘动态。", "rag_result": { "answer": "外部实时查询：目前证据不足，建议先明确要查融资还是招聘动态。", "route_decision": { "route_mode": "external", "effective_source_policy": "external", "decision_origin": "query_route_gateway", }, "project_operating_context": { "context_version": "v1", "current_project_id": 12, "current_repo_scope": "alpha-repo", "page_anchor": {"type": "project_board", "ref": "project:12"}, "thread_anchor": {"source": "pmflow"}, "fit_scope": "current_project_and_repo", "decision_strategy": "answer_then_offer_plan", }, "execution_recipe": { "primary_lane": "external", "fit_brief_required": True, "external_evidence_required": True, "plan_handoff_mode": "offer_after_answer", }, "fit_brief": { "project_summary": "Alpha 是一个 AI 项目管理工作台。", "repo_summary": "当前仓库负责统一项目上下文与 AI handoff。", }, "external_recommendation": { "retrieval_status": "evidence_found", "impact_status": "indirect_only", "external_impact": "目前更像背景相关，还不能直接映射到当前正式对象。", }, "analysis_packet": { "packet_version": "v1", "packet_type": "project_fit_analysis", "conclusion": "放到采购经理每日实时库存上报服务里，Responses API 更像接在摘要卡片后面的解释层，不像主计算链。", "focus_objects": [ { "name": "每日固定时段生成并推送结构化摘要卡片", "kind": "task", "stance": "better_fit", "reason": "这一步更接近结果出来之后的解释和追问层。", }, { "name": "WMS库存快照自动拉取与清洗", "kind": "task", "stance": "not_fit", "reason": "这一步属于确定性取数或计算主链。", }, ], "fit_verdict": "fit_later", "current_subject": "采购经理每日实时库存上报服务", "candidate_position": "每日固定时段生成并推送结构化摘要卡片", "rejected_position": "WMS库存快照自动拉取与清洗", "preferred_layer": "每日固定时段生成并推送结构化摘要卡片", "rejected_layer": "WMS库存快照自动拉取与清洗", "reason_chain": [ "WMS库存快照自动拉取与清洗这一步属于确定性取数或计算主链，结果要稳定、可复算。", "每日固定时段生成并推送结构化摘要卡片更接近结果出来之后的解释和追问层，更适合承接这类能力。", ], "missing_mapping": "还缺它在每日固定时段生成并推送结构化摘要卡片这一步到底承担解释、问答还是编排角色的明确映射。", "next_step": "先确认它是不是只用在摘要卡片之后。", }, "handoff_mode": "offer_after_answer", "can_enter_plan": True, "recommended_next_step": "如果你要，我可以基于当前结论进入 plan，继续把需求范围和下一步收清楚。", }, } monkeypatch.setattr(routes_query, "AIPMOKernelService", FakeKernel) app = _build_test_app() transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: response = await client.post( "/rag/ask", json={ "question": "OpenAI Responses API 放到当前库存上报项目里，适合放在哪一层？", "asked_via": "pmflow_manual", "project_id": 12, "source_chat_id": "chat-12", "action_mode": "read", }, ) assert response.status_code == 200 payload = response.json() assert payload["code"] == 0 data = payload["data"] assert data["route_decision"]["route_mode"] == "external" assert data["project_operating_context"]["current_project_id"] == 12 assert data["execution_recipe"]["plan_handoff_mode"] == "offer_after_answer" assert data["fit_brief"]["project_summary"] == "Alpha 是一个 AI 项目管理工作台。" assert data["external_recommendation"]["impact_status"] == "indirect_only" assert data["analysis_packet"]["packet_type"] == "project_fit_analysis" assert data["analysis_packet"]["current_subject"] == "采购经理每日实时库存上报服务" assert data["analysis_packet"]["candidate_position"] == "每日固定时段生成并推送结构化摘要卡片" assert data["analysis_packet"]["rejected_position"] == "WMS库存快照自动拉取与清洗" assert len(data["analysis_packet"]["reason_chain"]) == 2 assert data["analysis_packet"]["fit_verdict"] == "fit_later" assert data["handoff_mode"] == "offer_after_answer" assert data["can_enter_plan"] is True assert "进入 plan" in data["recommended_next_step"] @pytest.mark.asyncio async def test_http_ai_pmo_process_text_draft_pending_confirm_mainline(monkeypatch) -> None: captured: dict[str, object] = {} class FakeKernel: def __init__(self, db, tenant_id: int): _ = db captured["tenant_id"] = tenant_id async def process_ingress(self, **kwargs): captured["ingress_kwargs"] = kwargs return { "handled": True, "status": "draft_pending_confirm", "classification": "command", "direct_reply_required": True, "direct_reply_text": "草稿已生成，请确认。", "draft_version": 2, } monkeypatch.setattr(routes_core, "AIPMOKernelService", FakeKernel) app = _build_test_app() transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: response = await client.post( "/ai-pmo/process-text", json={ "message_text": "请创建任务并先给草稿", "sender": "产品经理A", "source_message_id": "manual-msg-100", "project_id": 12, }, ) assert response.status_code == 200 payload = response.json() assert payload["code"] == 0 data = payload["data"] assert data["status"] == "draft_pending_confirm" kwargs = captured["ingress_kwargs"] assert isinstance(kwargs, dict) assert kwargs["source_system"] == "pmflow_manual" assert kwargs["message_text"] == "请创建任务并先给草稿" assert kwargs["project_hint_id"] == 12 @pytest.mark.asyncio async def test_http_ai_pmo_process_text_should_attach_project_operating_context(monkeypatch) -> None: captured: dict[str, object] = {} class FakeKernel: def __init__(self, db, tenant_id: int): _ = db captured["tenant_id"] = tenant_id async def process_ingress(self, **kwargs): captured["ingress_kwargs"] = kwargs return { "handled": True, "status": "answered", "classification": "query", "direct_reply_required": True, "direct_reply_text": "这个方向先判断，再决定是否进入计划。", } monkeypatch.setattr(routes_core, "AIPMOKernelService", FakeKernel) app = _build_test_app() transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: response = await client.post( "/ai-pmo/process-text", json={ "message_text": "这个方案适不适合我们当前项目？", "sender": "产品经理A", "source_message_id": "manual-msg-200", "project_id": 12, }, ) assert response.status_code == 200 ingress_options = captured["ingress_kwargs"]["ingress_options"] packet = ingress_options["project_operating_context"] assert packet["current_project_id"] == 12 assert packet["fit_scope"] == "current_project_and_repo" assert packet["decision_strategy"] == "answer_then_offer_plan" @pytest.mark.asyncio async def test_http_ai_pmo_work_items_list_smoke(monkeypatch) -> None: class FakeResult: def all(self): return [] class FakeDB: async def execute(self, *_args, **_kwargs): return FakeResult() async def _fake_get_db(): yield FakeDB() app = _build_test_app() app.dependency_overrides[get_db] = _fake_get_db transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: response = await client.get("/ai-pmo/work-items", params={"limit": 18}) assert response.status_code == 200 payload = response.json() assert payload["code"] == 0 assert payload["data"]["items"] == [] def test_ai_pmo_api_router_uses_single_prefix() -> None: route_paths = {str(getattr(route, "path", "")).strip() for route in api_router.routes} assert "/ai-pmo/work-items" in route_paths assert "/ai-pmo/process-text" in route_paths assert "/ai-pmo/ai-pmo/work-items" not in route_paths assert "/ai-pmo/ai-pmo/process-text" not in route_paths