Function test_agent_replay

src/google/adk/cli/agent_test_runner.py:453–724

(agent_dir, test_file, monkeypatch)

    get_test_files(),
)
def test_agent_replay(agent_dir, test_file, monkeypatch):
  # Add agent_dir.parent to sys.path so relative imports work
  import sys

  sys_path_saved = list(sys.path)
  sys.path.insert(0, str(agent_dir.parent))

  try:
    import random

    random.seed(42)

    loader = AgentLoader(str(agent_dir.parent))
    loader.remove_agent_from_cache(agent_dir.name)
    agent_or_app = loader.load_agent(agent_dir.name)

    root_agent = (
        agent_or_app.root_agent
        if isinstance(agent_or_app, App)
        else agent_or_app
    )
    _make_nodes_sequential(root_agent)
    agent_names = _get_all_agent_names(root_agent)

    import inspect

    # Dynamically locate the loaded agent module from sys.modules
    mod = sys.modules.get(f"{agent_dir.name}.agent") or sys.modules.get(
        agent_dir.name
    )
    if not mod:
      # Fallback for namespace packages or nested imports
      for k, v in sys.modules.items():
        if k.endswith(f"{agent_dir.name}.agent") or k.endswith(agent_dir.name):
          mod = v
          break

    # Reflectively find all Agent instances defined in the module (e.g. dynamic agents)
    if mod:
      for _, obj in inspect.getmembers(mod):
        if isinstance(obj, BaseAgent) and hasattr(obj, "name"):
          agent_names.add(obj.name)

    with open(test_file, "r") as f:
      session_data = json.load(f)

    events_data = session_data.get("events", [])
    if not events_data:
      pytest.skip(f"No events in {test_file}")

    first_event = events_data[0]
    user_message = ""
    if first_event.get("author") == "user":
      parts = first_event.get("content", {}).get("parts", [])
      if parts and "text" in parts[0]:
        user_message = parts[0]["text"]

    if not user_message: