a8ccf1d35c
- Alembic 005: таблицы irm_alerts и incident_alert_links - Модуль alerts: API/UI, Ack/Resolve, привязка к инциденту через alert_ids - Вебхук Grafana: одна транзакция ingress + irm_alerts; разбор payload в grafana_payload - По умолчанию инцидент из вебхука не создаётся (AUTO_INCIDENT_FROM_ALERT) - Документация IRM_GRAFANA_PARITY.md, обновления IRM.md и CHANGELOG Made-with: Cursor
105 lines
3.4 KiB
Python
105 lines
3.4 KiB
Python
"""IRM-модули: API без БД и обработчик инцидента по событию."""
|
|
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from onguard24.domain.entities import Alert, Severity
|
|
from onguard24.domain.events import AlertReceived
|
|
|
|
|
|
def test_incidents_api_list_no_db(client: TestClient) -> None:
|
|
r = client.get("/api/v1/modules/incidents/")
|
|
assert r.status_code == 200
|
|
assert r.json() == {"items": [], "database": "disabled"}
|
|
|
|
|
|
def test_tasks_api_list_no_db(client: TestClient) -> None:
|
|
r = client.get("/api/v1/modules/tasks/")
|
|
assert r.status_code == 200
|
|
assert r.json()["database"] == "disabled"
|
|
|
|
|
|
def test_escalations_api_list_no_db(client: TestClient) -> None:
|
|
r = client.get("/api/v1/modules/escalations/")
|
|
assert r.status_code == 200
|
|
assert r.json()["database"] == "disabled"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_incident_not_created_from_alert_by_default() -> None:
|
|
"""По умолчанию AUTO_INCIDENT_FROM_ALERT выкл — инцидент из вебхука не создаётся."""
|
|
calls: list = []
|
|
|
|
async def fake_execute(_query, *args):
|
|
calls.append(args)
|
|
return "INSERT 0 1"
|
|
|
|
mock_conn = AsyncMock()
|
|
mock_conn.execute = fake_execute
|
|
mock_cm = AsyncMock()
|
|
mock_cm.__aenter__ = AsyncMock(return_value=mock_conn)
|
|
mock_cm.__aexit__ = AsyncMock(return_value=None)
|
|
mock_pool = MagicMock()
|
|
mock_pool.acquire = MagicMock(return_value=mock_cm)
|
|
|
|
from onguard24.domain.events import InMemoryEventBus
|
|
from onguard24.modules import incidents as inc_mod
|
|
|
|
bus = InMemoryEventBus()
|
|
inc_mod.register_events(bus, mock_pool)
|
|
|
|
uid = uuid4()
|
|
ev = AlertReceived(
|
|
alert=Alert(source="grafana", title="CPU high", severity=Severity.WARNING),
|
|
raw_payload_ref=uid,
|
|
)
|
|
await bus.publish(ev)
|
|
assert calls == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_incident_inserted_on_alert_when_auto_enabled(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""При AUTO_INCIDENT_FROM_ALERT=1 подписка снова создаёт инцидент (legacy)."""
|
|
monkeypatch.setenv("AUTO_INCIDENT_FROM_ALERT", "1")
|
|
inserted: dict = {}
|
|
|
|
async def fake_execute(_query, *args):
|
|
inserted["args"] = args
|
|
return "INSERT 0 1"
|
|
|
|
mock_conn = AsyncMock()
|
|
mock_conn.execute = fake_execute
|
|
mock_cm = AsyncMock()
|
|
mock_cm.__aenter__ = AsyncMock(return_value=mock_conn)
|
|
mock_cm.__aexit__ = AsyncMock(return_value=None)
|
|
mock_pool = MagicMock()
|
|
mock_pool.acquire = MagicMock(return_value=mock_cm)
|
|
|
|
from onguard24.domain.events import InMemoryEventBus
|
|
from onguard24.modules import incidents as inc_mod
|
|
|
|
bus = InMemoryEventBus()
|
|
inc_mod.register_events(bus, mock_pool)
|
|
|
|
uid = uuid4()
|
|
ev = AlertReceived(
|
|
alert=Alert(source="grafana", title="CPU high", severity=Severity.WARNING),
|
|
raw_payload_ref=uid,
|
|
)
|
|
await bus.publish(ev)
|
|
|
|
assert inserted.get("args") is not None
|
|
assert inserted["args"][0] == "CPU high"
|
|
assert inserted["args"][1] == "warning"
|
|
assert inserted["args"][2] == uid
|
|
assert inserted["args"][3] is None
|
|
assert inserted["args"][4] is None
|
|
|
|
|
|
def test_incidents_post_requires_db(client: TestClient) -> None:
|
|
r = client.post("/api/v1/modules/incidents/", json={"title": "x"})
|
|
assert r.status_code == 503
|