70 lines
2.2 KiB
Python
70 lines
2.2 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_inserted_on_alert_received() -> None:
|
||
|
|
"""При пуле БД подписка создаёт инцидент (INSERT)."""
|
||
|
|
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
|
||
|
|
|
||
|
|
|
||
|
|
def test_incidents_post_requires_db(client: TestClient) -> None:
|
||
|
|
r = client.post("/api/v1/modules/incidents/", json={"title": "x"})
|
||
|
|
assert r.status_code == 503
|