fix(tests): improve structure of telemetry tests for consistency

This commit is contained in:
Emilio Garcia 2025-10-29 11:56:11 -04:00
parent 583df48479
commit 79156bb08c
3 changed files with 55 additions and 24 deletions

View file

@ -13,6 +13,8 @@ from typing import Any
@dataclass
class SpanStub:
"""Unified span interface for both in-memory and OTLP collectors."""
name: str
attributes: Mapping[str, Any] | None = None
resource_attributes: dict[str, Any] | None = None
@ -20,6 +22,13 @@ class SpanStub:
trace_id: str | None = None
span_id: str | None = None
@property
def context(self):
"""Provide context-like interface for trace_id compatibility."""
if self.trace_id is None:
return None
return type("Context", (), {"trace_id": int(self.trace_id, 16)})()
def _value_to_python(value: Any) -> Any:
kind = value.WhichOneof("value")
@ -56,12 +65,18 @@ def events_to_list(events: Iterable[Any]) -> list[dict[str, Any]]:
class BaseTelemetryCollector:
"""Base class for telemetry collectors that ensures consistent return types.
All collectors must return SpanStub objects to ensure test compatibility
across both library-client and server modes.
"""
def get_spans(
self,
expected_count: int | None = None,
timeout: float = 5.0,
poll_interval: float = 0.05,
) -> tuple[Any, ...]:
) -> tuple[SpanStub, ...]:
import time
deadline = time.time() + timeout
@ -97,7 +112,7 @@ class BaseTelemetryCollector:
def clear(self) -> None:
self._clear_impl()
def _snapshot_spans(self) -> tuple[Any, ...]: # pragma: no cover - interface hook
def _snapshot_spans(self) -> tuple[SpanStub, ...]: # pragma: no cover - interface hook
raise NotImplementedError
def _snapshot_metrics(self) -> Any | None: # pragma: no cover - interface hook