agent-platform/tests/test_observability.py

import asyncio

import httpx
from core_shared.observability import add_observability
from fastapi import FastAPI


def test_observability_records_http_metrics() -> None:
    asyncio.run(_run_observability_smoke())


async def _run_observability_smoke() -> None:
    app = FastAPI()
    add_observability(app, "test-service")

    @app.get("/health")
    async def health() -> dict[str, str]:
        return {"status": "ok"}

    transport = httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
        health_response = await client.get("/health")
        metrics_response = await client.get("/metrics")

    assert health_response.status_code == 200
    assert metrics_response.status_code == 200
    assert health_response.headers["x-trace-id"]
    assert health_response.headers["x-span-id"]
    assert 'agent_platform_service_info{service="test-service"} 1' in metrics_response.text
    assert "agent_platform_http_requests_total" in metrics_response.text
    assert 'method="GET"' in metrics_response.text
    assert 'path="/health"' in metrics_response.text
    assert 'status_code="200"' in metrics_response.text


def test_observability_preserves_incoming_trace_id() -> None:
    asyncio.run(_run_trace_propagation_smoke())


async def _run_trace_propagation_smoke() -> None:
    app = FastAPI()
    add_observability(app, "test-service")

    @app.get("/health")
    async def health() -> dict[str, str]:
        return {"status": "ok"}

    transport = httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
        response = await client.get("/health", headers={"x-trace-id": "trace-123"})

    assert response.status_code == 200
    assert response.headers["x-trace-id"] == "trace-123"
    assert response.headers["x-span-id"]