Function _run_node

tests/unittests/runners/test_runner_node.py:59–70

Run a BaseNode via Runner(node=...) and return (events, ss, session).

(node, message='hello')

async def _run_node(node, message='hello'):
  """Run a BaseNode via Runner(node=...) and return (events, ss, session)."""
  ss = InMemorySessionService()
  runner = Runner(app_name='test', node=node, session_service=ss)
  session = await ss.create_session(app_name='test', user_id='u')
  msg = types.Content(parts=[types.Part(text=message)], role='user')
  events = []
  async for event in runner.run_async(
      user_id='u', session_id=session.id, new_message=msg
  ):
    events.append(event)
  return events, ss, session


def _make_interrupt_event(fc_name='get_input', fc_id='fc-1'):