290 lines
12 KiB
Python
290 lines
12 KiB
Python
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
|
|
|
|
|
|
async def _seed_saved_idea_prompt_release(session, *, actor_id: int):
|
|
release = await PromptReleaseService(session).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(actor_id),
|
|
)
|
|
await PromptReleaseService(session).activate_release(
|
|
tenant_id=PLATFORM_RUNTIME_TENANT_ID,
|
|
release_id=release["id"],
|
|
actor_id=int(actor_id),
|
|
)
|
|
return release
|
|
|
|
|
|
async def _seed_launch_draft_prompt_release(session, *, actor_id: int):
|
|
release = await PromptReleaseService(session).create_release(
|
|
tenant_id=PLATFORM_RUNTIME_TENANT_ID,
|
|
prompt_key=pmflow_launch_service.PMFlowLaunchService.DRAFT_PROMPT_KEY,
|
|
body_snapshot="ACTIVE LAUNCH DRAFT PROMPT: {input_text}",
|
|
created_by=int(actor_id),
|
|
)
|
|
await PromptReleaseService(session).activate_release(
|
|
tenant_id=PLATFORM_RUNTIME_TENANT_ID,
|
|
release_id=release["id"],
|
|
actor_id=int(actor_id),
|
|
)
|
|
return release
|
|
|
|
|
|
@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)
|
|
|
|
|
|
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)
|
|
await _seed_saved_idea_prompt_release(db_env, actor_id=int(user.id))
|
|
|
|
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)
|
|
await _seed_saved_idea_prompt_release(db_env, actor_id=int(user.id))
|
|
await _seed_launch_draft_prompt_release(db_env, actor_id=int(user.id))
|
|
|
|
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 _seed_saved_idea_prompt_release(db_env, 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 会议纪要工具，解决会后行动项分散的问题。"
|