447 lines
18 KiB
Python
447 lines
18 KiB
Python
"""
|
|
驾驶舱模块测试
|
|
"""
|
|
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
|
|
|
|
|
|
class TestDashboard:
|
|
"""驾驶舱核心接口测试"""
|
|
|
|
def test_summary_empty(self, client: TestClient, db: Session):
|
|
"""空系统时的驾驶舱摘要"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
|
|
resp = client.get("/api/cma/dashboard/summary", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["kpi_total"] == 0
|
|
assert data["alert_count"] == 0
|
|
assert data["dimension_stats"] == []
|
|
|
|
def test_summary_with_data(self, client: TestClient, db: Session):
|
|
"""有数据时的驾驶舱摘要"""
|
|
user = create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
kpi = create_test_kpi(db, kpi_code="F_REVENUE", dimension="finance")
|
|
kpi2 = create_test_kpi(db, kpi_code="C_SATISFACTION", kpi_name="客户满意度", dimension="customer")
|
|
alert = KPIAlert(kpi_id=kpi.id, alert_level="red", alert_message="营收预警", status="pending")
|
|
db.add(alert)
|
|
db.commit()
|
|
|
|
resp = client.get("/api/cma/dashboard/summary", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["kpi_total"] == 2
|
|
assert data["alert_count"] == 1
|
|
dims = {d["dimension"]: d["count"] for d in data["dimension_stats"]}
|
|
assert dims.get("finance") == 1
|
|
assert dims.get("customer") == 1
|
|
|
|
def test_kpis_empty(self, client: TestClient, db: Session):
|
|
"""无KPI时驾驶舱KPI列表"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
|
|
resp = client.get("/api/cma/dashboard/kpis", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["data"] == []
|
|
|
|
def test_kpis_with_data(self, client: TestClient, db: Session):
|
|
"""有KPI时驾驶舱KPI列表"""
|
|
user = create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
kpi = create_test_kpi(db, target_value=100.0)
|
|
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/dashboard/kpis", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert len(data["data"]) == 1
|
|
assert data["data"][0]["kpi_name"] == "测试KPI"
|
|
assert data["data"][0]["actual_value"] == 85.0
|
|
assert data["data"][0]["target_value"] == 100.0
|
|
|
|
def test_kpis_with_alert(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=50.0)
|
|
db.add(kpi_val)
|
|
alert = KPIAlert(kpi_id=kpi.id, alert_level="red", alert_message="严重偏离目标", status="pending")
|
|
db.add(alert)
|
|
db.commit()
|
|
|
|
resp = client.get("/api/cma/dashboard/kpis", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert len(data["data"]) == 1
|
|
assert data["data"][0]["alert_level"] == "red"
|
|
assert data["data"][0]["alert_message"] == "严重偏离目标"
|
|
|
|
def test_my_kpis(self, client: TestClient, db: Session):
|
|
"""我的KPI接口"""
|
|
user = create_test_user(db, username="biz_user", name="业务经理", role="business")
|
|
token = get_token_for_user(client, username="biz_user", password="admin123")
|
|
kpi = create_test_kpi(db, responsible_user="biz_user")
|
|
kpi2 = create_test_kpi(db, kpi_code="F_OTHER", kpi_name="无关KPI", responsible_user="其他人")
|
|
|
|
resp = client.get("/api/cma/dashboard/my-kpis", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
# business角色只看到自己的KPI
|
|
for k in data["data"]:
|
|
assert k["responsible_user"] == "biz_user"
|
|
|
|
def test_my_kpis_ceo_sees_all(self, client: TestClient, db: Session):
|
|
"""CEO角色的my-kpis看到所有有预警的KPI"""
|
|
user = create_test_user(db, username="ceo_user", name="CEO", role="ceo")
|
|
token = get_token_for_user(client, username="ceo_user", password="admin123")
|
|
kpi = create_test_kpi(db, kpi_code="F_KPI_A", responsible_user="张三")
|
|
|
|
resp = client.get("/api/cma/dashboard/my-kpis", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["user_role"] == "ceo"
|
|
|
|
def test_alert_stats_empty(self, client: TestClient, db: Session):
|
|
"""无预警时的预警统计"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
|
|
resp = client.get("/api/cma/dashboard/alert-stats", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["total_pending"] == 0
|
|
assert data["by_severity"] == {"red": 0, "yellow": 0, "green": 0}
|
|
assert data["by_dimension"] == []
|
|
|
|
def test_alert_stats_with_data(self, client: TestClient, db: Session):
|
|
"""有预警时的预警统计"""
|
|
user = create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
kpi = create_test_kpi(db, dimension="finance")
|
|
kpi2 = create_test_kpi(db, kpi_code="C_CODE", kpi_name="客户KPI", dimension="customer")
|
|
alert1 = KPIAlert(kpi_id=kpi.id, alert_level="red", alert_message="红色预警", status="pending")
|
|
alert2 = KPIAlert(kpi_id=kpi.id, alert_level="yellow", alert_message="黄色预警", status="pending")
|
|
alert3 = KPIAlert(kpi_id=kpi2.id, alert_level="yellow", alert_message="客户预警", status="pending")
|
|
db.add_all([alert1, alert2, alert3])
|
|
db.commit()
|
|
|
|
resp = client.get("/api/cma/dashboard/alert-stats", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["total_pending"] == 3
|
|
assert data["by_severity"].get("red") == 1
|
|
assert data["by_severity"].get("yellow") == 2
|
|
assert len(data["by_dimension"]) == 2
|
|
dim_dict = {d["dimension"]: d["count"] for d in data["by_dimension"]}
|
|
assert dim_dict.get("finance") == 2
|
|
assert dim_dict.get("customer") == 1
|
|
|
|
def test_my_dashboard(self, client: TestClient, db: Session):
|
|
"""个人工作台接口"""
|
|
user = create_test_user(db, username="ceo_user", name="CEO", role="ceo")
|
|
token = get_token_for_user(client, username="ceo_user", password="admin123")
|
|
kpi = create_test_kpi(db)
|
|
|
|
resp = client.get("/api/cma/dashboard/my-dashboard", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "kpis" in data
|
|
assert "action_plans" in data
|
|
assert "reminders" in data
|
|
|
|
def test_predict_empty(self, client: TestClient, db: Session):
|
|
"""无数据时预测返回空列表"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
|
|
resp = client.get("/api/cma/dashboard/predict", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["predictions"] == []
|
|
|
|
|
|
# ── Epic 2 新增接口测试 ──────────────────────────
|
|
|
|
class TestTrendAnalysisPost:
|
|
"""POST /api/cma/dashboard/trend-analysis"""
|
|
|
|
def test_trend_analysis_with_kpi_ids(self, client: TestClient, db: Session):
|
|
"""指定KPI ID进行趋势分析"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
kpi = create_test_kpi(db)
|
|
db.add(KPIValue(kpi_id=kpi.id, period="2026-05", actual_value=80.0))
|
|
db.add(KPIValue(kpi_id=kpi.id, period="2026-06", actual_value=85.0))
|
|
db.commit()
|
|
|
|
resp = client.post("/api/cma/dashboard/trend-analysis",
|
|
headers=auth_header(token),
|
|
json={"kpi_ids": [kpi.id], "period_type": "month", "compare_type": "mom"})
|
|
assert resp.status_code == 200
|
|
assert len(resp.json()) > 0
|
|
|
|
def test_trend_analysis_no_kpi_ids(self, client: TestClient, db: Session):
|
|
"""不传KPI ID时默认取所有活跃KPI"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
kpi = create_test_kpi(db)
|
|
db.add(KPIValue(kpi_id=kpi.id, period="2026-06", actual_value=85.0))
|
|
db.commit()
|
|
|
|
resp = client.post("/api/cma/dashboard/trend-analysis",
|
|
headers=auth_header(token),
|
|
json={"period_type": "month", "compare_type": "mom"})
|
|
assert resp.status_code == 200
|
|
assert len(resp.json()) >= 1
|
|
|
|
def test_trend_analysis_inactive_kpi(self, client: TestClient, db: Session):
|
|
"""指定不存在的KPI ID时返回空"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
create_test_kpi(db)
|
|
|
|
resp = client.post("/api/cma/dashboard/trend-analysis",
|
|
headers=auth_header(token),
|
|
json={"kpi_ids": [9999], "period_type": "month", "compare_type": "mom"})
|
|
assert resp.status_code == 200
|
|
assert resp.json().get("data") == []
|
|
|
|
def test_trend_analysis_unauthorized(self, client: TestClient, db: Session):
|
|
"""未认证无法访问"""
|
|
resp = client.post("/api/cma/dashboard/trend-analysis", json={})
|
|
assert resp.status_code == 403
|
|
|
|
|
|
class TestGetTrendAnalysis:
|
|
"""GET /api/cma/dashboard/trend-analysis"""
|
|
|
|
def test_get_trend_with_ids(self, client: TestClient, db: Session):
|
|
"""指定KPI查询趋势对比数据"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
kpi = create_test_kpi(db)
|
|
db.add(KPIValue(kpi_id=kpi.id, period="2026-04", actual_value=70.0))
|
|
db.add(KPIValue(kpi_id=kpi.id, period="2026-05", actual_value=80.0))
|
|
db.add(KPIValue(kpi_id=kpi.id, period="2026-06", actual_value=85.0))
|
|
db.commit()
|
|
|
|
resp = client.get(f"/api/cma/dashboard/trend-analysis?kpi_ids={kpi.id}",
|
|
headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert len(data["data"]) == 1
|
|
assert data["data"][0]["kpi_name"] == "测试KPI"
|
|
assert len(data["data"][0]["data"]) == 3
|
|
|
|
def test_get_trend_no_ids(self, client: TestClient, db: Session):
|
|
"""不传KPI ID返回空"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
|
|
resp = client.get("/api/cma/dashboard/trend-analysis",
|
|
headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
assert resp.json()["data"] == []
|
|
|
|
|
|
class TestPredictWithData:
|
|
"""GET /api/cma/dashboard/predict (有数据)"""
|
|
|
|
def test_predict_with_enough_data(self, client: TestClient, db: Session):
|
|
"""有足够数据点(>=3)时进行预测"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
kpi = create_test_kpi(db)
|
|
for i, val in enumerate([60.0, 65.0, 70.0, 75.0]):
|
|
db.add(KPIValue(kpi_id=kpi.id, period=f"2026-{3+i:02d}", actual_value=val))
|
|
db.commit()
|
|
|
|
resp = client.get("/api/cma/dashboard/predict", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert len(data["predictions"]) >= 1
|
|
assert data["predictable_count"] >= 1
|
|
|
|
|
|
class TestKpisEnhanced:
|
|
"""GET /api/cma/dashboard/kpis/enhanced"""
|
|
|
|
def test_enhanced_with_data(self, client: TestClient, db: Session):
|
|
"""增强版KPI列表"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
kpi = create_test_kpi(db)
|
|
db.add(KPIValue(kpi_id=kpi.id, period="2026-05", actual_value=80.0))
|
|
db.add(KPIValue(kpi_id=kpi.id, period="2026-06", actual_value=85.0))
|
|
db.commit()
|
|
|
|
resp = client.get("/api/cma/dashboard/kpis/enhanced", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert len(data["data"]) == 1
|
|
assert data["data"][0]["kpi_name"] == "测试KPI"
|
|
|
|
def test_enhanced_empty(self, client: TestClient, db: Session):
|
|
"""无KPI时返回空"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
|
|
resp = client.get("/api/cma/dashboard/kpis/enhanced", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
assert resp.json()["data"] == []
|
|
|
|
|
|
class TestKpiTrend:
|
|
"""GET /api/cma/dashboard/kpi-trend"""
|
|
|
|
def test_kpi_trend_with_ids(self, client: TestClient, db: Session):
|
|
"""KPI趋势分析详细版"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
kpi = create_test_kpi(db)
|
|
db.add(KPIValue(kpi_id=kpi.id, period="2026-04", actual_value=70.0))
|
|
db.add(KPIValue(kpi_id=kpi.id, period="2026-05", actual_value=80.0))
|
|
db.add(KPIValue(kpi_id=kpi.id, period="2026-06", actual_value=85.0))
|
|
db.commit()
|
|
|
|
resp = client.get(f"/api/cma/dashboard/kpi-trend?kpi_ids={kpi.id}",
|
|
headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert len(data["kpis"]) == 1
|
|
assert len(data["kpis"][0]["periods"]) == 3
|
|
assert "summary" in data
|
|
|
|
def test_kpi_trend_no_ids(self, client: TestClient, db: Session):
|
|
"""不传ID时默认取前5个活跃KPI"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
kpi = create_test_kpi(db)
|
|
db.add(KPIValue(kpi_id=kpi.id, period="2026-06", actual_value=85.0))
|
|
db.commit()
|
|
|
|
resp = client.get("/api/cma/dashboard/kpi-trend", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
assert len(resp.json()["kpis"]) >= 1
|
|
|
|
|
|
class TestKpiComparison:
|
|
"""GET /api/cma/dashboard/kpi-comparison"""
|
|
|
|
def test_kpi_comparison(self, client: TestClient, db: Session):
|
|
"""KPI多区间对比"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
kpi = create_test_kpi(db)
|
|
db.add(KPIValue(kpi_id=kpi.id, period="2026-05", actual_value=80.0))
|
|
db.add(KPIValue(kpi_id=kpi.id, period="2026-06", actual_value=85.0))
|
|
db.commit()
|
|
|
|
resp = client.get(f"/api/cma/dashboard/kpi-comparison?kpi_id={kpi.id}",
|
|
headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["kpi"]["kpi_name"] == "测试KPI"
|
|
assert "comparisons" in data
|
|
|
|
def test_kpi_comparison_missing_id(self, client: TestClient, db: Session):
|
|
"""缺少必填kpi_id参数"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
|
|
resp = client.get("/api/cma/dashboard/kpi-comparison",
|
|
headers=auth_header(token))
|
|
assert resp.status_code == 422
|
|
|
|
def test_kpi_comparison_not_found(self, client: TestClient, db: Session):
|
|
"""不存在的KPI ID"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
|
|
resp = client.get("/api/cma/dashboard/kpi-comparison?kpi_id=9999",
|
|
headers=auth_header(token))
|
|
assert resp.status_code == 404
|
|
|
|
|
|
class TestAlertTrend:
|
|
"""GET /api/cma/dashboard/alert-trend"""
|
|
|
|
def test_alert_trend_empty(self, client: TestClient, db: Session):
|
|
"""无预警时的趋势"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
|
|
resp = client.get("/api/cma/dashboard/alert-trend?days=30",
|
|
headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "daily_alerts" in data
|
|
assert "summary" in data
|
|
|
|
def test_alert_trend_with_data(self, client: TestClient, db: Session):
|
|
"""有预警时的按天分布"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
kpi = create_test_kpi(db)
|
|
db.add(KPIAlert(kpi_id=kpi.id, alert_level="red", alert_message="测试",
|
|
status="pending", created_at=datetime.now()))
|
|
db.commit()
|
|
|
|
resp = client.get("/api/cma/dashboard/alert-trend?days=30",
|
|
headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "daily_alerts" in data
|
|
|
|
|
|
class TestExport:
|
|
"""GET /api/cma/dashboard/export"""
|
|
|
|
def test_export_csv(self, client: TestClient, db: Session):
|
|
"""导出CSV文件"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
kpi = create_test_kpi(db)
|
|
db.add(KPIValue(kpi_id=kpi.id, period="2026-06", actual_value=85.0))
|
|
db.commit()
|
|
|
|
resp = client.get("/api/cma/dashboard/export", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
ct = resp.headers.get("content-type", "")
|
|
assert "csv" in ct or "text" in ct or "plain" in ct
|
|
|
|
def test_export_with_kpi_ids(self, client: TestClient, db: Session):
|
|
"""带KPI ID过滤的导出"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
kpi = create_test_kpi(db)
|
|
db.add(KPIValue(kpi_id=kpi.id, period="2026-06", actual_value=85.0))
|
|
db.commit()
|
|
|
|
resp = client.get(f"/api/cma/dashboard/export?kpi_ids={kpi.id}",
|
|
headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
body = resp.text
|
|
assert "TEST_001" in body or "测试KPI" in body
|
|
|
|
def test_export_empty(self, client: TestClient, db: Session):
|
|
"""无KPI时的导出"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
|
|
resp = client.get("/api/cma/dashboard/export", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
assert "KPI编码" in resp.text
|