Function run_node_scenario

tests/unittests/telemetry/functional_test_helpers.py:408–436

Runs the workflow scenario to completion, draining the event stream. If ``event_sink`` is provided, collected events are appended to it as they are drained. This lets callers inspect the events that were emitted before an exception propagates (e.g. when ``failing=True``).

(
    *, failing: bool = False, event_sink: list[Event] | None = None
)

async def run_node_scenario(
    *, failing: bool = False, event_sink: list[Event] | None = None
) -> list[Event]:
  """Runs the workflow scenario to completion, draining the event stream.

  If ``event_sink`` is provided, collected events are appended to it as they
  are drained. This lets callers inspect the events that were emitted before
  an exception propagates (e.g. when ``failing=True``).
  """
  workflow = build_test_workflow(failing=failing)
  runner = InMemoryRunner(app_name=NODE_APP_NAME, node=workflow)
  session = await runner.session_service.create_session(
      app_name=NODE_APP_NAME, user_id=NODE_USER_ID
  )
  content = Content(parts=[Part.from_text(text=USER_PROMPT)], role="user")

  collected_events: list[Event] = event_sink if event_sink is not None else []

  async with aclosing(
      runner.run_async(
          user_id=NODE_USER_ID,
          session_id=session.id,
          new_message=content,
      )
  ) as agen:
    async for event in agen:
      collected_events.append(event)

  return collected_events


async def run_agent_scenario(runner: TestInMemoryRunner) -> None: