88 lines
3.5 KiB
Python
88 lines
3.5 KiB
Python
|
"""告警引擎与运营看板单元测试（内存 SQLite）。"""
|
||
|
|
from datetime import datetime, timedelta
|
||
|
|
from types import SimpleNamespace
|
||
|
|
from unittest.mock import patch
|
||
|
|
|
||
|
|
from sqlmodel import Session, SQLModel, create_engine, select
|
||
|
|
|
||
|
|
from app.models import (Campaign, Notification, PointAccount, TaskQueueItem)
|
||
|
|
from app import ops
|
||
|
|
|
||
|
|
|
||
|
|
def _session():
|
||
|
|
engine = create_engine("sqlite://", connect_args={"check_same_thread": False})
|
||
|
|
SQLModel.metadata.create_all(engine)
|
||
|
|
return Session(engine)
|
||
|
|
|
||
|
|
|
||
|
|
def test_alert_dedup_within_window():
|
||
|
|
with _session() as s:
|
||
|
|
assert ops.raise_alert(s, "queue-dead", "队列告警：1 个死信 [queue-dead]") is True
|
||
|
|
assert ops.raise_alert(s, "queue-dead", "队列告警：2 个死信 [queue-dead]") is False
|
||
|
|
rows = s.exec(select(Notification)).all()
|
||
|
|
assert len(rows) == 1
|
||
|
|
|
||
|
|
|
||
|
|
def test_alert_dedup_expires():
|
||
|
|
with _session() as s:
|
||
|
|
old = (datetime.now() - timedelta(minutes=15)).isoformat()
|
||
|
|
s.add(Notification(type="alert", ref_id=0,
|
||
|
|
message="队列告警 [queue-dead]", created_at=old))
|
||
|
|
s.commit()
|
||
|
|
assert ops.raise_alert(s, "queue-dead", "队列告警 [queue-dead]") is True
|
||
|
|
|
||
|
|
|
||
|
|
def test_sweep_fires_dead_letter_alert():
|
||
|
|
with _session() as s:
|
||
|
|
s.add(TaskQueueItem(task_key="run:1", status="dead", error="x", attempt=3))
|
||
|
|
s.commit()
|
||
|
|
fired = ops.run_alert_sweep(s)
|
||
|
|
assert "queue-dead" in fired
|
||
|
|
alerts = s.exec(select(Notification).where(Notification.type == "alert")).all()
|
||
|
|
assert any("死信" in a.message for a in alerts)
|
||
|
|
|
||
|
|
|
||
|
|
def test_sweep_fires_low_points():
|
||
|
|
with _session() as s:
|
||
|
|
s.add(PointAccount(user_id=1, organization_id=1, balance_points=10))
|
||
|
|
s.commit()
|
||
|
|
fired = ops.run_alert_sweep(s)
|
||
|
|
assert "points-low" in fired
|
||
|
|
|
||
|
|
|
||
|
|
def test_sweep_healthy_no_alerts():
|
||
|
|
with _session() as s:
|
||
|
|
# 99.9 万点：高于 points-low(1000) 也高于充值提醒阈值(100万以下不算健康)
|
||
|
|
s.add(PointAccount(user_id=1, organization_id=1, balance_points=2_000_000))
|
||
|
|
s.commit()
|
||
|
|
fired = ops.run_alert_sweep(s)
|
||
|
|
assert fired == []
|
||
|
|
|
||
|
|
|
||
|
|
def test_dashboard_shape():
|
||
|
|
with _session() as s:
|
||
|
|
s.add(PointAccount(user_id=1, organization_id=1, balance_points=500000))
|
||
|
|
s.add(Campaign(name="任务A", status="done", has_failure=False))
|
||
|
|
s.commit()
|
||
|
|
data = ops.ops_dashboard(s)
|
||
|
|
assert set(data.keys()) >= {"queue", "campaigns", "cost_7d", "knowledge",
|
||
|
|
"providers", "dead_letter", "recent_alerts"}
|
||
|
|
assert data["campaigns"]["done"] == 1
|
||
|
|
# 点数池已从看板移除（用户要求：运营看板不出现点数余额）
|
||
|
|
assert "points" not in data["cost_7d"]
|
||
|
|
|
||
|
|
|
||
|
|
def test_dashboard_isolation_for_non_admin():
|
||
|
|
"""普通用户只能看到自己的任务/用量/告警，且拿不到系统级信息。"""
|
||
|
|
with _session() as s:
|
||
|
|
s.add(Campaign(name="我的任务", status="done", has_failure=False,
|
||
|
|
owner_user_id=7, organization_id=1))
|
||
|
|
s.add(Campaign(name="别人的任务", status="done", has_failure=False,
|
||
|
|
owner_user_id=8, organization_id=1))
|
||
|
|
s.commit()
|
||
|
|
mine = ops.ops_dashboard(s, owner_user_id=7)
|
||
|
|
assert mine["scope"] == "self"
|
||
|
|
assert mine["campaigns"]["total"] == 1 and mine["campaigns"]["done"] == 1
|
||
|
|
# 系统级信息对普通用户置空
|
||
|
|
assert mine["providers"] == [] and mine["backup"] == {}
|