Max Isbey
918e20aba0
The previous tests/server/conftest.py called trace.set_tracer_provider() directly, which is set-once per process and raced against logfire's capfire fixture (tests/shared/test_otel.py) under xdist — whichever ran first in a worker won, the other's tests broke. Converge on capfire as the single span-capture owner since logfire.configure() already handles repeat calls by swapping span processors instead of re-setting the provider: - tests/conftest.py: set LOGFIRE_DISTRIBUTED_TRACING=true so propagation tests don't trip logfire's 'found propagated trace context' RuntimeWarning. - tests/server/conftest.py: SpanCapture adapter over capfire.exporter — filters to the mcp-python-sdk instrumentation scope and excludes logfire's pending_span markers, so tests assert on raw ReadableSpan without importing logfire types. - tests/shared/test_otel.py: drop the now-unneeded filterwarnings decorator.
46 lines
1.5 KiB
Python
46 lines
1.5 KiB
Python
"""Shared fixtures for server-side tests."""
|
|
|
|
from collections.abc import Iterator
|
|
|
|
import pytest
|
|
from logfire.testing import CaptureLogfire, TestExporter
|
|
from opentelemetry.sdk.trace import ReadableSpan
|
|
|
|
|
|
class SpanCapture:
|
|
"""Thin adapter over logfire's `TestExporter` for asserting on MCP spans.
|
|
|
|
`finished()` returns the raw `ReadableSpan` objects emitted by the
|
|
``mcp-python-sdk`` instrumentation scope, filtered to exclude logfire's
|
|
synthetic ``pending_span`` markers, so tests can assert directly on
|
|
`.name`, `.kind`, `.status`, `.attributes`, `.parent`, `.events`.
|
|
"""
|
|
|
|
def __init__(self, exporter: TestExporter) -> None:
|
|
self._exporter = exporter
|
|
|
|
def clear(self) -> None:
|
|
self._exporter.clear()
|
|
|
|
def finished(self) -> list[ReadableSpan]:
|
|
return [
|
|
s
|
|
for s in self._exporter.exported_spans
|
|
if s.instrumentation_scope is not None
|
|
and s.instrumentation_scope.name == "mcp-python-sdk"
|
|
and not (s.attributes and s.attributes.get("logfire.span_type") == "pending_span")
|
|
]
|
|
|
|
|
|
@pytest.fixture
|
|
def spans(capfire: CaptureLogfire) -> Iterator[SpanCapture]:
|
|
"""In-memory MCP span capture, cleared before and after each test.
|
|
|
|
Backed by the project-level `capfire` override (see ``tests/conftest.py``)
|
|
so there is a single global tracer provider for the suite.
|
|
"""
|
|
capture = SpanCapture(capfire.exporter)
|
|
capture.clear()
|
|
yield capture
|
|
capture.clear()
|