52 lines
1.9 KiB
Python
52 lines
1.9 KiB
Python
import json
|
|
import logging
|
|
import uuid
|
|
|
|
from app.core.observability import JsonFormatter
|
|
|
|
|
|
def test_request_correlation_id_is_echoed_and_used_in_errors(client):
|
|
correlation_id = str(uuid.uuid4())
|
|
response = client.get(
|
|
"/api/v1/dashboard",
|
|
headers={"X-Correlation-Id": correlation_id},
|
|
)
|
|
assert response.status_code == 401
|
|
assert response.headers["X-Correlation-Id"] == correlation_id
|
|
assert response.json()["error"]["correlation_id"] == correlation_id
|
|
|
|
|
|
def test_invalid_request_correlation_id_is_replaced(client):
|
|
response = client.get("/health/live", headers={"X-Correlation-Id": "not-a-uuid"})
|
|
assert response.status_code == 200
|
|
assert uuid.UUID(response.headers["X-Correlation-Id"])
|
|
|
|
|
|
def test_metrics_expose_http_database_and_outbox_state(client):
|
|
client.get("/health/ready")
|
|
response = client.get("/metrics")
|
|
assert response.status_code == 200
|
|
assert "mobilityops_http_requests_total" in response.text
|
|
assert "mobilityops_database_ready 1.0" in response.text
|
|
assert 'mobilityops_outbox_events{scenario="synthetic",status="failed"}' in response.text
|
|
|
|
|
|
def test_metrics_token_is_enforced_when_configured(client, monkeypatch):
|
|
import app.api.routers.observability as router
|
|
|
|
monkeypatch.setattr(router.settings, "metrics_bearer_token", "metrics-secret")
|
|
assert client.get("/metrics").status_code == 401
|
|
assert (
|
|
client.get("/metrics", headers={"Authorization": "Bearer metrics-secret"}).status_code
|
|
== 200
|
|
)
|
|
|
|
|
|
def test_json_formatter_emits_machine_readable_fields():
|
|
record = logging.LogRecord("mobilityops.test", logging.INFO, __file__, 1, "ready", (), None)
|
|
record.status_code = 200
|
|
payload = json.loads(JsonFormatter().format(record))
|
|
assert payload["message"] == "ready"
|
|
assert payload["status_code"] == 200
|
|
assert payload["timestamp"].endswith("+00:00")
|