from __future__ import annotations from types import SimpleNamespace import pytest import pytest_asyncio from fastapi import FastAPI, HTTPException from httpx import ASGITransport, AsyncClient from sqlalchemy import JSON from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from app.api import ai_readiness_api from app.core.database import get_db from app.dependencies.auth import get_current_user from app.dependencies.permissions import require_platform_ai_ops from app.models.ai_call_log import AICallLog from app.models.base import Base from app.models.integration import Integration from app.services import ai_readiness_service as ai_readiness_module from app.services.ai_readiness_service import AIReadinessService def _use_sqlite_json_types(*tables) -> None: for table in tables: for column in table.columns: if column.type.__class__.__name__ in {"JSONB", "ARRAY", "Vector"}: column.type = JSON() @pytest_asyncio.fixture async def db_session(): engine = create_async_engine("sqlite+aiosqlite:///:memory:") tables = [Integration.__table__, AICallLog.__table__] _use_sqlite_json_types(*tables) 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() def _engine_bundle() -> dict: return { "active_engine_id": "high-1", "engines": [ { "id": "low-1", "name": "Qwen 低阶引擎", "provider": "dashscope", "model": "qwen-plus", "base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1", "api_key": "low-key", "enabled": True, "connection_status": "available", "last_test_status": "success", "capabilities": { "structured_output": True, "json_schema": True, "capability_probe_status": "passed", "capability_evidence": { "structured_output": {"status": "verified", "source": "probe"}, "json_schema": {"status": "verified", "source": "probe"}, }, }, }, { "id": "high-1", "name": "OpenAI 高阶引擎", "provider": "openai", "model": "GLM-5", "base_url": "https://api.edgefn.net/v1", "api_key": "high-key", "enabled": True, "connection_status": "available", "last_test_status": "success", }, ], } def _slot_bundle(*, include_structured: bool) -> dict: slots = { "low": {"primary_engine_id": "low-1", "failure_policy": "upgrade_to_high"}, "high": {"primary_engine_id": "high-1", "failure_policy": "explicit_fail"}, } if include_structured: slots["structured"] = {"primary_engine_id": "low-1", "failure_policy": "try_fallback_then_fail"} return {"active": {"slots": slots}, "active_version": "slot-v-test"} async def _seed_platform_config(session, *, include_structured: bool) -> None: session.add(Integration(tenant_id=1, type="llm_engine", config=_engine_bundle(), is_active=True)) session.add(Integration(tenant_id=1, type="ai_capability_slots", config=_slot_bundle(include_structured=include_structured), is_active=True)) await session.commit() @pytest.mark.asyncio async def test_project_launch_readiness_blocks_when_structured_slot_unpublished(db_session) -> None: await _seed_platform_config(db_session, include_structured=False) payload = await AIReadinessService(db_session, tenant_id=12).get_project_launch_readiness() assert payload["status"] == "blocked" assert payload["can_generate_project_draft"] is False assert payload["platform"]["checks"]["active_slot"] == "failed" assert any(item["code"] == "structured_slot_unpublished" for item in payload["platform"]["blockers"]) assert payload["title"] == "AI 项目草案暂不可用" @pytest.mark.asyncio async def test_project_launch_readiness_ready_after_structured_slot_and_success(db_session) -> None: await _seed_platform_config(db_session, include_structured=True) db_session.add( AICallLog( tenant_id=12, function_name="pmflow_launch_draft_v1", prompt="prompt", response="{}", tokens_used=1, cost=0, status="success", planned_tier="structured", executed_tier="structured", routing_decision="capability_slot", route_source="slot_binding", route_rule_name="structured", ) ) await db_session.commit() payload = await AIReadinessService(db_session, tenant_id=12).get_project_launch_readiness() assert payload["status"] == "ready" assert payload["can_generate_project_draft"] is True assert payload["platform"]["checks"]["capability_probe"] == "passed" assert payload["platform"]["checks"]["recent_smoke"] == "passed" @pytest.mark.asyncio async def test_project_launch_readiness_uses_platform_smoke_when_newer_than_tenant_failure(db_session) -> None: await _seed_platform_config(db_session, include_structured=True) db_session.add_all( [ AICallLog( tenant_id=12, function_name="pmflow_launch_draft_v1", prompt="prompt", response="old failure", tokens_used=1, cost=0, status="structured_http_error", planned_tier="low", executed_tier="low", routing_decision="rule_match", route_source="default", route_rule_name="engine_low_fallback", ), AICallLog( tenant_id=1, function_name="pmflow_launch_draft_v1", prompt="prompt", response="{}", tokens_used=1, cost=0, status="success", planned_tier="structured", executed_tier="structured", routing_decision="capability_slot", route_source="slot_binding", route_rule_name="structured", ), ] ) await db_session.commit() payload = await AIReadinessService(db_session, tenant_id=12).get_project_launch_readiness() assert payload["status"] == "ready" assert payload["can_generate_project_draft"] is True assert payload["platform"]["checks"]["recent_smoke"] == "passed" @pytest.mark.asyncio async def test_project_launch_readiness_keeps_ready_after_single_recent_failure_with_recent_success(db_session) -> None: await _seed_platform_config(db_session, include_structured=True) db_session.add_all( [ AICallLog( tenant_id=12, function_name="pmflow_launch_draft_v1", prompt="prompt", response="{}", tokens_used=1, cost=0, status="success", planned_tier="structured", executed_tier="structured", routing_decision="capability_slot", route_source="slot_binding", route_rule_name="structured", ), AICallLog( tenant_id=12, function_name="pmflow_launch_draft_v1", prompt="prompt", response="模型调用失败：structured_empty", tokens_used=1, cost=0, status="structured_empty", planned_tier="structured", executed_tier="structured", routing_decision="capability_slot", route_source="slot_binding", route_rule_name="structured", ), ] ) await db_session.commit() payload = await AIReadinessService(db_session, tenant_id=12).get_project_launch_readiness() assert payload["status"] == "ready" assert payload["can_generate_project_draft"] is True assert payload["platform"]["checks"]["recent_smoke"] == "passed" assert payload["platform"]["checks"]["recent_failures"] == "warning" assert any(item["code"] == "recent_generation_failed" for item in payload["platform"]["blockers"]) @pytest.mark.asyncio async def test_project_launch_readiness_recovers_after_three_consecutive_recent_failures(db_session) -> None: await _seed_platform_config(db_session, include_structured=True) db_session.add_all( [ AICallLog( tenant_id=12, function_name="pmflow_launch_draft_v1", prompt="prompt", response="{}", tokens_used=1, cost=0, status="success", planned_tier="structured", executed_tier="structured", routing_decision="capability_slot", route_source="slot_binding", route_rule_name="structured", ), AICallLog( tenant_id=12, function_name="pmflow_launch_draft_v1", prompt="prompt", response="structured_empty", tokens_used=1, cost=0, status="structured_empty", planned_tier="structured", executed_tier="structured", routing_decision="capability_slot", route_source="slot_binding", route_rule_name="structured", ), AICallLog( tenant_id=12, function_name="pmflow_launch_draft_v1", prompt="prompt", response="structured_empty", tokens_used=1, cost=0, status="structured_empty", planned_tier="structured", executed_tier="structured", routing_decision="capability_slot", route_source="slot_binding", route_rule_name="structured", ), AICallLog( tenant_id=12, function_name="pmflow_launch_draft_v1", prompt="prompt", response="structured_empty", tokens_used=1, cost=0, status="structured_empty", planned_tier="structured", executed_tier="structured", routing_decision="capability_slot", route_source="slot_binding", route_rule_name="structured", ), ] ) await db_session.commit() payload = await AIReadinessService(db_session, tenant_id=12).get_project_launch_readiness() assert payload["status"] == "recovering" assert payload["can_generate_project_draft"] is False assert payload["platform"]["checks"]["recent_smoke"] == "failed" assert payload["platform"]["checks"]["recent_failures"] == "failed" @pytest.mark.asyncio async def test_project_launch_smoke_runs_against_platform_runtime_tenant(db_session, monkeypatch) -> None: await _seed_platform_config(db_session, include_structured=True) captured: dict[str, int] = {} class FakePMFlowLaunchService: def __init__(self, db, *, tenant_id: int): _ = db captured["tenant_id"] = int(tenant_id) async def generate_draft_payload_for_readiness_check(self, *, user_id: int, profile_stage: str = "active"): captured["user_id"] = int(user_id) captured["profile_stage"] = profile_stage return SimpleNamespace() monkeypatch.setattr(ai_readiness_module, "PMFlowLaunchService", FakePMFlowLaunchService) payload = await AIReadinessService(db_session, tenant_id=12).run_project_launch_check(user_id=101) assert captured == {"tenant_id": 1, "user_id": 101, "profile_stage": "active"} assert payload["platform"]["checks"]["provider_runtime"] == "passed" def _build_app(session, *, role: str = "OWNER") -> FastAPI: app = FastAPI() app.include_router(ai_readiness_api.router) fake_user = SimpleNamespace(id=101, tenant_id=12, username="tester", role=role) async def _fake_get_db(): yield session async def _fake_platform_ai_ops(): if role != "OWNER": raise HTTPException(status_code=403, detail="Platform capability required: platform.ai_routing.manage") return fake_user app.dependency_overrides[get_db] = _fake_get_db app.dependency_overrides[get_current_user] = lambda: fake_user app.dependency_overrides[require_platform_ai_ops] = _fake_platform_ai_ops return app @pytest.mark.asyncio async def test_ai_readiness_api_returns_enveloped_payload(db_session) -> None: await _seed_platform_config(db_session, include_structured=False) app = _build_app(db_session) async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: response = await client.get("/ai-readiness/project-launch") assert response.status_code == 200 body = response.json() assert body["code"] == 0 assert body["data"]["scene"] == "project_launch" assert body["data"]["status"] == "blocked"