Function install_telemetry

tests/unittests/telemetry/functional_test_helpers.py:280–322

Installs in-memory tracer + log exporter and patches ADK's globals. Spans and logs emitted by ADK during the test are written into the provided exporters. Both exporters MUST be passed in so each test makes the choice of exporter explicit (e.g. ``InMemoryLogRecordExporter`` vs ``WebUILogExp

(
    monkeypatch: pytest.MonkeyPatch,
    span_exporter: InMemorySpanExporter,
    log_exporter: InMemoryLogRecordExporter,
)

def install_telemetry(
    monkeypatch: pytest.MonkeyPatch,
    span_exporter: InMemorySpanExporter,
    log_exporter: InMemoryLogRecordExporter,
) -> None:
  """Installs in-memory tracer + log exporter and patches ADK's globals.

  Spans and logs emitted by ADK during the test are written into the
  provided exporters. Both exporters MUST be passed in so each test makes
  the choice of exporter explicit (e.g. ``InMemoryLogRecordExporter`` vs
  ``WebUILogExporter``).
  """
  tracer_provider = TracerProvider()
  tracer_provider.add_span_processor(SimpleSpanProcessor(span_exporter))
  real_tracer = tracer_provider.get_tracer(__name__)

  monkeypatch.setattr(
      tracing.tracer,
      "start_as_current_span",
      real_tracer.start_as_current_span,
  )
  monkeypatch.setattr(
      tracing.tracer,
      "start_span",
      real_tracer.start_span,
  )
  monkeypatch.setattr(
      node_tracing.tracer,
      "start_as_current_span",
      real_tracer.start_as_current_span,
  )
  monkeypatch.setattr(
      node_tracing.tracer,
      "start_span",
      real_tracer.start_span,
  )

  logger_provider = LoggerProvider()
  logger_provider.add_log_record_processor(
      SimpleLogRecordProcessor(log_exporter)
  )
  real_logger = logger_provider.get_logger(__name__)
  monkeypatch.setattr(tracing.otel_logger, "emit", real_logger.emit)


# ---------------------------------------------------------------------------