de29270a25
Co-authored-by: Cursor <cursoragent@cursor.com>
54 lines
2.1 KiB
Python
54 lines
2.1 KiB
Python
"""Tests for per-event-type severity overrides."""
|
|
|
|
import uuid
|
|
|
|
from app.models.event_severity_override import EventSeverityOverride
|
|
|
|
VALID_EVENT = {
|
|
"schema_version": "1.0",
|
|
"event_id": "550e8400-e29b-41d4-a716-446655440000",
|
|
"occurred_at": "2026-05-27T10:00:00+03:00",
|
|
"source": {"product": "rdp-login-monitor", "product_version": "1.2.3-SAC"},
|
|
"host": {"hostname": "test-host", "os_family": "windows"},
|
|
"category": "auth",
|
|
"type": "rdp.login.success",
|
|
"severity": "info",
|
|
"title": "RDP login",
|
|
"summary": "pytest",
|
|
}
|
|
|
|
|
|
def test_severity_override_applied_on_ingest(client, auth_headers, jwt_headers, db_session):
|
|
db_session.add(EventSeverityOverride(event_type="rdp.login.success", severity="warning"))
|
|
db_session.commit()
|
|
|
|
event_id = str(uuid.uuid4())
|
|
payload = {**VALID_EVENT, "event_id": event_id}
|
|
r = client.post("/api/v1/events", json=payload, headers=auth_headers)
|
|
assert r.status_code == 201
|
|
|
|
listed = client.get("/api/v1/events", headers=jwt_headers, params={"type": "rdp.login.success"})
|
|
row = next(i for i in listed.json()["items"] if i["event_id"] == event_id)
|
|
assert row["severity"] == "warning"
|
|
|
|
|
|
def test_severity_overrides_settings_api(client, jwt_headers, db_session):
|
|
db_session.add(EventSeverityOverride(event_type="ssh.login.failed", severity="high"))
|
|
db_session.commit()
|
|
|
|
r = client.get("/api/v1/settings/notifications/severity-overrides", headers=jwt_headers)
|
|
assert r.status_code == 200
|
|
items = {i["event_type"]: i for i in r.json()["items"]}
|
|
assert items["ssh.login.failed"]["override_severity"] == "high"
|
|
assert items["rdp.login.success"]["default_severity"] == "info"
|
|
|
|
put = client.put(
|
|
"/api/v1/settings/notifications/severity-overrides",
|
|
headers=jwt_headers,
|
|
json={"overrides": {"rdp.login.success": "warning", "ssh.login.failed": None}},
|
|
)
|
|
assert put.status_code == 200
|
|
updated = {i["event_type"]: i for i in put.json()["items"]}
|
|
assert updated["rdp.login.success"]["override_severity"] == "warning"
|
|
assert updated["ssh.login.failed"]["override_severity"] is None
|