Function test_metrics

tests/unittests/telemetry/test_functional.py:236–380

(monkeypatch)

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

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

  async def generate_random_number():
    return 42

  mock_model = MockModel.create(
      responses=[
          Part.from_function_call(name="get_current_time", args={}),
          Part.from_function_call(name="generate_random_number", args={}),
          Part.from_text(text="Both tools executed."),
      ],
      usage_metadata=types.GenerateContentResponseUsageMetadata(
          prompt_token_count=10,
          candidates_token_count=20,
          tool_use_prompt_token_count=5,
          thoughts_token_count=10,
          total_token_count=45,
      ),
  )
  test_agent = Agent(
      name="complex_agent",
      model=mock_model,
      tools=[
          FunctionTool(get_current_time),
          FunctionTool(generate_random_number),
      ],
  )

  runner = InMemoryRunner(root_agent=test_agent)
  await runner.run_async("Run both tools")

  metrics_data = reader.get_metrics_data()
  assert len(metrics_data.resource_metrics) > 0
  scope_metrics = metrics_data.resource_metrics[0].scope_metrics
  assert len(scope_metrics) > 0
  metrics_list = scope_metrics[0].metrics
  got_invocation = _extract_metrics(
      metrics_list, "gen_ai.agent.invocation.duration", "complex_agent"
  )
  assert len(got_invocation) == 1
  for p in got_invocation:
    p.value = None
  want_invocation = [
      MetricPoint(
          attributes={
              "gen_ai.agent.name": "complex_agent",
          },
          value=None,
      )
  ]
  assert got_invocation == want_invocation
  got_tool_exec = _extract_metrics(
      metrics_list, "gen_ai.tool.execution.duration", "complex_agent"
  )