wehub-resource-sync
1018 lines
39 KiB
Python
1018 lines
39 KiB
Python
"""Unit tests for ``OpenInferenceSpanInterceptor`` driven by Google-ADK-shaped spans.
|
||
|
||
Mirrors ``tests/test_integrations/test_agentcore/test_span_interceptor.py``
|
||
(itself a port of the Pydantic AI suite). The interceptor under test is
|
||
shared across every OpenInference-backed integration — Google ADK is the
|
||
first user of it on the deepeval side, so this file is the canonical
|
||
synthetic-span coverage.
|
||
|
||
What this file verifies on the OpenInference span interceptor:
|
||
|
||
- Trace-level reads from ``current_trace_context`` with
|
||
``OpenInferenceInstrumentationSettings`` defaults as fallback, FRESH
|
||
resolution at on_end (so ``update_current_trace(...)`` from inside
|
||
a tool body still lands), and metadata merge with context winning.
|
||
- Span placeholder push/pop on ``current_span_context`` so
|
||
``update_current_span(...)`` from anywhere in the call stack
|
||
serializes back to ``confident.span.*`` at on_end.
|
||
- Implicit ``Trace`` placeholder (``_is_otel_implicit=True``) push for bare ADK callers
|
||
(no enclosing ``@observe`` / ``with trace(...)``) so
|
||
``update_current_trace(...)`` works without a user-pushed context.
|
||
- Parent bridge: ``confident.span.parent_uuid`` stamped on OTel roots
|
||
enclosed in a real (non-implicit) deepeval span.
|
||
- ``next_*_span(...)`` consumption at on_start; component-level
|
||
``BaseMetric`` instances stashed via ``stash_pending_metrics``
|
||
(gated on ``trace_manager.is_evaluating``).
|
||
- Removed top-level kwargs (the OTel POC migration) raise
|
||
``TypeError`` on both ``OpenInferenceInstrumentationSettings`` and
|
||
``instrument_google_adk``.
|
||
- OpenInference framework-attr extraction:
|
||
``openinference.span.kind`` → ``confident.span.type``,
|
||
``llm.input_messages.{idx}.message.content`` → ``confident.span.input``,
|
||
``llm.output_messages.{idx}...`` → ``confident.span.output``,
|
||
nested ``...tool_calls.{tc}.tool_call.function.{name,arguments}`` →
|
||
``confident.span.tools_called``, ``llm.token_count.{prompt,completion}``
|
||
→ ``confident.llm.{input,output}_token_count``,
|
||
``llm.model_name`` → ``confident.llm.model``,
|
||
tool spans' ``tool.name`` / ``tool.parameters`` →
|
||
``confident.span.tools_called`` (1-element list) +
|
||
``confident.span.input``.
|
||
|
||
These tests do NOT require ``google-adk`` /
|
||
``openinference-instrumentation-google-adk`` — they drive the
|
||
interceptor with synthetic OTel spans built from ``MagicMock``.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from itertools import count
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
import pytest
|
||
|
||
from deepeval.integrations.openinference.instrumentator import (
|
||
OpenInferenceInstrumentationSettings,
|
||
OpenInferenceSpanInterceptor,
|
||
)
|
||
from deepeval.tracing.context import (
|
||
current_span_context,
|
||
current_trace_context,
|
||
next_agent_span,
|
||
next_llm_span,
|
||
next_tool_span,
|
||
update_current_span,
|
||
update_current_trace,
|
||
)
|
||
from deepeval.tracing.trace_context import trace
|
||
|
||
|
||
_span_id_counter = count(start=1)
|
||
_trace_id_counter = count(start=1)
|
||
|
||
|
||
def _make_mock_span(
|
||
*,
|
||
span_kind: str | None = None,
|
||
agent_name: str | None = None,
|
||
tool_name: str | None = None,
|
||
span_name: str = "",
|
||
parent: object | None = None,
|
||
extra_attrs: dict | None = None,
|
||
):
|
||
"""Mock OTel span shaped to match ``OpenInferenceSpanInterceptor``'s
|
||
expectations.
|
||
|
||
Mirrors the OTel SDK invariant that ``Span.attributes`` is a view
|
||
over the same underlying ``_attributes`` mapping — so writes via
|
||
either ``set_attribute(...)`` or direct ``_attributes[k] = v``
|
||
(used by ``_set_attr_post_end`` to bypass the ended-span guard) are
|
||
observable via ``span.attributes.get(...)``.
|
||
|
||
OpenInference / Google-ADK-specific differences from the
|
||
AgentCore mock:
|
||
|
||
- Classification reads ``openinference.span.kind`` (uppercased)
|
||
instead of ``gen_ai.operation.name``. Recognized values:
|
||
``"AGENT"`` / ``"CHAIN"`` → agent, ``"LLM"`` → llm,
|
||
``"TOOL"`` → tool, ``"RETRIEVER"`` → retriever; anything else
|
||
→ ``"custom"``; missing → ``None`` (interceptor leaves it alone).
|
||
- Agent / tool name come from ``agent.name`` / ``tool.name``
|
||
(no ``gen_ai.`` prefix).
|
||
- ``span.name`` is a plain string (used as the fallback for
|
||
``_get_agent_name`` / ``_get_tool_name``). Default empty so
|
||
the fallback doesn't fire spuriously.
|
||
- ``span.events`` defaults to ``[]`` for parity with the
|
||
AgentCore mock; the OpenInference interceptor doesn't read
|
||
events directly but downstream attr extraction is event-free.
|
||
"""
|
||
span = MagicMock()
|
||
backing: dict = {}
|
||
span._attributes = backing
|
||
span.attributes = backing
|
||
span.name = span_name
|
||
span.events = []
|
||
span.start_time = None # forces _push_span_context to use perf_counter()
|
||
span.parent = parent # None → root span
|
||
if span_kind:
|
||
backing["openinference.span.kind"] = span_kind
|
||
if agent_name:
|
||
backing["agent.name"] = agent_name
|
||
if tool_name:
|
||
backing["tool.name"] = tool_name
|
||
if extra_attrs:
|
||
backing.update(extra_attrs)
|
||
span.set_attribute.side_effect = lambda k, v: backing.__setitem__(k, v)
|
||
span.get_span_context.return_value = MagicMock(
|
||
trace_id=next(_trace_id_counter),
|
||
span_id=next(_span_id_counter),
|
||
)
|
||
return span
|
||
|
||
|
||
def _make_settings(**kwargs):
|
||
"""Return a minimal mock ``OpenInferenceInstrumentationSettings``.
|
||
|
||
``spec=[]`` disallows auto-attrs so a typo on the interceptor side
|
||
surfaces as ``AttributeError`` rather than a silent ``MagicMock``.
|
||
|
||
Settings carries only trace-level fields (no per-span
|
||
metric_collection / prompt / metrics) — span-level config is a
|
||
runtime concern (``update_current_span(...)`` from inside a tool
|
||
body, or ``with next_*_span(...)`` at the call site).
|
||
"""
|
||
settings = MagicMock(spec=[])
|
||
settings.thread_id = kwargs.get("thread_id")
|
||
settings.name = kwargs.get("name")
|
||
settings.metadata = kwargs.get("metadata")
|
||
settings.user_id = kwargs.get("user_id")
|
||
settings.tags = kwargs.get("tags")
|
||
settings.metric_collection = kwargs.get("metric_collection")
|
||
settings.test_case_id = kwargs.get("test_case_id")
|
||
settings.turn_id = kwargs.get("turn_id")
|
||
settings.environment = kwargs.get("environment")
|
||
return settings
|
||
|
||
|
||
def _make_agent_span_mock(agent_name: str = "agent_x"):
|
||
"""Mock an OpenInference-shaped root agent span (kind=AGENT)."""
|
||
return _make_mock_span(span_kind="AGENT", agent_name=agent_name)
|
||
|
||
|
||
def _make_tool_span_mock(tool_name: str = "calculate"):
|
||
"""Mock an OpenInference-shaped tool span (kind=TOOL)."""
|
||
return _make_mock_span(span_kind="TOOL", tool_name=tool_name)
|
||
|
||
|
||
def _make_llm_span_mock():
|
||
"""Mock an OpenInference-shaped LLM span (kind=LLM)."""
|
||
return _make_mock_span(span_kind="LLM")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Trace-context reads — settings fallback + runtime override.
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestTraceContextReads:
|
||
def test_uses_settings_when_no_trace_context(self):
|
||
"""Falls back to settings when current_trace_context is None."""
|
||
token = current_trace_context.set(None)
|
||
try:
|
||
settings = _make_settings(
|
||
thread_id="settings-thread",
|
||
name="settings-name",
|
||
metadata={"source": "settings"},
|
||
)
|
||
interceptor = OpenInferenceSpanInterceptor(settings)
|
||
span = _make_mock_span()
|
||
|
||
interceptor.on_start(span, None)
|
||
interceptor.on_end(span)
|
||
|
||
assert (
|
||
span.attributes.get("confident.trace.thread_id")
|
||
== "settings-thread"
|
||
)
|
||
assert (
|
||
span.attributes.get("confident.trace.name") == "settings-name"
|
||
)
|
||
assert json.loads(span.attributes["confident.trace.metadata"]) == {
|
||
"source": "settings"
|
||
}
|
||
finally:
|
||
current_trace_context.reset(token)
|
||
|
||
def test_prefers_trace_context_over_settings_for_scalars(self):
|
||
settings = _make_settings(
|
||
thread_id="settings-thread",
|
||
name="settings-name",
|
||
)
|
||
interceptor = OpenInferenceSpanInterceptor(settings)
|
||
span = _make_mock_span()
|
||
|
||
with trace(thread_id="ctx-thread", name="ctx-name"):
|
||
interceptor.on_start(span, None)
|
||
interceptor.on_end(span)
|
||
|
||
assert span.attributes.get("confident.trace.thread_id") == "ctx-thread"
|
||
assert span.attributes.get("confident.trace.name") == "ctx-name"
|
||
|
||
def test_metadata_is_merged_with_context_winning(self):
|
||
settings = _make_settings(
|
||
metadata={"base_key": "base_val", "shared_key": "from_settings"},
|
||
)
|
||
interceptor = OpenInferenceSpanInterceptor(settings)
|
||
span = _make_mock_span()
|
||
|
||
with trace(metadata={"ctx_key": "ctx_val", "shared_key": "from_ctx"}):
|
||
interceptor.on_start(span, None)
|
||
interceptor.on_end(span)
|
||
|
||
result = json.loads(span.attributes["confident.trace.metadata"])
|
||
assert result["base_key"] == "base_val"
|
||
assert result["ctx_key"] == "ctx_val"
|
||
assert result["shared_key"] == "from_ctx"
|
||
|
||
def test_update_current_trace_after_on_start_lands_on_otel_attrs(self):
|
||
"""Trace attrs are snapshotted FRESH at on_end, not on_start.
|
||
|
||
Regression guard for the at-on_start asymmetry: if a downstream
|
||
caller mutates the active trace via ``update_current_trace``
|
||
AFTER the OTel span's ``on_start`` has fired (e.g. from inside
|
||
an ADK tool body), the new values must still land on
|
||
``confident.trace.*`` when ``on_end`` runs.
|
||
"""
|
||
settings = _make_settings(name="settings-name")
|
||
interceptor = OpenInferenceSpanInterceptor(settings)
|
||
span = _make_mock_span()
|
||
|
||
with trace(name="initial-name"):
|
||
interceptor.on_start(span, None)
|
||
|
||
update_current_trace(
|
||
name="updated-name",
|
||
user_id="updated-user",
|
||
metadata={"phase": "post-start"},
|
||
)
|
||
|
||
interceptor.on_end(span)
|
||
|
||
assert span.attributes.get("confident.trace.name") == "updated-name"
|
||
assert span.attributes.get("confident.trace.user_id") == "updated-user"
|
||
assert json.loads(span.attributes["confident.trace.metadata"]) == {
|
||
"phase": "post-start"
|
||
}
|
||
|
||
def test_trace_metric_collection_resolution_order(self):
|
||
settings = _make_settings(metric_collection="settings-mc")
|
||
interceptor = OpenInferenceSpanInterceptor(settings)
|
||
span = _make_mock_span()
|
||
|
||
with trace(metric_collection="ctx-mc"):
|
||
interceptor.on_start(span, None)
|
||
interceptor.on_end(span)
|
||
|
||
assert (
|
||
span.attributes.get("confident.trace.metric_collection") == "ctx-mc"
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Span placeholder push / pop on current_span_context.
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestSpanContextPushPop:
|
||
def test_current_span_context_set_during_span_lifetime(self):
|
||
settings = _make_settings()
|
||
interceptor = OpenInferenceSpanInterceptor(settings)
|
||
span = _make_mock_span()
|
||
|
||
before = current_span_context.get()
|
||
interceptor.on_start(span, None)
|
||
during = current_span_context.get()
|
||
|
||
assert during is not None
|
||
assert during is not before
|
||
|
||
interceptor.on_end(span)
|
||
after = current_span_context.get()
|
||
assert after is before
|
||
|
||
def test_update_current_span_metadata_lands_in_otel_attrs(self):
|
||
settings = _make_settings()
|
||
interceptor = OpenInferenceSpanInterceptor(settings)
|
||
span = _make_mock_span()
|
||
|
||
interceptor.on_start(span, None)
|
||
update_current_span(
|
||
metadata={"weather_source": "mock", "city": "Paris"},
|
||
input={"query": "Weather?"},
|
||
output="Sunny",
|
||
)
|
||
interceptor.on_end(span)
|
||
|
||
assert span.attributes.get("confident.span.metadata") is not None
|
||
assert json.loads(span.attributes["confident.span.metadata"]) == {
|
||
"weather_source": "mock",
|
||
"city": "Paris",
|
||
}
|
||
assert json.loads(span.attributes["confident.span.input"]) == {
|
||
"query": "Weather?"
|
||
}
|
||
assert json.loads(span.attributes["confident.span.output"]) == "Sunny"
|
||
|
||
def test_update_current_span_metric_collection_lands_in_otel_attrs(self):
|
||
"""``update_current_span(metric_collection=...)`` from inside an
|
||
ADK tool body lands on the tool span's OTel attrs. Direct analog
|
||
of the ``special_tool`` flow in ``apps/googleadk_eval_app.py``."""
|
||
settings = _make_settings()
|
||
interceptor = OpenInferenceSpanInterceptor(settings)
|
||
span = _make_tool_span_mock("special_tool")
|
||
|
||
interceptor.on_start(span, None)
|
||
update_current_span(metric_collection="runtime-collection")
|
||
interceptor.on_end(span)
|
||
|
||
assert (
|
||
span.attributes.get("confident.span.metric_collection")
|
||
== "runtime-collection"
|
||
)
|
||
|
||
def test_nested_spans_lifo_pop_restores_parent_placeholder(self):
|
||
settings = _make_settings()
|
||
interceptor = OpenInferenceSpanInterceptor(settings)
|
||
outer = _make_mock_span()
|
||
inner = _make_mock_span(parent=MagicMock())
|
||
|
||
interceptor.on_start(outer, None)
|
||
outer_placeholder = current_span_context.get()
|
||
|
||
interceptor.on_start(inner, None)
|
||
inner_placeholder = current_span_context.get()
|
||
assert inner_placeholder is not outer_placeholder
|
||
|
||
interceptor.on_end(inner)
|
||
assert current_span_context.get() is outer_placeholder
|
||
|
||
interceptor.on_end(outer)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Implicit trace placeholder push for bare ADK callers.
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestImplicitTraceContext:
|
||
"""Symmetric to ``TestSpanContextPushPop`` but at the trace level.
|
||
The interceptor pushes an implicit ``Trace`` placeholder onto
|
||
``current_trace_context`` for the OTel root span's lifetime so
|
||
``update_current_trace(...)`` from inside ADK tools / nested
|
||
helpers can mutate something. The placeholder is tagged
|
||
``_is_otel_implicit=True`` so ``ContextAwareSpanProcessor`` keeps
|
||
routing to OTLP for those callers.
|
||
"""
|
||
|
||
def test_root_span_pushes_implicit_trace_when_no_user_context(self):
|
||
token = current_trace_context.set(None)
|
||
try:
|
||
settings = _make_settings()
|
||
interceptor = OpenInferenceSpanInterceptor(settings)
|
||
root = _make_mock_span()
|
||
|
||
interceptor.on_start(root, None)
|
||
during = current_trace_context.get()
|
||
|
||
assert during is not None
|
||
assert during._is_otel_implicit is True
|
||
|
||
interceptor.on_end(root)
|
||
assert current_trace_context.get() is None
|
||
finally:
|
||
current_trace_context.reset(token)
|
||
|
||
def test_does_not_overwrite_user_pushed_trace_context(self):
|
||
settings = _make_settings()
|
||
interceptor = OpenInferenceSpanInterceptor(settings)
|
||
root = _make_mock_span()
|
||
|
||
with trace() as user_trace:
|
||
assert user_trace._is_otel_implicit is False
|
||
|
||
interceptor.on_start(root, None)
|
||
during = current_trace_context.get()
|
||
|
||
assert during is user_trace
|
||
assert during._is_otel_implicit is False
|
||
|
||
interceptor.on_end(root)
|
||
|
||
assert current_trace_context.get() is user_trace
|
||
|
||
def test_child_span_does_not_push_its_own_placeholder(self):
|
||
token = current_trace_context.set(None)
|
||
try:
|
||
settings = _make_settings()
|
||
interceptor = OpenInferenceSpanInterceptor(settings)
|
||
root = _make_mock_span()
|
||
child = _make_mock_span(parent=MagicMock())
|
||
|
||
interceptor.on_start(root, None)
|
||
implicit = current_trace_context.get()
|
||
assert implicit is not None
|
||
|
||
interceptor.on_start(child, None)
|
||
assert current_trace_context.get() is implicit
|
||
|
||
interceptor.on_end(child)
|
||
assert current_trace_context.get() is implicit
|
||
|
||
interceptor.on_end(root)
|
||
assert current_trace_context.get() is None
|
||
finally:
|
||
current_trace_context.reset(token)
|
||
|
||
def test_update_current_trace_in_implicit_context_lands_on_otel_attrs(
|
||
self,
|
||
):
|
||
token = current_trace_context.set(None)
|
||
try:
|
||
settings = _make_settings()
|
||
interceptor = OpenInferenceSpanInterceptor(settings)
|
||
root = _make_mock_span()
|
||
|
||
interceptor.on_start(root, None)
|
||
update_current_trace(
|
||
name="bare-trace",
|
||
user_id="user-bare",
|
||
tags=["bare"],
|
||
metadata={"source": "tool", "request_id": "req-bare-1"},
|
||
)
|
||
interceptor.on_end(root)
|
||
|
||
assert root.attributes.get("confident.trace.name") == "bare-trace"
|
||
assert root.attributes.get("confident.trace.user_id") == "user-bare"
|
||
assert root.attributes.get("confident.trace.tags") == ["bare"]
|
||
assert json.loads(root.attributes["confident.trace.metadata"]) == {
|
||
"source": "tool",
|
||
"request_id": "req-bare-1",
|
||
}
|
||
finally:
|
||
current_trace_context.reset(token)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Parent bridge: confident.span.parent_uuid stamping for OTel roots
|
||
# inside an enclosing deepeval (real, non-implicit) span.
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestParentBridge:
|
||
def test_stamps_parent_uuid_when_enclosed_in_deepeval_span(self):
|
||
"""When a real deepeval span is on ``current_span_context`` and
|
||
the OTel span is a root (no native parent), the interceptor
|
||
stamps ``confident.span.parent_uuid`` so the exporter can
|
||
re-parent the OTel root onto the deepeval span instead of
|
||
emitting it as a sibling.
|
||
"""
|
||
from deepeval.tracing.types import BaseSpan, TraceSpanStatus
|
||
|
||
outer = BaseSpan(
|
||
uuid="deepeval-outer-uuid",
|
||
trace_uuid="deepeval-trace-uuid",
|
||
status=TraceSpanStatus.IN_PROGRESS,
|
||
start_time=0.0,
|
||
)
|
||
token = current_span_context.set(outer)
|
||
try:
|
||
settings = _make_settings()
|
||
interceptor = OpenInferenceSpanInterceptor(settings)
|
||
root = _make_mock_span() # parent=None makes it a root
|
||
|
||
interceptor.on_start(root, None)
|
||
interceptor.on_end(root)
|
||
|
||
assert (
|
||
root.attributes.get("confident.span.parent_uuid")
|
||
== "deepeval-outer-uuid"
|
||
)
|
||
finally:
|
||
current_span_context.reset(token)
|
||
|
||
def test_no_parent_uuid_when_otel_span_has_native_parent(self):
|
||
"""OTel children already have a real parent_id pointing into
|
||
the same OTel trace — no need to bridge."""
|
||
from deepeval.tracing.types import BaseSpan, TraceSpanStatus
|
||
|
||
outer = BaseSpan(
|
||
uuid="deepeval-outer-uuid",
|
||
trace_uuid="deepeval-trace-uuid",
|
||
status=TraceSpanStatus.IN_PROGRESS,
|
||
start_time=0.0,
|
||
)
|
||
token = current_span_context.set(outer)
|
||
try:
|
||
settings = _make_settings()
|
||
interceptor = OpenInferenceSpanInterceptor(settings)
|
||
child = _make_mock_span(parent=MagicMock())
|
||
|
||
interceptor.on_start(child, None)
|
||
interceptor.on_end(child)
|
||
|
||
assert "confident.span.parent_uuid" not in child.attributes
|
||
finally:
|
||
current_span_context.reset(token)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# next_*_span(...) consumption + stash_pending_metrics gating.
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestNextSpanInterceptorIntegration:
|
||
def test_next_agent_span_metric_collection_lands_on_otel_attrs(self):
|
||
settings = _make_settings()
|
||
interceptor = OpenInferenceSpanInterceptor(settings)
|
||
span = _make_agent_span_mock()
|
||
|
||
with next_agent_span(metric_collection="agent_metrics_v1"):
|
||
interceptor.on_start(span, None)
|
||
interceptor.on_end(span)
|
||
|
||
assert (
|
||
span.attributes.get("confident.span.metric_collection")
|
||
== "agent_metrics_v1"
|
||
)
|
||
|
||
def test_next_agent_span_consumed_only_by_first_agent_span(self):
|
||
"""One-shot semantics through the interceptor: a second agent
|
||
span inside the same ``with`` block does NOT inherit."""
|
||
settings = _make_settings()
|
||
interceptor = OpenInferenceSpanInterceptor(settings)
|
||
first = _make_agent_span_mock("agent_one")
|
||
second = _make_agent_span_mock("agent_two")
|
||
|
||
with next_agent_span(metric_collection="only-first"):
|
||
interceptor.on_start(first, None)
|
||
interceptor.on_end(first)
|
||
|
||
interceptor.on_start(second, None)
|
||
interceptor.on_end(second)
|
||
|
||
assert (
|
||
first.attributes.get("confident.span.metric_collection")
|
||
== "only-first"
|
||
)
|
||
assert second.attributes.get("confident.span.metric_collection") is None
|
||
|
||
def test_next_agent_span_does_not_affect_non_agent_span(self):
|
||
"""Typed slot is NOT consumed by spans of a different type. An
|
||
LLM span fired inside ``with next_agent_span(...)`` should pop
|
||
nothing from the agent slot."""
|
||
settings = _make_settings()
|
||
interceptor = OpenInferenceSpanInterceptor(settings)
|
||
llm_span = _make_llm_span_mock()
|
||
agent_span = _make_agent_span_mock()
|
||
|
||
with next_agent_span(metric_collection="agent-only"):
|
||
interceptor.on_start(llm_span, None)
|
||
interceptor.on_end(llm_span)
|
||
|
||
interceptor.on_start(agent_span, None)
|
||
interceptor.on_end(agent_span)
|
||
|
||
assert (
|
||
llm_span.attributes.get("confident.span.metric_collection") is None
|
||
)
|
||
assert (
|
||
agent_span.attributes.get("confident.span.metric_collection")
|
||
== "agent-only"
|
||
)
|
||
|
||
def test_next_tool_span_metric_collection_lands_on_tool_otel_attrs(self):
|
||
"""Mirrors the ``test_tool_metric_collection`` flow in test_sync.py
|
||
— ``with next_tool_span(metric_collection=...)`` sets the value
|
||
on the FIRST tool span emitted inside the block."""
|
||
settings = _make_settings()
|
||
interceptor = OpenInferenceSpanInterceptor(settings)
|
||
tool_span = _make_tool_span_mock("calculate")
|
||
|
||
with next_tool_span(metric_collection="calculator-metrics"):
|
||
interceptor.on_start(tool_span, None)
|
||
interceptor.on_end(tool_span)
|
||
|
||
assert (
|
||
tool_span.attributes.get("confident.span.metric_collection")
|
||
== "calculator-metrics"
|
||
)
|
||
|
||
def test_next_llm_span_metric_collection_lands_on_llm_otel_attrs(self):
|
||
settings = _make_settings()
|
||
interceptor = OpenInferenceSpanInterceptor(settings)
|
||
llm_span = _make_llm_span_mock()
|
||
|
||
with next_llm_span(metric_collection="llm_metrics_v1"):
|
||
interceptor.on_start(llm_span, None)
|
||
interceptor.on_end(llm_span)
|
||
|
||
assert (
|
||
llm_span.attributes.get("confident.span.metric_collection")
|
||
== "llm_metrics_v1"
|
||
)
|
||
|
||
def test_update_current_span_overrides_next_agent_span_after_creation(
|
||
self,
|
||
):
|
||
"""Last-write-wins: ``next_agent_span`` sets the floor at
|
||
on_start; later ``update_current_span(...)`` (e.g. from inside
|
||
a tool body) overwrites."""
|
||
settings = _make_settings()
|
||
interceptor = OpenInferenceSpanInterceptor(settings)
|
||
span = _make_agent_span_mock()
|
||
|
||
with next_agent_span(metric_collection="from-wrapper"):
|
||
interceptor.on_start(span, None)
|
||
update_current_span(metric_collection="from-update")
|
||
interceptor.on_end(span)
|
||
|
||
assert (
|
||
span.attributes.get("confident.span.metric_collection")
|
||
== "from-update"
|
||
)
|
||
|
||
def test_next_agent_span_metrics_stashed_when_evaluating(self):
|
||
"""``with next_agent_span(metrics=[...])`` populates the
|
||
placeholder; at on_end the interceptor calls
|
||
``stash_pending_metrics`` so ``ConfidentSpanExporter`` can
|
||
re-attach the ``BaseMetric`` instances after rebuilding the
|
||
span (they don't fit in OTel primitives-only attrs).
|
||
|
||
Gated on ``trace_manager.is_evaluating`` to keep the registry
|
||
from growing in production paths.
|
||
"""
|
||
from deepeval.metrics import AnswerRelevancyMetric
|
||
|
||
settings = _make_settings()
|
||
interceptor = OpenInferenceSpanInterceptor(settings)
|
||
span = _make_agent_span_mock()
|
||
metric = AnswerRelevancyMetric()
|
||
|
||
with patch(
|
||
"deepeval.integrations.openinference.instrumentator."
|
||
"stash_pending_metrics"
|
||
) as stash, patch(
|
||
"deepeval.integrations.openinference.instrumentator.trace_manager"
|
||
) as fake_tm:
|
||
fake_tm.is_evaluating = True
|
||
with next_agent_span(metrics=[metric]):
|
||
interceptor.on_start(span, None)
|
||
interceptor.on_end(span)
|
||
|
||
stash.assert_called_once()
|
||
# First positional arg = uuid (16-char hex), second = metrics list.
|
||
args, _ = stash.call_args
|
||
assert isinstance(args[0], str) and len(args[0]) == 16
|
||
assert args[1] == [metric]
|
||
|
||
def test_next_agent_span_metrics_not_stashed_outside_eval_mode(self):
|
||
"""In production paths (``is_evaluating=False``) the metrics
|
||
overlay would leak — gate prevents the stash."""
|
||
from deepeval.metrics import AnswerRelevancyMetric
|
||
|
||
settings = _make_settings()
|
||
interceptor = OpenInferenceSpanInterceptor(settings)
|
||
span = _make_agent_span_mock()
|
||
metric = AnswerRelevancyMetric()
|
||
|
||
with patch(
|
||
"deepeval.integrations.openinference.instrumentator."
|
||
"stash_pending_metrics"
|
||
) as stash, patch(
|
||
"deepeval.integrations.openinference.instrumentator.trace_manager"
|
||
) as fake_tm:
|
||
fake_tm.is_evaluating = False
|
||
with next_agent_span(metrics=[metric]):
|
||
interceptor.on_start(span, None)
|
||
interceptor.on_end(span)
|
||
|
||
stash.assert_not_called()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# OpenInference framework-attr extraction (the bit that's specific to
|
||
# this interceptor — AgentCore reads gen_ai.* / Strands events instead).
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestFrameworkAttrExtraction:
|
||
"""Verifies the ``_serialize_framework_attrs`` path: classification,
|
||
flattened message extraction, tool-call extraction (Scenario A:
|
||
span IS a tool, Scenario B: tool calls embedded in an LLM output
|
||
message), token counts, and model name. Framework attrs run with
|
||
``setdefault`` semantics — the placeholder serializer ran first
|
||
so ``update_current_span(...)`` writes win over framework writes."""
|
||
|
||
def test_agent_span_kind_lands_as_confident_span_type_agent(self):
|
||
settings = _make_settings()
|
||
interceptor = OpenInferenceSpanInterceptor(settings)
|
||
span = _make_agent_span_mock("planner")
|
||
|
||
interceptor.on_start(span, None)
|
||
interceptor.on_end(span)
|
||
|
||
assert span.attributes.get("confident.span.type") == "agent"
|
||
assert span.attributes.get("confident.span.name") == "planner"
|
||
|
||
def test_chain_span_kind_classified_as_agent(self):
|
||
"""OpenInference uses CHAIN for orchestration nodes that look
|
||
agent-shaped to deepeval — both flow into AgentSpan."""
|
||
settings = _make_settings()
|
||
interceptor = OpenInferenceSpanInterceptor(settings)
|
||
span = _make_mock_span(span_kind="CHAIN", agent_name="root_chain")
|
||
|
||
interceptor.on_start(span, None)
|
||
interceptor.on_end(span)
|
||
|
||
assert span.attributes.get("confident.span.type") == "agent"
|
||
|
||
def test_llm_span_kind_lands_as_confident_span_type_llm(self):
|
||
settings = _make_settings()
|
||
interceptor = OpenInferenceSpanInterceptor(settings)
|
||
span = _make_llm_span_mock()
|
||
|
||
interceptor.on_start(span, None)
|
||
interceptor.on_end(span)
|
||
|
||
assert span.attributes.get("confident.span.type") == "llm"
|
||
|
||
def test_tool_span_kind_lands_as_confident_span_type_tool(self):
|
||
settings = _make_settings()
|
||
interceptor = OpenInferenceSpanInterceptor(settings)
|
||
span = _make_tool_span_mock("calculate")
|
||
|
||
interceptor.on_start(span, None)
|
||
interceptor.on_end(span)
|
||
|
||
assert span.attributes.get("confident.span.type") == "tool"
|
||
assert span.attributes.get("confident.span.name") == "calculate"
|
||
|
||
def test_unknown_span_kind_classified_as_custom(self):
|
||
"""Anything that's not AGENT / CHAIN / LLM / TOOL / RETRIEVER
|
||
falls through to ``custom`` so non-standard OpenInference
|
||
instrumentors still get represented."""
|
||
settings = _make_settings()
|
||
interceptor = OpenInferenceSpanInterceptor(settings)
|
||
span = _make_mock_span(span_kind="GUARDRAIL")
|
||
|
||
interceptor.on_start(span, None)
|
||
interceptor.on_end(span)
|
||
|
||
assert span.attributes.get("confident.span.type") == "custom"
|
||
|
||
def test_missing_span_kind_leaves_type_unset(self):
|
||
"""Spans without ``openinference.span.kind`` are not
|
||
OpenInference-emitted; the interceptor must not stamp a type
|
||
on them so they don't get rebuilt as malformed deepeval spans."""
|
||
settings = _make_settings()
|
||
interceptor = OpenInferenceSpanInterceptor(settings)
|
||
span = _make_mock_span() # no kind set
|
||
|
||
interceptor.on_start(span, None)
|
||
interceptor.on_end(span)
|
||
|
||
assert "confident.span.type" not in span.attributes
|
||
|
||
def test_llm_span_extracts_flattened_input_output_messages(self):
|
||
"""OpenInference flattens chat history into
|
||
``llm.{input,output}_messages.{idx}.message.content``. The
|
||
interceptor walks the indexes until a hole, takes the LAST
|
||
seen content, and writes it to ``confident.span.{input,output}``.
|
||
"""
|
||
settings = _make_settings()
|
||
interceptor = OpenInferenceSpanInterceptor(settings)
|
||
span = _make_mock_span(
|
||
span_kind="LLM",
|
||
extra_attrs={
|
||
"llm.input_messages.0.message.role": "system",
|
||
"llm.input_messages.0.message.content": "You are concise.",
|
||
"llm.input_messages.1.message.role": "user",
|
||
"llm.input_messages.1.message.content": "Hello?",
|
||
"llm.output_messages.0.message.role": "assistant",
|
||
"llm.output_messages.0.message.content": "Hi!",
|
||
},
|
||
)
|
||
|
||
interceptor.on_start(span, None)
|
||
interceptor.on_end(span)
|
||
|
||
# Last input message wins (assistant context normally trails
|
||
# at output); for input we expect the latest user turn.
|
||
assert span.attributes.get("confident.span.input") == "Hello?"
|
||
assert span.attributes.get("confident.span.output") == "Hi!"
|
||
|
||
def test_llm_span_extracts_token_counts_and_model_name(self):
|
||
settings = _make_settings()
|
||
interceptor = OpenInferenceSpanInterceptor(settings)
|
||
span = _make_mock_span(
|
||
span_kind="LLM",
|
||
extra_attrs={
|
||
"llm.token_count.prompt": 42,
|
||
"llm.token_count.completion": 17,
|
||
"llm.model_name": "gemini-2.0-flash",
|
||
},
|
||
)
|
||
|
||
interceptor.on_start(span, None)
|
||
interceptor.on_end(span)
|
||
|
||
assert span.attributes.get("confident.llm.input_token_count") == 42
|
||
assert span.attributes.get("confident.llm.output_token_count") == 17
|
||
assert span.attributes.get("confident.llm.model") == "gemini-2.0-flash"
|
||
|
||
def test_llm_span_extracts_tool_calls_from_output_messages(self):
|
||
"""Scenario B: tool calls embedded inside an LLM output
|
||
message via the flattened
|
||
``llm.output_messages.{idx}.message.tool_calls.{tc}.tool_call.function.{name,arguments}``
|
||
attrs. The interceptor walks ``msg_idx`` outer × ``tc_idx``
|
||
inner, JSON-parses ``arguments``, and emits a
|
||
``confident.span.tools_called`` JSON list of ``ToolCall``s.
|
||
"""
|
||
settings = _make_settings()
|
||
interceptor = OpenInferenceSpanInterceptor(settings)
|
||
span = _make_mock_span(
|
||
span_kind="LLM",
|
||
extra_attrs={
|
||
"llm.output_messages.0.message.role": "assistant",
|
||
"llm.output_messages.0.message.content": "",
|
||
"llm.output_messages.0.message.tool_calls.0."
|
||
"tool_call.function.name": "get_weather",
|
||
"llm.output_messages.0.message.tool_calls.0."
|
||
"tool_call.function.arguments": '{"city": "Tokyo"}',
|
||
"llm.output_messages.0.message.tool_calls.1."
|
||
"tool_call.function.name": "get_time",
|
||
"llm.output_messages.0.message.tool_calls.1."
|
||
"tool_call.function.arguments": '{"city": "Tokyo"}',
|
||
},
|
||
)
|
||
|
||
interceptor.on_start(span, None)
|
||
interceptor.on_end(span)
|
||
|
||
raw = span.attributes.get("confident.span.tools_called")
|
||
assert raw is not None
|
||
# Each entry is a ToolCall.model_dump_json() string.
|
||
parsed = [json.loads(item) for item in raw]
|
||
names = sorted(p["name"] for p in parsed)
|
||
assert names == ["get_time", "get_weather"]
|
||
for p in parsed:
|
||
assert p["input_parameters"] == {"city": "Tokyo"}
|
||
|
||
def test_tool_span_extracts_self_as_single_tool_call(self):
|
||
"""Scenario A: the span itself is a tool span (kind=TOOL),
|
||
so the framework extractor builds a 1-element
|
||
``confident.span.tools_called`` from ``tool.name`` /
|
||
``tool.parameters`` and copies the parameters into
|
||
``confident.span.input`` for visibility."""
|
||
settings = _make_settings()
|
||
interceptor = OpenInferenceSpanInterceptor(settings)
|
||
span = _make_mock_span(
|
||
span_kind="TOOL",
|
||
tool_name="get_weather",
|
||
extra_attrs={
|
||
"tool.parameters": '{"city": "Paris"}',
|
||
},
|
||
)
|
||
|
||
interceptor.on_start(span, None)
|
||
interceptor.on_end(span)
|
||
|
||
raw = span.attributes.get("confident.span.tools_called")
|
||
assert raw is not None
|
||
assert len(raw) == 1
|
||
parsed = json.loads(raw[0])
|
||
assert parsed["name"] == "get_weather"
|
||
assert parsed["input_parameters"] == {"city": "Paris"}
|
||
# ``confident.span.input`` was empty (no update_current_span);
|
||
# framework path fills it from the tool params.
|
||
assert json.loads(span.attributes["confident.span.input"]) == {
|
||
"city": "Paris"
|
||
}
|
||
|
||
def test_agent_span_input_output_also_lands_on_trace_attrs(self):
|
||
"""Agent (root) spans surface their I/O onto
|
||
``confident.trace.{input,output}`` too so the trace card has
|
||
prompt + response without re-walking spans server-side."""
|
||
settings = _make_settings()
|
||
interceptor = OpenInferenceSpanInterceptor(settings)
|
||
span = _make_mock_span(
|
||
span_kind="AGENT",
|
||
agent_name="planner",
|
||
extra_attrs={
|
||
"input.value": "What's the weather in Tokyo?",
|
||
"output.value": "Sunny, 72F.",
|
||
},
|
||
)
|
||
|
||
interceptor.on_start(span, None)
|
||
interceptor.on_end(span)
|
||
|
||
assert (
|
||
span.attributes.get("confident.span.input")
|
||
== "What's the weather in Tokyo?"
|
||
)
|
||
assert span.attributes.get("confident.span.output") == "Sunny, 72F."
|
||
assert (
|
||
span.attributes.get("confident.trace.input")
|
||
== "What's the weather in Tokyo?"
|
||
)
|
||
assert span.attributes.get("confident.trace.output") == "Sunny, 72F."
|
||
|
||
def test_update_current_span_input_wins_over_framework_input(self):
|
||
"""Framework path uses ``setdefault`` semantics — when the
|
||
placeholder serializer (which runs first) already stamped
|
||
``confident.span.input``, the framework path must not
|
||
overwrite it. Regression guard for the layering order."""
|
||
settings = _make_settings()
|
||
interceptor = OpenInferenceSpanInterceptor(settings)
|
||
span = _make_mock_span(
|
||
span_kind="LLM",
|
||
extra_attrs={
|
||
"llm.input_messages.0.message.role": "user",
|
||
"llm.input_messages.0.message.content": "framework-input",
|
||
},
|
||
)
|
||
|
||
interceptor.on_start(span, None)
|
||
update_current_span(input="user-supplied-input")
|
||
interceptor.on_end(span)
|
||
|
||
assert (
|
||
json.loads(span.attributes["confident.span.input"])
|
||
== "user-supplied-input"
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Removed kwargs: settings + instrument_google_adk signature.
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
_REMOVED_KWARGS = [
|
||
"is_test_mode",
|
||
"agent_metric_collection",
|
||
"llm_metric_collection",
|
||
"tool_metric_collection_map",
|
||
"trace_metric_collection",
|
||
"agent_metrics",
|
||
"confident_prompt",
|
||
]
|
||
|
||
|
||
@pytest.mark.parametrize("kwarg", _REMOVED_KWARGS)
|
||
def test_removed_kwargs_raise_typeerror_on_settings(kwarg):
|
||
"""Span-level kwargs were removed in the OTel POC migration. Each
|
||
must raise ``TypeError`` on construction so callers see exactly
|
||
which kwarg to migrate."""
|
||
with pytest.raises(TypeError) as exc:
|
||
OpenInferenceInstrumentationSettings(
|
||
api_key="dummy", **{kwarg: object()}
|
||
)
|
||
|
||
# The error message names the removed kwarg, so a future expansion
|
||
# of ``_REMOVED_KWARGS`` doesn't accidentally swallow it.
|
||
assert kwarg in str(exc.value)
|
||
|
||
|
||
@pytest.mark.parametrize("kwarg", _REMOVED_KWARGS)
|
||
def test_removed_kwargs_raise_typeerror_on_instrument_google_adk(kwarg):
|
||
"""Same guard at the ``instrument_google_adk(...)`` entry point —
|
||
catches callers that bypass the settings constructor. The kwarg
|
||
check fires BEFORE the GoogleADKInstrumentor import, so this test
|
||
works without ``openinference-instrumentation-google-adk`` installed.
|
||
"""
|
||
from deepeval.integrations.google_adk import instrument_google_adk
|
||
|
||
with pytest.raises(TypeError) as exc:
|
||
instrument_google_adk(api_key="dummy", **{kwarg: object()})
|
||
|
||
assert kwarg in str(exc.value)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Optional Confident AI api_key — must NOT be required.
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_settings_no_api_key_does_not_raise(monkeypatch):
|
||
"""Constructor must succeed when no api_key is supplied or in env.
|
||
|
||
The OTel pipeline still wires up locally — only the outbound auth
|
||
header is gated on a key being present (handled in
|
||
``ContextAwareSpanProcessor``, not the settings constructor).
|
||
"""
|
||
monkeypatch.delenv("CONFIDENT_API_KEY", raising=False)
|
||
instance = OpenInferenceInstrumentationSettings()
|
||
assert instance is not None
|
||
assert instance.api_key is None
|