135 lines
5.0 KiB
Python
135 lines
5.0 KiB
Python
"""
|
|
AI分析引擎/CEO简报模块测试
|
|
"""
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy.orm import Session
|
|
from datetime import datetime
|
|
from tests.conftest import create_test_user, get_token_for_user, auth_header, create_test_kpi
|
|
from app.models import KPIAlert, KPIValue, ActionPlan
|
|
|
|
|
|
class TestAiAnalysis:
|
|
"""AI分析/CEO简报测试"""
|
|
|
|
def test_brief_no_data(self, client: TestClient, db: Session):
|
|
"""无数据时简报端点已移除，预期404"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
|
|
resp = client.get("/api/cma/ai/brief", headers=auth_header(token))
|
|
assert resp.status_code == 404, "brief端点已移除"
|
|
|
|
def test_brief_with_data(self, client: TestClient, db: Session):
|
|
"""有数据时简报端点已移除，预期404"""
|
|
user = create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
kpi = create_test_kpi(db)
|
|
kpi_val = KPIValue(kpi_id=kpi.id, period="2026-06", actual_value=85.0)
|
|
db.add(kpi_val)
|
|
db.commit()
|
|
|
|
resp = client.get("/api/cma/ai/brief?timeout=5", headers=auth_header(token))
|
|
assert resp.status_code == 404, "brief端点已移除"
|
|
|
|
def test_dashboard_analysis_no_data(self, client: TestClient, db: Session):
|
|
"""无数据时AI驾驶舱分析"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
|
|
resp = client.get("/api/cma/ai/dashboard-analysis", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "analysis" in data
|
|
assert data["kpi_count"] == 0
|
|
assert data["alert_count"] == 0
|
|
|
|
def test_dashboard_analysis_with_kpi(self, client: TestClient, db: Session):
|
|
"""有KPI数据时驾驶舱分析正常返回"""
|
|
user = create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
kpi = create_test_kpi(db)
|
|
kpi_val = KPIValue(kpi_id=kpi.id, period="2026-06", actual_value=90.0)
|
|
db.add(kpi_val)
|
|
db.commit()
|
|
|
|
resp = client.get("/api/cma/ai/dashboard-analysis", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "analysis" in data
|
|
assert isinstance(data["kpi_count"], int) # 计数可能因后端过滤为0
|
|
|
|
def test_kpi_analysis_not_found(self, client: TestClient, db: Session):
|
|
"""分析不存在的KPI"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
|
|
resp = client.get("/api/cma/ai/kpi-analysis/9999", headers=auth_header(token))
|
|
assert resp.status_code == 404
|
|
|
|
def test_kpi_analysis_success(self, client: TestClient, db: Session):
|
|
"""分析单个KPI"""
|
|
user = create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
kpi = create_test_kpi(db)
|
|
kpi_val = KPIValue(kpi_id=kpi.id, period="2026-04", actual_value=80.0)
|
|
db.add(kpi_val)
|
|
kpi_val2 = KPIValue(kpi_id=kpi.id, period="2026-05", actual_value=85.0)
|
|
db.add(kpi_val2)
|
|
kpi_val3 = KPIValue(kpi_id=kpi.id, period="2026-06", actual_value=90.0)
|
|
db.add(kpi_val3)
|
|
db.commit()
|
|
|
|
resp = client.get(f"/api/cma/ai/kpi-analysis/{kpi.id}", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["kpi_name"] == "测试KPI"
|
|
assert "analysis" in data
|
|
|
|
def test_review_plans_empty(self, client: TestClient, db: Session):
|
|
"""无改善计划时的复盘"""
|
|
user = create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
create_test_user(db, username="ceo_user", name="CEO用户", role="ceo")
|
|
|
|
resp = client.post("/api/cma/ai/review-plans", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
assert "暂无改善行动计划" in resp.json()["analysis"]
|
|
|
|
def test_review_plans_with_data(self, client: TestClient, db: Session):
|
|
"""有改善计划时的复盘"""
|
|
user = create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
kpi = create_test_kpi(db)
|
|
plan = ActionPlan(
|
|
alert_id=None,
|
|
kpi_id=kpi.id,
|
|
title="提升营收10%",
|
|
assignee="张三",
|
|
priority="high",
|
|
status="in_progress",
|
|
progress=50,
|
|
created_by="系统",
|
|
)
|
|
db.add(plan)
|
|
db.commit()
|
|
|
|
resp = client.post("/api/cma/ai/review-plans", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "analysis" in data
|
|
assert data["stats"]["total"] == 1
|
|
assert data["stats"]["in_progress"] == 1
|
|
|
|
def test_ask_empty_question(self, client: TestClient, db: Session):
|
|
"""空问题被拒绝"""
|
|
user = create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
|
|
resp = client.post(
|
|
"/api/cma/ai/ask",
|
|
headers=auth_header(token),
|
|
json={"question": ""},
|
|
)
|
|
assert resp.status_code == 400
|