Function test_metrics_tool_error

tests/unittests/telemetry/test_functional.py:384–441

(monkeypatch)

@pytest.mark.asyncio
async def test_metrics_tool_error(monkeypatch):
  reader = _setup_test_metrics(monkeypatch)

  async def get_current_time():
    return "2026-04-15T14:26:03Z"

  async def failing_tool():
    raise ValueError("Tool failed")

  mock_model = MockModel.create(
      responses=[
          Part.from_function_call(name="get_current_time", args={}),
          Part.from_function_call(name="failing_tool", args={}),
          Part.from_text(text="Should not reach here"),
      ]
  )
  test_agent = Agent(
      name="error_agent",
      model=mock_model,
      tools=[FunctionTool(get_current_time), FunctionTool(failing_tool)],
  )

  runner = InMemoryRunner(root_agent=test_agent)
  with pytest.raises(ValueError, match="Tool failed"):
    await runner.run_async("Run tools")

  metrics_data = reader.get_metrics_data()
  metrics_list = metrics_data.resource_metrics[0].scope_metrics[0].metrics

  # Verify Tool Execution Duration
  got = _extract_metrics(
      metrics_list, "gen_ai.tool.execution.duration", "error_agent"
  )
  assert len(got) == 2
  for p in got:
    p.value = None

  want = [
      MetricPoint(
          attributes={
              "gen_ai.agent.name": "error_agent",
              "gen_ai.tool.name": "failing_tool",
              "error.type": "ValueError",
          },
          value=None,
      ),
      MetricPoint(
          attributes={
              "gen_ai.agent.name": "error_agent",
              "gen_ai.tool.name": "get_current_time",
          },
          value=None,
      ),
  ]

  got.sort(key=lambda p: str(p.attributes.get("gen_ai.tool.name", "")))
  want.sort(key=lambda p: str(p.attributes.get("gen_ai.tool.name", "")))
  assert got == want