Function test_resume_async

tests/unittests/agents/test_loop_agent.py:143–168

(request: pytest.FixtureRequest)

@pytest.mark.asyncio
async def test_resume_async(request: pytest.FixtureRequest):
  agent_1 = _TestingAgent(name=f'{request.function.__name__}_test_agent_1')
  agent_2 = _TestingAgent(name=f'{request.function.__name__}_test_agent_2')
  loop_agent = LoopAgent(
      name=f'{request.function.__name__}_test_loop_agent',
      max_iterations=2,
      sub_agents=[
          agent_1,
          agent_2,
      ],
  )
  parent_ctx = await _create_parent_invocation_context(
      request.function.__name__, loop_agent, resumable=True
  )
  parent_ctx.agent_states[loop_agent.name] = LoopAgentState(
      current_sub_agent=agent_2.name, times_looped=1
  ).model_dump(mode='json')

  events = [e async for e in loop_agent.run_async(parent_ctx)]

  simplified_events = testing_utils.simplify_resumable_app_events(events)
  expected_events = [
      (agent_2.name, f'Hello, async {agent_2.name}!'),
      (loop_agent.name, END_OF_AGENT),
  ]
  assert simplified_events == expected_events


@pytest.mark.asyncio