126 lines
5.0 KiB
Python
126 lines
5.0 KiB
Python
import pytest
|
|
from conftest import auth_header
|
|
|
|
|
|
class TestAlarmRules:
|
|
async def test_list_alarm_rules(self, client, admin_user, admin_token, seed_alarm_rule):
|
|
resp = await client.get("/api/v1/alarms/rules", headers=auth_header(admin_token))
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert isinstance(body, list)
|
|
assert len(body) >= 1
|
|
|
|
async def test_create_alarm_rule(self, client, admin_user, admin_token):
|
|
resp = await client.post(
|
|
"/api/v1/alarms/rules",
|
|
json={
|
|
"name": "新告警规则", "data_type": "power", "condition": "gt",
|
|
"threshold": 100.0, "severity": "critical",
|
|
},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["name"] == "新告警规则"
|
|
assert body["severity"] == "critical"
|
|
|
|
async def test_create_alarm_rule_as_visitor_forbidden(self, client, normal_user, user_token):
|
|
resp = await client.post(
|
|
"/api/v1/alarms/rules",
|
|
json={"name": "Test", "data_type": "power", "condition": "gt", "threshold": 50.0},
|
|
headers=auth_header(user_token),
|
|
)
|
|
assert resp.status_code == 403
|
|
|
|
async def test_update_alarm_rule(self, client, admin_user, admin_token, seed_alarm_rule):
|
|
resp = await client.put(
|
|
f"/api/v1/alarms/rules/{seed_alarm_rule.id}",
|
|
json={
|
|
"name": "更新后规则", "data_type": "temperature", "condition": "gt",
|
|
"threshold": 90.0, "severity": "critical",
|
|
},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["name"] == "更新后规则"
|
|
|
|
async def test_update_nonexistent_rule(self, client, admin_user, admin_token):
|
|
resp = await client.put(
|
|
"/api/v1/alarms/rules/99999",
|
|
json={"name": "Ghost", "data_type": "power", "condition": "gt"},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
async def test_delete_alarm_rule(self, client, admin_user, admin_token, seed_alarm_rule):
|
|
resp = await client.delete(
|
|
f"/api/v1/alarms/rules/{seed_alarm_rule.id}",
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
async def test_delete_nonexistent_rule(self, client, admin_user, admin_token):
|
|
resp = await client.delete("/api/v1/alarms/rules/99999", headers=auth_header(admin_token))
|
|
assert resp.status_code == 404
|
|
|
|
|
|
class TestAlarmEvents:
|
|
async def test_list_alarm_events(self, client, admin_user, admin_token, seed_alarm_event):
|
|
resp = await client.get("/api/v1/alarms/events", headers=auth_header(admin_token))
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert "total" in body
|
|
assert "items" in body
|
|
assert body["total"] >= 1
|
|
|
|
async def test_list_alarm_events_filter_status(self, client, admin_user, admin_token, seed_alarm_event):
|
|
resp = await client.get(
|
|
"/api/v1/alarms/events",
|
|
params={"status": "active"},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 200
|
|
for item in resp.json()["items"]:
|
|
assert item["status"] == "active"
|
|
|
|
async def test_acknowledge_alarm(self, client, admin_user, admin_token, seed_alarm_event):
|
|
resp = await client.post(
|
|
f"/api/v1/alarms/events/{seed_alarm_event.id}/acknowledge",
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
async def test_acknowledge_nonexistent_alarm(self, client, admin_user, admin_token):
|
|
resp = await client.post(
|
|
"/api/v1/alarms/events/99999/acknowledge",
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
async def test_resolve_alarm(self, client, admin_user, admin_token, seed_alarm_event):
|
|
resp = await client.post(
|
|
f"/api/v1/alarms/events/{seed_alarm_event.id}/resolve",
|
|
params={"note": "已修复"},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
async def test_resolve_nonexistent_alarm(self, client, admin_user, admin_token):
|
|
resp = await client.post(
|
|
"/api/v1/alarms/events/99999/resolve",
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
|
|
class TestAlarmStats:
|
|
async def test_get_alarm_stats(self, client, admin_user, admin_token, seed_alarm_event):
|
|
resp = await client.get("/api/v1/alarms/stats", headers=auth_header(admin_token))
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert isinstance(body, dict)
|
|
|
|
async def test_get_alarm_stats_empty(self, client, admin_user, admin_token):
|
|
resp = await client.get("/api/v1/alarms/stats", headers=auth_header(admin_token))
|
|
assert resp.status_code == 200
|