Function test_span_data

app/api/tests/public/test_public_api.py:130–168

Create test span data in ClickHouse.

(async_clickhouse_client, test_trace_data, test_span_id, test_trace_id)

@pytest.fixture
async def test_span_data(async_clickhouse_client, test_trace_data, test_span_id, test_trace_id):
    """Create test span data in ClickHouse."""
    # Use the same project ID as the trace data
    project_id_str = test_trace_data["ResourceAttributes"]["agentops.project.id"]

    # Insert test span
    span_data = {
        "Timestamp": datetime.now(timezone.utc),
        "TraceId": test_trace_id,
        "SpanId": test_span_id,
        "ParentSpanId": "",
        "TraceState": "",
        "SpanName": "test_span",
        "SpanKind": "SPAN_KIND_INTERNAL",
        "ServiceName": "test_service",
        "ResourceAttributes": {"agentops.project.id": project_id_str, "service.name": "test_service"},
        "ScopeName": "",
        "ScopeVersion": "",
        "SpanAttributes": {"operation": "test", "test": "value"},
        "Duration": 1000000,  # 1ms in nanoseconds
        "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(span_data.values())], column_names=list(span_data.keys())
    )

    yield span_data

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


@pytest.fixture