126 lines
4.0 KiB
Python
126 lines
4.0 KiB
Python
"""Tests for Prometheus metrics collection."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
from starlette.requests import Request
|
|
from starlette.responses import PlainTextResponse
|
|
|
|
from app.middleware.metrics import MetricsMiddleware, _normalize_path
|
|
from app.utils.metrics import get_metrics
|
|
|
|
|
|
class TestMetricsUtils:
|
|
"""Test metrics utility functions."""
|
|
|
|
def test_normalize_path_with_uuid(self) -> None:
|
|
"""Test path normalization with UUID."""
|
|
path = "/api/resource/550e8400-e29b-41d4-a716-446655440000"
|
|
normalized = _normalize_path(path)
|
|
assert normalized == "/api/{id}"
|
|
|
|
def test_normalize_path_with_numeric_id(self) -> None:
|
|
"""Test path normalization with numeric ID."""
|
|
path = "/api/resource/123"
|
|
normalized = _normalize_path(path)
|
|
assert normalized == "/api/{id}"
|
|
|
|
def test_normalize_path_without_id(self) -> None:
|
|
"""Test path without ID remains unchanged."""
|
|
path = "/api/resource"
|
|
normalized = _normalize_path(path)
|
|
assert normalized == "/api/resource"
|
|
|
|
def test_get_metrics_returns_bytes(self) -> None:
|
|
"""Test that get_metrics returns bytes."""
|
|
metrics = get_metrics()
|
|
assert isinstance(metrics, bytes)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestMetricsMiddleware:
|
|
"""Test metrics collection middleware."""
|
|
|
|
async def test_middleware_tracks_request_metrics(self) -> None:
|
|
"""Test middleware tracks request metrics."""
|
|
middleware = MetricsMiddleware(app=MagicMock())
|
|
|
|
request = MagicMock(spec=Request)
|
|
request.method = "GET"
|
|
request.url.path = "/api/test"
|
|
|
|
response = PlainTextResponse("OK")
|
|
response.status_code = 200
|
|
|
|
call_next = AsyncMock(return_value=response)
|
|
|
|
result = await middleware.dispatch(request, call_next)
|
|
|
|
assert result == response
|
|
assert call_next.called
|
|
|
|
async def test_middleware_skips_metrics_endpoint(self) -> None:
|
|
"""Test middleware skips /metrics endpoint."""
|
|
middleware = MetricsMiddleware(app=MagicMock())
|
|
|
|
request = MagicMock(spec=Request)
|
|
request.method = "GET"
|
|
request.url.path = "/metrics"
|
|
|
|
response = PlainTextResponse("metrics")
|
|
response.status_code = 200
|
|
|
|
call_next = AsyncMock(return_value=response)
|
|
|
|
result = await middleware.dispatch(request, call_next)
|
|
|
|
assert result == response
|
|
|
|
async def test_middleware_tracks_error_responses(self) -> None:
|
|
"""Test middleware tracks error response status codes."""
|
|
middleware = MetricsMiddleware(app=MagicMock())
|
|
|
|
request = MagicMock(spec=Request)
|
|
request.method = "GET"
|
|
request.url.path = "/api/test"
|
|
|
|
response = PlainTextResponse("Not Found")
|
|
response.status_code = 404
|
|
|
|
call_next = AsyncMock(return_value=response)
|
|
|
|
result = await middleware.dispatch(request, call_next)
|
|
|
|
assert result == response
|
|
assert result.status_code == 404
|
|
|
|
async def test_middleware_handles_exceptions(self) -> None:
|
|
"""Test middleware handles exceptions during request processing."""
|
|
middleware = MetricsMiddleware(app=MagicMock())
|
|
|
|
request = MagicMock(spec=Request)
|
|
request.method = "GET"
|
|
request.url.path = "/api/test"
|
|
|
|
call_next = AsyncMock(side_effect=RuntimeError("Test error"))
|
|
|
|
with pytest.raises(RuntimeError):
|
|
await middleware.dispatch(request, call_next)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestMetricsEndpoint:
|
|
"""Test the /metrics endpoint."""
|
|
|
|
async def test_metrics_endpoint_returns_prometheus_format(self) -> None:
|
|
"""Test metrics endpoint returns Prometheus format."""
|
|
from app.routers.metrics import get_application_metrics
|
|
|
|
response = await get_application_metrics()
|
|
|
|
assert response.status_code == 200
|
|
assert response.media_type.startswith("text/plain")
|
|
assert b"bangui_http_requests_total" in response.body
|