66 lines
2.5 KiB
Python
66 lines
2.5 KiB
Python
"""routes_alerts 薄层测试(本地跑, CI 不含 tests/api): 列表筛选/summary/ack 三态."""
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from sanguo_api.main import build_app
|
|
from sanguo_api.auth import create_token, set_jwt_config
|
|
from sanguo_data import alerts_store as store
|
|
|
|
|
|
def _cfg(tmp_path) -> str:
|
|
cfg = tmp_path / "bt.yaml"
|
|
cfg.write_text(
|
|
"backtest:\n max_workers: 1\n db_path: %s\n file_dir: %s\n"
|
|
"api:\n host: 0.0.0.0\n port: 8000\n"
|
|
"auth:\n username: admin\n password_hash: x\n jwt_secret: s\n token_expire_minutes: 60\n"
|
|
"pool:\n max_workers: 1\n" % (tmp_path / "r.db", tmp_path / "f")
|
|
)
|
|
return str(cfg)
|
|
|
|
|
|
@pytest.fixture()
|
|
def client(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("SANGUO_ALERTS_DB", str(tmp_path / "alerts.db"))
|
|
with store.connect(str(tmp_path / "alerts.db")) as conn:
|
|
store.upsert_alert(conn, {
|
|
"alert_id": "data-t-1-20260925", "check_key": "data-t-1",
|
|
"source": "data", "host": "vps", "severity": "red",
|
|
"title": "t1", "detail": "", "evidence": {"x": 1}})
|
|
set_jwt_config(secret="s", expire_minutes=60)
|
|
app = build_app(_cfg(tmp_path))
|
|
c = TestClient(app)
|
|
c.headers.update({"Authorization": f"Bearer {create_token('admin')}"})
|
|
return c
|
|
|
|
|
|
def test_list_and_summary(client):
|
|
r = client.get("/api/v1/alerts")
|
|
assert r.status_code == 200 and r.json()["total"] == 1
|
|
assert r.json()["items"][0]["evidence"] == {"x": 1}
|
|
r = client.get("/api/v1/alerts/summary")
|
|
assert r.json() == {"open_red": 1,
|
|
"by_source": {"data": 1, "strategy": 0,
|
|
"factor": 0, "infra": 0}}
|
|
|
|
|
|
def test_list_filter_param(client):
|
|
assert client.get("/api/v1/alerts",
|
|
params={"severity": "yellow"}).json()["total"] == 0
|
|
|
|
|
|
def test_ack_flow(client):
|
|
r = client.post("/api/v1/alerts/data-t-1-20260925/ack")
|
|
assert r.status_code == 200 and r.json()["status"] == "acked"
|
|
# 已 ack(非 open) 再 ack → 400 语义(resolved/no-op 同通道)
|
|
r2 = client.post("/api/v1/alerts/data-t-1-20260925/ack")
|
|
assert r2.status_code == 400
|
|
r3 = client.post("/api/v1/alerts/nope/ack")
|
|
assert r3.status_code == 404
|
|
|
|
|
|
def test_unauthorized_rejected(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("SANGUO_ALERTS_DB", str(tmp_path / "alerts.db"))
|
|
set_jwt_config(secret="s", expire_minutes=60)
|
|
c = TestClient(build_app(_cfg(tmp_path)))
|
|
assert c.get("/api/v1/alerts").status_code == 401
|