275 lines
10 KiB
Python
275 lines
10 KiB
Python
"""
|
|
预警闭环模块测试
|
|
"""
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy.orm import Session
|
|
from datetime import datetime, timedelta
|
|
from tests.conftest import create_test_user, get_token_for_user, auth_header, create_test_kpi
|
|
from app.models import KPIAlert, ActionPlan
|
|
|
|
|
|
def create_test_alert(db: Session, **kwargs) -> KPIAlert:
|
|
"""创建测试预警"""
|
|
defaults = {
|
|
"kpi_id": 1,
|
|
"alert_level": "yellow",
|
|
"alert_message": "测试预警消息",
|
|
"status": "pending",
|
|
"assignee": None,
|
|
"created_at": datetime.now(),
|
|
}
|
|
defaults.update(kwargs)
|
|
alert = KPIAlert(**defaults)
|
|
db.add(alert)
|
|
db.commit()
|
|
db.refresh(alert)
|
|
return alert
|
|
|
|
|
|
class TestAlerts:
|
|
"""预警闭环测试"""
|
|
|
|
def test_list_alerts_empty(self, client: TestClient, db: Session):
|
|
"""空预警列表"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
|
|
resp = client.get("/api/cma/alerts", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["total"] == 0
|
|
assert data["data"] == []
|
|
|
|
def test_list_alerts_with_data(self, client: TestClient, db: Session):
|
|
"""有预警数据时列表正常"""
|
|
user = create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
kpi = create_test_kpi(db)
|
|
alert = create_test_alert(db, kpi_id=kpi.id, alert_message="营收下滑预警")
|
|
|
|
resp = client.get("/api/cma/alerts", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["total"] == 1
|
|
assert data["data"][0]["alert_message"] == "营收下滑预警"
|
|
|
|
def test_list_alerts_filter_by_status(self, client: TestClient, db: Session):
|
|
"""按状态过滤预警"""
|
|
user = create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
kpi = create_test_kpi(db)
|
|
create_test_alert(db, kpi_id=kpi.id, alert_message="黄色预警", status="pending")
|
|
create_test_alert(db, kpi_id=kpi.id, alert_message="处理中", status="processing")
|
|
create_test_alert(db, kpi_id=kpi.id, alert_message="已解决", status="resolved")
|
|
|
|
resp = client.get("/api/cma/alerts?status=pending", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["total"] == 1
|
|
assert data["data"][0]["status"] == "pending"
|
|
|
|
def test_process_alert_success(self, client: TestClient, db: Session):
|
|
"""标记预警为处理中"""
|
|
user = create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
kpi = create_test_kpi(db)
|
|
alert = create_test_alert(db, kpi_id=kpi.id)
|
|
|
|
resp = client.post(
|
|
f"/api/cma/alerts/{alert.id}/process",
|
|
headers=auth_header(token),
|
|
json={"assignee": "张三"},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["message"] == "已标记为处理中"
|
|
assert resp.json()["alert"]["status"] == "processing"
|
|
assert resp.json()["alert"]["assignee"] == "张三"
|
|
|
|
def test_process_alert_missing_assignee(self, client: TestClient, db: Session):
|
|
"""缺少处理人参数"""
|
|
user = create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
kpi = create_test_kpi(db)
|
|
alert = create_test_alert(db, kpi_id=kpi.id)
|
|
|
|
resp = client.post(
|
|
f"/api/cma/alerts/{alert.id}/process",
|
|
headers=auth_header(token),
|
|
json={},
|
|
)
|
|
assert resp.status_code == 400
|
|
assert "缺少" in resp.json()["detail"]
|
|
|
|
def test_process_alert_not_found(self, client: TestClient, db: Session):
|
|
"""处理不存在的预警"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
|
|
resp = client.post(
|
|
"/api/cma/alerts/9999/process",
|
|
headers=auth_header(token),
|
|
json={"assignee": "张三"},
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
def test_process_resolved_alert_fails(self, client: TestClient, db: Session):
|
|
"""已处理的预警不能重复处理"""
|
|
user = create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
kpi = create_test_kpi(db)
|
|
alert = create_test_alert(db, kpi_id=kpi.id, status="resolved")
|
|
|
|
resp = client.post(
|
|
f"/api/cma/alerts/{alert.id}/process",
|
|
headers=auth_header(token),
|
|
json={"assignee": "张三"},
|
|
)
|
|
assert resp.status_code == 400
|
|
assert "已处理" in resp.json()["detail"]
|
|
|
|
def test_resolve_alert(self, client: TestClient, db: Session):
|
|
"""解决预警"""
|
|
user = create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
kpi = create_test_kpi(db)
|
|
alert = create_test_alert(db, kpi_id=kpi.id)
|
|
|
|
resp = client.post(
|
|
f"/api/cma/alerts/{alert.id}/resolve",
|
|
headers=auth_header(token),
|
|
json={"assignee": "李四", "resolution": "已采取措施改善"},
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["message"] == "已处理"
|
|
assert data["alert"]["status"] == "resolved"
|
|
assert data["suggest_create_action_plan"] is True
|
|
|
|
def test_resolve_alert_not_found(self, client: TestClient, db: Session):
|
|
"""解决不存在的预警"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
|
|
resp = client.post(
|
|
"/api/cma/alerts/9999/resolve",
|
|
headers=auth_header(token),
|
|
json={},
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
def test_escalate_yellow_to_red(self, client: TestClient, db: Session):
|
|
"""升级黄色预警为红色"""
|
|
user = create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
kpi = create_test_kpi(db)
|
|
alert = create_test_alert(db, kpi_id=kpi.id, alert_level="yellow")
|
|
|
|
resp = client.post(
|
|
f"/api/cma/alerts/{alert.id}/escalate",
|
|
headers=auth_header(token),
|
|
json={"assignee": "上级主管"},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["message"] == "已升级"
|
|
assert resp.json()["alert"]["alert_level"] == "red"
|
|
|
|
def test_escalate_red_stays_red(self, client: TestClient, db: Session):
|
|
"""已经是红色的预警保持红色"""
|
|
user = create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
kpi = create_test_kpi(db)
|
|
alert = create_test_alert(db, kpi_id=kpi.id, alert_level="red")
|
|
|
|
resp = client.post(
|
|
f"/api/cma/alerts/{alert.id}/escalate",
|
|
headers=auth_header(token),
|
|
json={"assignee": "上级主管"},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["alert"]["alert_level"] == "red"
|
|
|
|
def test_escalate_missing_assignee(self, client: TestClient, db: Session):
|
|
"""升级预警缺少处理人"""
|
|
user = create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
kpi = create_test_kpi(db)
|
|
alert = create_test_alert(db, kpi_id=kpi.id)
|
|
|
|
resp = client.post(
|
|
f"/api/cma/alerts/{alert.id}/escalate",
|
|
headers=auth_header(token),
|
|
json={},
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
def test_check_timeout(self, client: TestClient, db: Session):
|
|
"""超时预警检测"""
|
|
user = create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
kpi = create_test_kpi(db)
|
|
# 创建超过24小时且仍为pending的预警
|
|
old_time = datetime.now() - timedelta(hours=48)
|
|
create_test_alert(db, kpi_id=kpi.id, created_at=old_time, alert_level="yellow")
|
|
# 创建一个未超时的
|
|
create_test_alert(db, kpi_id=kpi.id, alert_level="yellow")
|
|
|
|
resp = client.get("/api/cma/alerts/check-timeout", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["total_timeout"] == 1
|
|
assert data["upgraded_count"] == 1
|
|
|
|
def test_create_action_plan_from_alert(self, client: TestClient, db: Session):
|
|
"""从预警创建改善行动计划"""
|
|
user = create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
kpi = create_test_kpi(db)
|
|
alert = create_test_alert(db, kpi_id=kpi.id, alert_level="red", alert_message="营收严重下滑")
|
|
|
|
resp = client.post(
|
|
f"/api/cma/alerts/{alert.id}/create-action-plan",
|
|
headers=auth_header(token),
|
|
json={"assignee": "王五", "created_by": "系统"},
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["message"] == "改善行动计划已创建"
|
|
assert data["plan_id"] is not None
|
|
assert data["priority"] == "high" # red → high
|
|
|
|
def test_create_action_plan_duplicate(self, client: TestClient, db: Session):
|
|
"""重复创建行动计划被拒绝"""
|
|
user = create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
kpi = create_test_kpi(db)
|
|
alert = create_test_alert(db, kpi_id=kpi.id, alert_message="测试预警")
|
|
|
|
# 第一次创建
|
|
client.post(
|
|
f"/api/cma/alerts/{alert.id}/create-action-plan",
|
|
headers=auth_header(token),
|
|
json={"assignee": "王五"},
|
|
)
|
|
|
|
# 重复创建
|
|
resp = client.post(
|
|
f"/api/cma/alerts/{alert.id}/create-action-plan",
|
|
headers=auth_header(token),
|
|
json={"assignee": "王五"},
|
|
)
|
|
assert resp.status_code == 400
|
|
assert "已关联" in resp.json()["detail"]
|
|
|
|
def test_create_action_plan_alert_not_found(self, client: TestClient, db: Session):
|
|
"""基于不存在的预警创建行动计划"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
|
|
resp = client.post(
|
|
"/api/cma/alerts/9999/create-action-plan",
|
|
headers=auth_header(token),
|
|
json={"assignee": "王五"},
|
|
)
|
|
assert resp.status_code == 404
|