d30436573b
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Failing after 2m0s
CD Pipeline / build-and-deploy (push) Has been skipped
CD Pipeline / post-deploy-checks (push) Has been skipped
78 lines
2.1 KiB
Python
78 lines
2.1 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
from fastapi import BackgroundTasks, HTTPException
|
|
from prometheus_client import generate_latest
|
|
from starlette.requests import Request
|
|
|
|
from src.api.v1.webhooks import AlertmanagerPayload, alertmanager_webhook
|
|
from src.core.metrics import WEBHOOK_REQUESTS_TOTAL
|
|
|
|
|
|
def _request(
|
|
*,
|
|
client_host: str = "127.0.0.1",
|
|
headers: list[tuple[bytes, bytes]] | None = None,
|
|
) -> Request:
|
|
return Request(
|
|
{
|
|
"type": "http",
|
|
"method": "POST",
|
|
"path": "/api/v1/webhooks/alertmanager",
|
|
"scheme": "http",
|
|
"server": ("testserver", 80),
|
|
"client": (client_host, 50000),
|
|
"query_string": b"",
|
|
"headers": headers or [],
|
|
}
|
|
)
|
|
|
|
|
|
def _webhook_count(status: str) -> float:
|
|
return WEBHOOK_REQUESTS_TOTAL.labels(
|
|
source="alertmanager",
|
|
status=status,
|
|
)._value.get()
|
|
|
|
|
|
def _metrics_text() -> str:
|
|
return generate_latest().decode("utf-8")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_alertmanager_no_firing_payload_records_success_metric() -> None:
|
|
before = _webhook_count("success")
|
|
|
|
response = await alertmanager_webhook(
|
|
_request(),
|
|
AlertmanagerPayload(status="resolved", alerts=[]),
|
|
BackgroundTasks(),
|
|
)
|
|
|
|
assert response.success is True
|
|
assert response.message == "No firing alerts to process"
|
|
assert _webhook_count("success") == before + 1
|
|
assert (
|
|
'awoooi_webhook_requests_total{source="alertmanager",status="success"}'
|
|
in _metrics_text()
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_alertmanager_external_reject_records_error_metric() -> None:
|
|
before = _webhook_count("error")
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await alertmanager_webhook(
|
|
_request(client_host="8.8.8.8"),
|
|
AlertmanagerPayload(status="resolved", alerts=[]),
|
|
BackgroundTasks(),
|
|
)
|
|
|
|
assert exc_info.value.status_code == 403
|
|
assert _webhook_count("error") == before + 1
|
|
assert (
|
|
'awoooi_webhook_requests_total{source="alertmanager",status="error"}'
|
|
in _metrics_text()
|
|
)
|