def test_to_cloud_run_happy_path(
    monkeypatch: pytest.MonkeyPatch,
    agent_dir: Callable[[bool], Path],
    include_requirements: bool,
) -> None:
  """
  End-to-end execution test for `to_cloud_run` covering both presence and
  absence of *requirements.txt*.
  """
  tmp_dir = Path(tempfile.mkdtemp())
  src_dir = agent_dir(include_requirements)

  copy_recorder = _Recorder()
  run_recorder = _Recorder()

  # Cache the ORIGINAL copytree before patching
  original_copytree = cli_deploy.shutil.copytree

  def _recording_copytree(*args: Any, **kwargs: Any):
    copy_recorder(*args, **kwargs)
    return original_copytree(*args, **kwargs)

  monkeypatch.setattr(cli_deploy.shutil, "copytree", _recording_copytree)
  # Skip actual cleanup so that we can inspect generated files later.
  monkeypatch.setattr(cli_deploy.shutil, "rmtree", lambda *_a, **_k: None)
  monkeypatch.setattr(subprocess, "run", run_recorder)

  cli_deploy.to_cloud_run(
      agent_folder=str(src_dir),
      project="proj",
      region="asia-northeast1",
      service_name="svc",
      app_name="app",
      temp_folder=str(tmp_dir),
      port=8080,
      trace_to_cloud=True,
      with_ui=True,
      verbosity="info",
      session_service_uri="sqlite://",
      artifact_service_uri="gs://bucket",
      memory_service_uri="rag://",
      adk_version="0.0.5",
  )

  # Assertions
  assert (
      len(copy_recorder.calls) == 1
  ), "Agent sources must be copied exactly once."
  assert run_recorder.calls, "gcloud command should be executed at least once."
  assert (tmp_dir / "Dockerfile").exists(), "Dockerfile must be generated."

  # Manual cleanup because we disabled rmtree in the monkeypatch.
  shutil.rmtree(tmp_dir, ignore_errors=True)

  End-to-end execution test for `to_cloud_run` covering both presence and
  absence of *requirements.txt*.
  
tests/unittests/cli/utils/test_cli_deploy.py