240 lines
11 KiB
Python
240 lines
11 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"] >= 0
|
|
assert isinstance(data["dimension_stats"], list)
|
|
|
|
def test_kpis_empty(self, client: TestClient, db: Session):
|
|
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
|
|
assert resp.json()["data"] == []
|
|
|
|
def test_kpis_with_data(self, client: TestClient, db: Session):
|
|
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 "target_value" in data["data"][0]
|
|
|
|
def test_kpis_with_alert(self, client: TestClient, db: Session):
|
|
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 "alert_level" in data["data"][0]
|
|
|
|
def test_my_kpis(self, client: TestClient, db: Session):
|
|
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
|
|
|
|
def test_my_kpis_ceo_sees_all(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, kpi_code="F_KPI_A", responsible_user="张三")
|
|
resp = client.get("/api/cma/dashboard/my-kpis", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
assert resp.json()["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 == 404
|
|
|
|
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")
|
|
resp = client.get("/api/cma/dashboard/alert-stats", headers=auth_header(token))
|
|
assert resp.status_code == 404
|
|
|
|
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
|
|
assert resp.json()["predictions"] == []
|
|
|
|
|
|
class TestTrendAnalysisPost:
|
|
"""POST /api/cma/dashboard/trend-analysis (端点已移除)"""
|
|
def test_trend_analysis_with_kpi_ids(self, client, db):
|
|
create_test_user(db); t = get_token_for_user(client)
|
|
r = client.post("/api/cma/dashboard/trend-analysis", headers=auth_header(t), json={})
|
|
assert r.status_code == 404
|
|
def test_trend_analysis_no_kpi_ids(self, client, db):
|
|
create_test_user(db); t = get_token_for_user(client)
|
|
r = client.post("/api/cma/dashboard/trend-analysis", headers=auth_header(t), json={})
|
|
assert r.status_code == 404
|
|
def test_trend_analysis_inactive_kpi(self, client, db):
|
|
create_test_user(db); t = get_token_for_user(client)
|
|
r = client.post("/api/cma/dashboard/trend-analysis", headers=auth_header(t), json={})
|
|
assert r.status_code == 404
|
|
def test_trend_analysis_unauthorized(self, client, db):
|
|
r = client.post("/api/cma/dashboard/trend-analysis", json={})
|
|
assert r.status_code in (401, 403, 404)
|
|
|
|
|
|
class TestGetTrendAnalysis:
|
|
"""GET /api/cma/dashboard/trend-analysis (端点已移除)"""
|
|
def test_get_trend_with_ids(self, client, db):
|
|
create_test_user(db); t = get_token_for_user(client)
|
|
r = client.get("/api/cma/dashboard/trend-analysis?kpi_ids=1", headers=auth_header(t))
|
|
assert r.status_code == 404
|
|
def test_get_trend_no_ids(self, client, db):
|
|
create_test_user(db); t = get_token_for_user(client)
|
|
r = client.get("/api/cma/dashboard/trend-analysis", headers=auth_header(t))
|
|
assert r.status_code == 404
|
|
|
|
|
|
class TestPredictWithData:
|
|
"""GET /api/cma/dashboard/predict"""
|
|
def test_predict_with_enough_data(self, client, db):
|
|
create_test_user(db)
|
|
t = 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()
|
|
r = client.get("/api/cma/dashboard/predict", headers=auth_header(t))
|
|
assert r.status_code == 200
|
|
data = r.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, db):
|
|
create_test_user(db); t = get_token_for_user(client)
|
|
r = client.get("/api/cma/dashboard/kpis/enhanced", headers=auth_header(t))
|
|
assert r.status_code == 404
|
|
def test_enhanced_empty(self, client, db):
|
|
create_test_user(db); t = get_token_for_user(client)
|
|
r = client.get("/api/cma/dashboard/kpis/enhanced", headers=auth_header(t))
|
|
assert r.status_code == 404
|
|
|
|
|
|
class TestKpiTrend:
|
|
"""GET /api/cma/dashboard/kpi-trend (端点已移除)"""
|
|
def test_kpi_trend_with_ids(self, client, db):
|
|
create_test_user(db); t = get_token_for_user(client)
|
|
r = client.get("/api/cma/dashboard/kpi-trend?kpi_ids=1", headers=auth_header(t))
|
|
assert r.status_code == 404
|
|
def test_kpi_trend_no_ids(self, client, db):
|
|
create_test_user(db); t = get_token_for_user(client)
|
|
r = client.get("/api/cma/dashboard/kpi-trend", headers=auth_header(t))
|
|
assert r.status_code == 404
|
|
|
|
|
|
class TestKpiComparison:
|
|
"""GET /api/cma/dashboard/kpi-comparison (端点已移除)"""
|
|
def test_kpi_comparison(self, client, db):
|
|
create_test_user(db); t = get_token_for_user(client)
|
|
r = client.get("/api/cma/dashboard/kpi-comparison?kpi_id=1", headers=auth_header(t))
|
|
assert r.status_code == 404
|
|
def test_kpi_comparison_missing_id(self, client, db):
|
|
create_test_user(db); t = get_token_for_user(client)
|
|
r = client.get("/api/cma/dashboard/kpi-comparison", headers=auth_header(t))
|
|
assert r.status_code == 404
|
|
def test_kpi_comparison_not_found(self, client, db):
|
|
create_test_user(db); t = get_token_for_user(client)
|
|
r = client.get("/api/cma/dashboard/kpi-comparison?kpi_id=9999", headers=auth_header(t))
|
|
assert r.status_code == 404
|
|
|
|
|
|
class TestAlertTrend:
|
|
"""GET /api/cma/dashboard/alert-trend (端点已移除)"""
|
|
def test_alert_trend_empty(self, client, db):
|
|
create_test_user(db); t = get_token_for_user(client)
|
|
r = client.get("/api/cma/dashboard/alert-trend?days=30", headers=auth_header(t))
|
|
assert r.status_code == 404
|
|
def test_alert_trend_with_data(self, client, db):
|
|
create_test_user(db); t = 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()
|
|
r = client.get("/api/cma/dashboard/alert-trend?days=30", headers=auth_header(t))
|
|
assert r.status_code == 404
|
|
|
|
|
|
class TestExport:
|
|
"""GET /api/cma/dashboard/export (端点已移除)"""
|
|
def test_export_csv(self, client, db):
|
|
create_test_user(db); t = get_token_for_user(client)
|
|
r = client.get("/api/cma/dashboard/export", headers=auth_header(t))
|
|
assert r.status_code == 404
|
|
def test_export_with_kpi_ids(self, client, db):
|
|
create_test_user(db); t = get_token_for_user(client)
|
|
r = client.get("/api/cma/dashboard/export?kpi_ids=1", headers=auth_header(t))
|
|
assert r.status_code == 404
|
|
def test_export_empty(self, client, db):
|
|
create_test_user(db); t = get_token_for_user(client)
|
|
r = client.get("/api/cma/dashboard/export", headers=auth_header(t))
|
|
assert r.status_code == 404
|