"""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