onGuard24/tests/test_ingress.py

import json
from unittest.mock import AsyncMock, MagicMock, patch

from fastapi.testclient import TestClient


def test_grafana_webhook_no_db(client: TestClient) -> None:
    """Без пула БД — 202, запись не падает."""
    r = client.post(
        "/api/v1/ingress/grafana",
        content=json.dumps({"title": "t"}),
        headers={"Content-Type": "application/json"},
    )
    assert r.status_code == 202


def test_grafana_webhook_unauthorized_when_secret_set(client: TestClient) -> None:
    app = client.app
    real = app.state.settings.grafana_webhook_secret
    app.state.settings.grafana_webhook_secret = "s3cr3t"
    try:
        r = client.post(
            "/api/v1/ingress/grafana",
            content=b"{}",
            headers={"Content-Type": "application/json"},
        )
        assert r.status_code == 401
        r2 = client.post(
            "/api/v1/ingress/grafana",
            content=b"{}",
            headers={"Content-Type": "application/json", "X-OnGuard-Secret": "s3cr3t"},
        )
        assert r2.status_code == 202
    finally:
        app.state.settings.grafana_webhook_secret = real


def test_grafana_webhook_inserts_with_mock_pool(client: TestClient) -> None:
    from uuid import uuid4

    mock_conn = AsyncMock()
    uid = uuid4()
    mock_conn.fetchrow = AsyncMock(return_value={"id": uid})
    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)

    app = client.app
    real_pool = app.state.pool
    app.state.pool = mock_pool
    try:
        r = client.post(
            "/api/v1/ingress/grafana",
            content=json.dumps({"a": 1}),
            headers={"Content-Type": "application/json"},
        )
        assert r.status_code == 202
        mock_conn.fetchrow.assert_called_once()
    finally:
        app.state.pool = real_pool


def test_grafana_webhook_publishes_alert_received(client: TestClient) -> None:
    from unittest.mock import patch
    from uuid import uuid4

    mock_conn = AsyncMock()
    uid = uuid4()
    mock_conn.fetchrow = AsyncMock(return_value={"id": uid})
    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)

    app = client.app
    bus = app.state.event_bus
    with patch.object(bus, "publish_alert_received", new_callable=AsyncMock) as spy:
        real_pool = app.state.pool
        app.state.pool = mock_pool
        try:
            r = client.post(
                "/api/v1/ingress/grafana",
                content=json.dumps({"title": "x"}),
                headers={"Content-Type": "application/json"},
            )
            assert r.status_code == 202
            spy.assert_awaited_once()
            assert spy.await_args.kwargs.get("raw_payload_ref") == uid
        finally:
            app.state.pool = real_pool