473 lines
18 KiB
Python
473 lines
18 KiB
Python
"""Unit tests for telemetry module — OpenTelemetry integration"""
|
|
|
|
import asyncio
|
|
import importlib
|
|
import sys
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
|
|
# ── No-op behavior when OTel not installed ──────────────────────────
|
|
|
|
|
|
class TestNoOpWhenOTelNotInstalled:
|
|
"""All operations are no-op when opentelemetry is not installed."""
|
|
|
|
def test_tracing_noop_span_context_manager(self):
|
|
"""_NoOpSpan works as context manager without errors."""
|
|
from agentkit.telemetry.tracing import _NoOpSpan
|
|
|
|
span = _NoOpSpan()
|
|
with span as s:
|
|
s.set_attribute("key", "value")
|
|
s.add_event("event")
|
|
s.set_status("ok")
|
|
s.record_exception(Exception("test"))
|
|
|
|
def test_get_tracer_returns_none_without_otel(self):
|
|
"""get_tracer returns None when OTel is not installed."""
|
|
from agentkit.telemetry.tracing import _OTEL_AVAILABLE, get_tracer
|
|
|
|
if _OTEL_AVAILABLE:
|
|
pytest.skip("OTel is installed, skipping no-op test")
|
|
assert get_tracer() is None
|
|
|
|
def test_start_span_returns_noop_without_otel(self):
|
|
"""start_span returns no-op span when OTel is not installed."""
|
|
from agentkit.telemetry.tracing import _OTEL_AVAILABLE, start_span, _NoOpSpan
|
|
|
|
if _OTEL_AVAILABLE:
|
|
pytest.skip("OTel is installed, skipping no-op test")
|
|
span_cm = start_span("test.span")
|
|
assert isinstance(span_cm, _NoOpSpan)
|
|
|
|
def test_metrics_noop_counter(self):
|
|
"""No-op counter add() does not raise."""
|
|
from agentkit.telemetry.metrics import _NoOpCounter
|
|
|
|
counter = _NoOpCounter()
|
|
counter.add(1, {"key": "value"}) # Should not raise
|
|
|
|
def test_metrics_noop_histogram(self):
|
|
"""No-op histogram record() does not raise."""
|
|
from agentkit.telemetry.metrics import _NoOpHistogram
|
|
|
|
hist = _NoOpHistogram()
|
|
hist.record(100, {"key": "value"}) # Should not raise
|
|
|
|
def test_metrics_get_meter_returns_none_without_otel(self):
|
|
"""get_meter returns None when OTel is not installed."""
|
|
from agentkit.telemetry.metrics import _OTEL_AVAILABLE, get_meter
|
|
|
|
if _OTEL_AVAILABLE:
|
|
pytest.skip("OTel is installed, skipping no-op test")
|
|
assert get_meter() is None
|
|
|
|
def test_metric_helpers_return_noop_without_otel(self):
|
|
"""Metric helper functions return no-op instruments when OTel not installed."""
|
|
from agentkit.telemetry.metrics import (
|
|
_OTEL_AVAILABLE,
|
|
_NoOpCounter,
|
|
_NoOpHistogram,
|
|
agent_request_counter,
|
|
agent_duration_histogram,
|
|
llm_token_histogram,
|
|
tool_duration_histogram,
|
|
pipeline_step_histogram,
|
|
)
|
|
|
|
if _OTEL_AVAILABLE:
|
|
pytest.skip("OTel is installed, skipping no-op test")
|
|
|
|
# Reset lazy singletons to force re-creation
|
|
import agentkit.telemetry.metrics as m
|
|
m._agent_request_counter = None
|
|
m._agent_duration_histogram = None
|
|
m._llm_token_histogram = None
|
|
m._tool_duration_histogram = None
|
|
m._pipeline_step_histogram = None
|
|
|
|
assert isinstance(agent_request_counter(), _NoOpCounter)
|
|
assert isinstance(agent_duration_histogram(), _NoOpHistogram)
|
|
assert isinstance(llm_token_histogram(), _NoOpHistogram)
|
|
assert isinstance(tool_duration_histogram(), _NoOpHistogram)
|
|
assert isinstance(pipeline_step_histogram(), _NoOpHistogram)
|
|
|
|
|
|
# ── Tracing decorator tests ─────────────────────────────────────────
|
|
|
|
|
|
class TestTraceAgentDecorator:
|
|
"""trace_agent decorator works with and without OTel."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_decorator_works_without_otel(self):
|
|
"""trace_agent decorator passes through when OTel not installed."""
|
|
from agentkit.telemetry.tracing import _OTEL_AVAILABLE, trace_agent
|
|
|
|
if _OTEL_AVAILABLE:
|
|
pytest.skip("OTel is installed, skipping no-op test")
|
|
|
|
@trace_agent("test_agent", "react")
|
|
async def my_func():
|
|
return "result"
|
|
|
|
result = await my_func()
|
|
assert result == "result"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_decorator_propagates_exception_without_otel(self):
|
|
"""trace_agent propagates exceptions when OTel not installed."""
|
|
from agentkit.telemetry.tracing import _OTEL_AVAILABLE, trace_agent
|
|
|
|
if _OTEL_AVAILABLE:
|
|
pytest.skip("OTel is installed, skipping no-op test")
|
|
|
|
@trace_agent("test_agent")
|
|
async def my_func():
|
|
raise ValueError("test error")
|
|
|
|
with pytest.raises(ValueError, match="test error"):
|
|
await my_func()
|
|
|
|
|
|
class TestTraceToolDecorator:
|
|
"""trace_tool decorator tests."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_decorator_works_without_otel(self):
|
|
"""trace_tool decorator passes through when OTel not installed."""
|
|
from agentkit.telemetry.tracing import _OTEL_AVAILABLE, trace_tool
|
|
|
|
if _OTEL_AVAILABLE:
|
|
pytest.skip("OTel is installed, skipping no-op test")
|
|
|
|
@trace_tool("my_tool")
|
|
async def my_func():
|
|
return {"result": "ok"}
|
|
|
|
result = await my_func()
|
|
assert result == {"result": "ok"}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_decorator_propagates_exception_without_otel(self):
|
|
"""trace_tool propagates exceptions when OTel not installed."""
|
|
from agentkit.telemetry.tracing import _OTEL_AVAILABLE, trace_tool
|
|
|
|
if _OTEL_AVAILABLE:
|
|
pytest.skip("OTel is installed, skipping no-op test")
|
|
|
|
@trace_tool("my_tool")
|
|
async def my_func():
|
|
raise RuntimeError("tool error")
|
|
|
|
with pytest.raises(RuntimeError, match="tool error"):
|
|
await my_func()
|
|
|
|
|
|
class TestTraceLLMDecorator:
|
|
"""trace_llm decorator tests."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_decorator_works_without_otel(self):
|
|
"""trace_llm decorator passes through when OTel not installed."""
|
|
from agentkit.telemetry.tracing import _OTEL_AVAILABLE, trace_llm
|
|
|
|
if _OTEL_AVAILABLE:
|
|
pytest.skip("OTel is installed, skipping no-op test")
|
|
|
|
@trace_llm("openai", "gpt-4")
|
|
async def my_func():
|
|
return MagicMock(usage=MagicMock(prompt_tokens=10, completion_tokens=20))
|
|
|
|
result = await my_func()
|
|
assert result is not None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_decorator_propagates_exception_without_otel(self):
|
|
"""trace_llm propagates exceptions when OTel not installed."""
|
|
from agentkit.telemetry.tracing import _OTEL_AVAILABLE, trace_llm
|
|
|
|
if _OTEL_AVAILABLE:
|
|
pytest.skip("OTel is installed, skipping no-op test")
|
|
|
|
@trace_llm("openai", "gpt-4")
|
|
async def my_func():
|
|
raise ConnectionError("LLM error")
|
|
|
|
with pytest.raises(ConnectionError, match="LLM error"):
|
|
await my_func()
|
|
|
|
|
|
class TestTracePipelineStepDecorator:
|
|
"""trace_pipeline_step decorator tests."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_decorator_works_without_otel(self):
|
|
"""trace_pipeline_step decorator passes through when OTel not installed."""
|
|
from agentkit.telemetry.tracing import _OTEL_AVAILABLE, trace_pipeline_step
|
|
|
|
if _OTEL_AVAILABLE:
|
|
pytest.skip("OTel is installed, skipping no-op test")
|
|
|
|
@trace_pipeline_step("my_pipeline", "step_1")
|
|
async def my_func():
|
|
return "step_result"
|
|
|
|
result = await my_func()
|
|
assert result == "step_result"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_decorator_propagates_exception_without_otel(self):
|
|
"""trace_pipeline_step propagates exceptions when OTel not installed."""
|
|
from agentkit.telemetry.tracing import _OTEL_AVAILABLE, trace_pipeline_step
|
|
|
|
if _OTEL_AVAILABLE:
|
|
pytest.skip("OTel is installed, skipping no-op test")
|
|
|
|
@trace_pipeline_step("my_pipeline", "step_1")
|
|
async def my_func():
|
|
raise RuntimeError("step failed")
|
|
|
|
with pytest.raises(RuntimeError, match="step failed"):
|
|
await my_func()
|
|
|
|
|
|
# ── OTel installed (mocked) tests ───────────────────────────────────
|
|
|
|
|
|
class TestTracingWithMockedOTel:
|
|
"""Test tracing with mocked OTel imports."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trace_agent_with_mocked_otel(self):
|
|
"""trace_agent creates span with correct attributes when OTel is available."""
|
|
mock_span = MagicMock()
|
|
mock_span_cm = MagicMock()
|
|
mock_span_cm.__enter__ = MagicMock(return_value=mock_span)
|
|
mock_span_cm.__exit__ = MagicMock(return_value=False)
|
|
|
|
mock_tracer = MagicMock()
|
|
mock_tracer.start_as_current_span.return_value = mock_span_cm
|
|
|
|
with patch("agentkit.telemetry.tracing._OTEL_AVAILABLE", True), \
|
|
patch("agentkit.telemetry.tracing.get_tracer", return_value=mock_tracer), \
|
|
patch("agentkit.telemetry.tracing.SpanKind"), \
|
|
patch("agentkit.telemetry.tracing.Status"), \
|
|
patch("agentkit.telemetry.tracing.StatusCode"):
|
|
|
|
from agentkit.telemetry.tracing import trace_agent
|
|
|
|
@trace_agent("test_agent", "react")
|
|
async def my_func():
|
|
return "result"
|
|
|
|
result = await my_func()
|
|
assert result == "result"
|
|
mock_tracer.start_as_current_span.assert_called_once()
|
|
call_kwargs = mock_tracer.start_as_current_span.call_args
|
|
assert call_kwargs[1]["attributes"]["agent.name"] == "test_agent"
|
|
assert call_kwargs[1]["attributes"]["agent.type"] == "react"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trace_tool_with_mocked_otel(self):
|
|
"""trace_tool creates span with tool.name attribute."""
|
|
mock_span = MagicMock()
|
|
mock_span_cm = MagicMock()
|
|
mock_span_cm.__enter__ = MagicMock(return_value=mock_span)
|
|
mock_span_cm.__exit__ = MagicMock(return_value=False)
|
|
|
|
mock_tracer = MagicMock()
|
|
mock_tracer.start_as_current_span.return_value = mock_span_cm
|
|
|
|
with patch("agentkit.telemetry.tracing._OTEL_AVAILABLE", True), \
|
|
patch("agentkit.telemetry.tracing.get_tracer", return_value=mock_tracer), \
|
|
patch("agentkit.telemetry.tracing.SpanKind"), \
|
|
patch("agentkit.telemetry.tracing.Status"), \
|
|
patch("agentkit.telemetry.tracing.StatusCode"):
|
|
|
|
from agentkit.telemetry.tracing import trace_tool
|
|
|
|
@trace_tool("search_tool")
|
|
async def my_func():
|
|
return {"found": True}
|
|
|
|
result = await my_func()
|
|
assert result == {"found": True}
|
|
call_kwargs = mock_tracer.start_as_current_span.call_args
|
|
assert call_kwargs[1]["attributes"]["tool.name"] == "search_tool"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trace_llm_with_mocked_otel(self):
|
|
"""trace_llm creates span with gen_ai semantic conventions."""
|
|
mock_span = MagicMock()
|
|
mock_span_cm = MagicMock()
|
|
mock_span_cm.__enter__ = MagicMock(return_value=mock_span)
|
|
mock_span_cm.__exit__ = MagicMock(return_value=False)
|
|
|
|
mock_tracer = MagicMock()
|
|
mock_tracer.start_as_current_span.return_value = mock_span_cm
|
|
|
|
mock_usage = MagicMock()
|
|
mock_usage.prompt_tokens = 50
|
|
mock_usage.completion_tokens = 100
|
|
mock_response = MagicMock()
|
|
mock_response.usage = mock_usage
|
|
|
|
with patch("agentkit.telemetry.tracing._OTEL_AVAILABLE", True), \
|
|
patch("agentkit.telemetry.tracing.get_tracer", return_value=mock_tracer), \
|
|
patch("agentkit.telemetry.tracing.SpanKind"), \
|
|
patch("agentkit.telemetry.tracing.Status"), \
|
|
patch("agentkit.telemetry.tracing.StatusCode"):
|
|
|
|
from agentkit.telemetry.tracing import trace_llm
|
|
|
|
@trace_llm("openai", "gpt-4")
|
|
async def my_func():
|
|
return mock_response
|
|
|
|
result = await my_func()
|
|
assert result is mock_response
|
|
call_kwargs = mock_tracer.start_as_current_span.call_args
|
|
attrs = call_kwargs[1]["attributes"]
|
|
assert attrs["gen_ai.system"] == "openai"
|
|
assert attrs["gen_ai.operation.name"] == "chat"
|
|
assert attrs["gen_ai.request.model"] == "gpt-4"
|
|
# Token usage should be recorded on span
|
|
mock_span.set_attribute.assert_any_call("gen_ai.usage.input_tokens", 50)
|
|
mock_span.set_attribute.assert_any_call("gen_ai.usage.output_tokens", 100)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trace_pipeline_step_with_mocked_otel(self):
|
|
"""trace_pipeline_step creates span with pipeline and step attributes."""
|
|
mock_span = MagicMock()
|
|
mock_span_cm = MagicMock()
|
|
mock_span_cm.__enter__ = MagicMock(return_value=mock_span)
|
|
mock_span_cm.__exit__ = MagicMock(return_value=False)
|
|
|
|
mock_tracer = MagicMock()
|
|
mock_tracer.start_as_current_span.return_value = mock_span_cm
|
|
|
|
with patch("agentkit.telemetry.tracing._OTEL_AVAILABLE", True), \
|
|
patch("agentkit.telemetry.tracing.get_tracer", return_value=mock_tracer), \
|
|
patch("agentkit.telemetry.tracing.SpanKind"), \
|
|
patch("agentkit.telemetry.tracing.Status"), \
|
|
patch("agentkit.telemetry.tracing.StatusCode"):
|
|
|
|
from agentkit.telemetry.tracing import trace_pipeline_step
|
|
|
|
@trace_pipeline_step("geo_pipeline", "analyze")
|
|
async def my_func():
|
|
return "done"
|
|
|
|
result = await my_func()
|
|
assert result == "done"
|
|
call_kwargs = mock_tracer.start_as_current_span.call_args
|
|
attrs = call_kwargs[1]["attributes"]
|
|
assert attrs["pipeline.name"] == "geo_pipeline"
|
|
assert attrs["step.name"] == "analyze"
|
|
|
|
|
|
# ── setup_telemetry tests ───────────────────────────────────────────
|
|
|
|
|
|
class TestSetupTelemetry:
|
|
"""setup_telemetry initialization tests."""
|
|
|
|
def test_no_config_is_noop(self):
|
|
"""setup_telemetry with no config is a no-op."""
|
|
from agentkit.telemetry.setup import setup_telemetry
|
|
|
|
mock_app = MagicMock()
|
|
setup_telemetry(mock_app, None) # Should not raise
|
|
# No auto-instrumentation should happen
|
|
mock_app.state = MagicMock() # Just ensure no crash
|
|
|
|
def test_disabled_config_is_noop(self):
|
|
"""setup_telemetry with enabled=False is a no-op."""
|
|
from agentkit.telemetry.setup import setup_telemetry
|
|
|
|
mock_app = MagicMock()
|
|
setup_telemetry(mock_app, {"enabled": False}) # Should not raise
|
|
|
|
def test_config_without_otel_logs_warning(self):
|
|
"""setup_telemetry with config but OTel not installed logs warning."""
|
|
from agentkit.telemetry.setup import setup_telemetry
|
|
|
|
mock_app = MagicMock()
|
|
# This should not raise even if OTel is not installed
|
|
# It will log a warning internally
|
|
config = {"enabled": True, "service_name": "test"}
|
|
# If OTel is installed, this will try to set up providers
|
|
# If not, it will log a warning and return
|
|
setup_telemetry(mock_app, config) # Should not raise
|
|
|
|
def test_empty_config_is_noop(self):
|
|
"""setup_telemetry with empty dict is a no-op (enabled defaults to False)."""
|
|
from agentkit.telemetry.setup import setup_telemetry
|
|
|
|
mock_app = MagicMock()
|
|
setup_telemetry(mock_app, {}) # Should not raise
|
|
|
|
|
|
# ── Integration: Tool safe_execute with telemetry ───────────────────
|
|
|
|
|
|
class TestToolTelemetryIntegration:
|
|
"""Test that Tool.safe_execute records telemetry."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_safe_execute_records_noop_telemetry(self):
|
|
"""safe_execute works with no-op telemetry (OTel not installed)."""
|
|
from agentkit.tools.base import Tool
|
|
|
|
class DummyTool(Tool):
|
|
async def execute(self, **kwargs):
|
|
return {"result": "ok"}
|
|
|
|
tool = DummyTool(name="test_tool", description="A test tool")
|
|
result = await tool.safe_execute(query="hello")
|
|
assert result == {"result": "ok"}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_safe_execute_error_records_telemetry(self):
|
|
"""safe_execute records error telemetry on exception."""
|
|
from agentkit.tools.base import Tool
|
|
|
|
class FailingTool(Tool):
|
|
async def execute(self, **kwargs):
|
|
raise ValueError("tool failed")
|
|
|
|
tool = FailingTool(name="failing_tool", description="A failing tool")
|
|
with pytest.raises(ValueError, match="tool failed"):
|
|
await tool.safe_execute(query="hello")
|
|
|
|
|
|
# ── start_span helper tests ─────────────────────────────────────────
|
|
|
|
|
|
class TestStartSpan:
|
|
"""Test start_span helper function."""
|
|
|
|
def test_start_span_noop_without_otel(self):
|
|
"""start_span returns no-op span context manager without OTel."""
|
|
from agentkit.telemetry.tracing import _OTEL_AVAILABLE, start_span, _NoOpSpan
|
|
|
|
if _OTEL_AVAILABLE:
|
|
pytest.skip("OTel is installed, skipping no-op test")
|
|
|
|
cm = start_span("test.span", attributes={"key": "value"})
|
|
assert isinstance(cm, _NoOpSpan)
|
|
# Should work as context manager
|
|
with cm:
|
|
pass # No error
|
|
|
|
def test_start_span_with_attributes(self):
|
|
"""start_span accepts attributes parameter without error."""
|
|
from agentkit.telemetry.tracing import start_span
|
|
|
|
cm = start_span("test.span", attributes={"key": "value", "count": 42})
|
|
with cm:
|
|
pass # No error regardless of OTel availability
|