Function test_trace_data

app/api/tests/public/test_public_api.py:86–126

Create test trace data in ClickHouse.

(async_clickhouse_client, test_project_with_api_key, test_trace_id)

@pytest.fixture
async def test_trace_data(async_clickhouse_client, test_project_with_api_key, test_trace_id):
    """Create test trace data in ClickHouse."""
    import random

    project = test_project_with_api_key
    project_id_str = str(project.id)

    # Insert test trace
    trace_data = {
        "Timestamp": datetime.now(timezone.utc),
        "TraceId": test_trace_id,
        "SpanId": str(random.randint(10**19, 10**20 - 1)),  # Unique span ID for the root span
        "ParentSpanId": "",
        "TraceState": "",
        "SpanName": "test_trace",
        "SpanKind": "SPAN_KIND_INTERNAL",
        "ServiceName": "test_service",
        "ResourceAttributes": {"agentops.project.id": project_id_str},
        "ScopeName": "",
        "ScopeVersion": "",
        "SpanAttributes": {"tags": "test,public-api"},
        "Duration": 1000000,
        "StatusCode": "STATUS_CODE_OK",
        "StatusMessage": "",
        "Events.Timestamp": [],
        "Events.Name": [],
        "Events.Attributes": [],
        "Links.TraceId": [],
        "Links.SpanId": [],
        "Links.TraceState": [],
        "Links.Attributes": [],
    }

    await async_clickhouse_client.insert(
        table="otel_traces", data=[list(trace_data.values())], column_names=list(trace_data.keys())
    )

    yield trace_data

    # Cleanup: delete the trace data after the test
    await async_clickhouse_client.command(f"DELETE FROM otel_traces WHERE TraceId = '{test_trace_id}'")


@pytest.fixture