358 lines
13 KiB
Python
358 lines
13 KiB
Python
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"
|