from __future__ import annotations import os from types import SimpleNamespace import pytest import pytest_asyncio from fastapi import FastAPI from httpx import ASGITransport, AsyncClient from sqlalchemy import JSON, select from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://tester:tester@localhost:5432/pmflow_test") from app.api import space_api # noqa: E402 from app.core.database import get_db # noqa: E402 from app.dependencies.auth import get_current_user # noqa: E402 from app.models import OperationLog, PMFlowLaunchDraft, Project, SavedIdea, User # noqa: E402 from app.models.base import Base # noqa: E402 from app.models.reply_iteration import PromptRelease # noqa: E402 from app.services import pmflow_launch_service # noqa: E402 from app.services import space_saved_idea_service # noqa: E402 from app.services.ai_routing_governance_service import PLATFORM_RUNTIME_TENANT_ID # noqa: E402 from app.services.prompt_release.service import PromptReleaseService # noqa: E402 TABLES = [ User.__table__, Project.__table__, SavedIdea.__table__, PMFlowLaunchDraft.__table__, OperationLog.__table__, PromptRelease.__table__, ] def _use_sqlite_json_types() -> None: for table in TABLES: for column in table.columns: type_name = column.type.__class__.__name__.upper() if type_name in {"JSONB", "ARRAY"}: column.type = JSON() @pytest_asyncio.fixture async def db_env(): engine = create_async_engine("sqlite+aiosqlite:///:memory:") _use_sqlite_json_types() async with engine.begin() as conn: await conn.run_sync(lambda sync_conn: Base.metadata.create_all(sync_conn, tables=TABLES)) maker = async_sessionmaker(engine, expire_on_commit=False) async with maker() as session: yield session await engine.dispose() async def _seed_user(session, *, tenant_id: int = 1): user = User( tenant_id=tenant_id, username="owner", email="owner@example.com", password_hash="hashed", name="Owner", role="MEMBER", assigned_domains=[], workload=0, status="active", is_active=True, ) session.add(user) await session.commit() return user @pytest.mark.asyncio async def test_space_ai_concierge_sessions_facade_uses_core_space_route(db_env, monkeypatch): user = await _seed_user(db_env) captured: dict[str, object] = {} class FakeAdaptiveRAGService: def __init__(self, db, tenant_id: int): captured["tenant_id"] = tenant_id async def list_qa_sessions(self, **kwargs): captured["list_kwargs"] = kwargs return {"items": [{"id": 7, "title": "空间问答", "message_count": 1}], "page": 1, "total_pages": 1} monkeypatch.setattr(space_api, "AdaptiveRAGService", FakeAdaptiveRAGService) app = _build_app(db_env, user) async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: response = await client.get("/space/ai-concierge/sessions?page=1&page_size=20") assert response.status_code == 200 payload = response.json()["data"] assert payload["items"][0]["title"] == "空间问答" assert captured["tenant_id"] == int(user.tenant_id) assert captured["list_kwargs"]["user_id"] == int(user.id) @pytest.mark.asyncio async def test_space_ai_concierge_ask_facade_uses_kernel_ingress(db_env, monkeypatch): user = await _seed_user(db_env) captured: dict[str, object] = {} class FakeAIPMOKernelService: def __init__(self, db, tenant_id: int): captured["tenant_id"] = tenant_id async def process_ingress(self, **kwargs): captured["ingress_kwargs"] = kwargs return { "handled": True, "direct_reply_text": "先处理待确认事项。", "rag_result": { "answer": "先处理待确认事项。", "reply_text": "先处理待确认事项。", "sources": [{"source": "space", "label": "个人空间", "score": 1}], }, } monkeypatch.setattr(space_api, "AIPMOKernelService", FakeAIPMOKernelService) app = _build_app(db_env, user) async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: response = await client.post( "/space/ai-concierge/ask", json={ "question": "我今天应该先处理什么？", "asked_via": "pmflow", "thinking_mode": "deep", "new_session": True, "action_mode": "auto", "sync_requirement": False, }, ) assert response.status_code == 200 payload = response.json()["data"] assert payload["reply_text"] == "先处理待确认事项。" assert captured["tenant_id"] == int(user.tenant_id) assert captured["ingress_kwargs"]["message_text"] == "我今天应该先处理什么？" def _build_app(session, current_user) -> FastAPI: app = FastAPI() app.include_router(space_api.router) async def _fake_get_db(): yield session app.dependency_overrides[get_db] = _fake_get_db app.dependency_overrides[get_current_user] = lambda: current_user return app def _draft_payload(): return { "title": "AI 会议纪要工具", "one_liner": "自动整理会议结论和待办", "target_user": "产品经理、研发负责人", "problem": "会议后行动项分散", "solution": "从会议内容中提取结论、风险和任务", "requirements": [ {"title": "上传会议内容", "summary": "支持录音或文本输入", "acceptance_criteria": ["能提交文本"]}, {"title": "提取行动项", "summary": "输出责任人和待办", "acceptance_criteria": ["能生成任务"]}, {"title": "同步项目上下文", "summary": "把结论带入项目", "acceptance_criteria": ["能进入项目页"]}, ], "risks": [{"title": "隐私风险", "summary": "会议内容敏感", "level": "high", "mitigation": "确认权限"}], "decisions": [{"title": "先做文本输入", "reason": "降低复杂度", "owner": "用户确认"}], "tasks": [ {"title": "设计输入表单", "summary": "完成输入", "group": "product", "priority": "medium", "requirement_index": 0} ], "evidence": [{"source": "user_input", "excerpt": "AI 会议纪要工具"}], "confidence": 0.8, "warnings": [], } @pytest.mark.asyncio async def test_saved_idea_create_list_archive(db_env, monkeypatch): user = await _seed_user(db_env) async def _fake_enrich_fail(*args, **kwargs): raise RuntimeError("ai unavailable") monkeypatch.setattr(space_saved_idea_service.llm_service, "generate_structured_object", _fake_enrich_fail) app = _build_app(db_env, user) async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: create_response = await client.post( "/space/saved-ideas", json={ "raw_text": "我想做一个面向产品经理的 AI 会议纪要工具，解决会后行动项分散的问题。", "source_type": "project_launch", "input_mode": "idea", }, ) assert create_response.status_code == 200 created = create_response.json()["data"] assert created["id"].startswith("idea_") assert created["raw_text"].startswith("我想做一个") assert created["status"] in {"enriched", "draft_ready", "needs_more_info"} assert created["enrichment_status"] == "ai_failed" assert created["enrichment_payload"]["source"] == "system_baseline" list_response = await client.get("/space/saved-ideas") assert list_response.status_code == 200 assert list_response.json()["data"]["items"][0]["id"] == created["id"] archive_response = await client.post(f"/space/saved-ideas/{created['id']}/archive") assert archive_response.status_code == 200 assert archive_response.json()["data"]["status"] == "archived" active_list_response = await client.get("/space/saved-ideas") assert active_list_response.json()["data"]["items"] == [] @pytest.mark.asyncio async def test_saved_idea_promotes_to_project_launch_draft(db_env, monkeypatch): user = await _seed_user(db_env) async def _fake_generate_structured_object(*args, **kwargs): if kwargs.get("function_name") == "space_saved_idea_enrichment_v1": raise RuntimeError("ai unavailable") return SimpleNamespace(payload=_draft_payload(), raw_text="{}", provider="test", model="test") monkeypatch.setattr(pmflow_launch_service.llm_service, "generate_structured_object", _fake_generate_structured_object) app = _build_app(db_env, user) async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: create_response = await client.post( "/space/saved-ideas", json={"raw_text": "我想做一个面向产品经理的 AI 会议纪要工具，解决会后行动项分散的问题。"}, ) idea_id = create_response.json()["data"]["id"] promote_response = await client.post(f"/space/saved-ideas/{idea_id}/promote") assert promote_response.status_code == 200 payload = promote_response.json()["data"] assert payload["idea"]["status"] == "promoted" assert payload["draft"]["draft_id"].startswith("draft_") assert payload["redirect_url"] == "/project-launch?state=human_review" row = (await db_env.execute(select(SavedIdea).where(SavedIdea.idea_id == idea_id))).scalar_one() assert row.launch_draft_id draft = (await db_env.execute(select(PMFlowLaunchDraft).where(PMFlowLaunchDraft.draft_id == row.launch_draft_id))).scalar_one() assert draft.input_text.startswith("我想做一个") @pytest.mark.asyncio async def test_saved_idea_uses_ai_enrichment_when_available(db_env, monkeypatch): user = await _seed_user(db_env) release = await PromptReleaseService(db_env).create_release( tenant_id=PLATFORM_RUNTIME_TENANT_ID, prompt_key=space_saved_idea_service.SpaceSavedIdeaService.ENRICHMENT_PROMPT_KEY, body_snapshot="ACTIVE SAVED IDEA PROMPT: {raw_text}", created_by=int(user.id), ) await PromptReleaseService(db_env).activate_release( tenant_id=PLATFORM_RUNTIME_TENANT_ID, release_id=release["id"], actor_id=int(user.id), ) captured: dict[str, object] = {} async def _fake_ai_enrich(*args, **kwargs): captured["prompt"] = args[0] assert kwargs["function_name"] == "space_saved_idea_enrichment_v1" assert kwargs["routing_meta"]["response_contract"] == "structured" assert kwargs["routing_meta"]["prompt_key"] == space_saved_idea_service.SpaceSavedIdeaService.ENRICHMENT_PROMPT_KEY assert kwargs["routing_meta"]["prompt_release_id"] == release["id"] assert kwargs["routing_meta"]["prompt_source"] == "active_release" return SimpleNamespace( payload={ "summary": "面向产品经理的会议纪要工具", "target_user": "产品经理", "problem": "会后行动项分散", "value_hypothesis": "减少会后整理成本", "validation_signal": "行动项整理时间下降", "constraints": ["先支持文本输入"], "missing_questions": ["目标会议来源是什么？"], "readiness_score": 0.82, "status": "draft_ready", "confidence": 0.76, }, raw_text="{}", provider="test", model="test", ) monkeypatch.setattr(space_saved_idea_service.llm_service, "generate_structured_object", _fake_ai_enrich) app = _build_app(db_env, user) async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: response = await client.post( "/space/saved-ideas", json={"raw_text": "我想做一个面向产品经理的 AI 会议纪要工具，解决会后行动项分散的问题。"}, ) assert response.status_code == 200 created = response.json()["data"] assert created["status"] == "draft_ready" assert created["status_label"] == "可生成草案" assert created["summary"] == "面向产品经理的会议纪要工具" assert created["ai_summary"] == "面向产品经理的会议纪要工具" assert created["missing_questions"] == ["目标会议来源是什么？"] assert created["readiness_score"] == 0.82 assert created["enrichment_status"] == "ai_enriched" assert created["enrichment_payload"]["target_user"] == "产品经理" assert captured["prompt"] == "ACTIVE SAVED IDEA PROMPT: 我想做一个面向产品经理的 AI 会议纪要工具，解决会后行动项分散的问题。"