Function test_run_async

tests/unittests/agents/test_loop_agent.py:105–139

(request: pytest.FixtureRequest, resumable: bool)

@pytest.mark.asyncio
@pytest.mark.parametrize('resumable', [True, False])
async def test_run_async(request: pytest.FixtureRequest, resumable: bool):
  agent = _TestingAgent(name=f'{request.function.__name__}_test_agent')
  loop_agent = LoopAgent(
      name=f'{request.function.__name__}_test_loop_agent',
      max_iterations=2,
      sub_agents=[
          agent,
      ],
  )
  parent_ctx = await _create_parent_invocation_context(
      request.function.__name__, loop_agent, resumable=resumable
  )
  events = [e async for e in loop_agent.run_async(parent_ctx)]

  simplified_events = testing_utils.simplify_resumable_app_events(events)
  if resumable:
    expected_events = [
        (
            loop_agent.name,
            {'current_sub_agent': agent.name, 'times_looped': 0},
        ),
        (agent.name, f'Hello, async {agent.name}!'),
        (
            loop_agent.name,
            {'current_sub_agent': agent.name, 'times_looped': 1},
        ),
        (agent.name, f'Hello, async {agent.name}!'),
        (loop_agent.name, END_OF_AGENT),
    ]
  else:
    expected_events = [
        (agent.name, f'Hello, async {agent.name}!'),
        (agent.name, f'Hello, async {agent.name}!'),
    ]
  assert simplified_events == expected_events


@pytest.mark.asyncio