code: def test_e2e_complex(call_ray_start, tmp_path):
      """Test multiple runtime_env options across multiple client connections.
  
      1.  Run a Ray Client job with both working_dir and pip specified. Check the
          environment using imports and file reads in tasks and actors.
      2.  On the same cluster, run a job as above but using the Ray Summit
          2021 demo's pip requirements.txt.  Also, check that per-task and
          per-actor pip requirements work, all using the job's working_dir.
      """
      # Create a file to use to test working_dir
      specific_path = tmp_path / "test"
      specific_path.write_text("Hello")
  
      with ray.client(f"localhost:{CLIENT_SERVER_PORT}").env(
          {"working_dir": str(tmp_path), "pip": ["pip-install-test"]}
      ).connect():
          # Test that a task is started in the working_dir.
          @ray.remote
          def test_read():
              return Path("./test").read_text()
  
          assert ray.get(test_read.remote()) == "Hello"
  
          # Check a task has the job's pip requirements and working_dir.
          @ray.remote
          def test_pip():
              import pip_install_test  # noqa
              import ray  # noqa
  
              return Path("./test").read_text()
  
          assert ray.get(test_pip.remote()) == "Hello"
  
          # Check an actor has the job's pip requirements and working_dir.
          @ray.remote
          class TestActor:
              def test(self):
                  import pip_install_test  # noqa
  
                  return Path("./test").read_text()
  
          a = TestActor.remote()
          assert ray.get(a.test.remote()) == "Hello"
  
      # pip requirements file from Ray Summit 2021 demo; updated to be compatible with
      # recent python versions
      requirement_path = tmp_path / "requirements.txt"
      requirement_path.write_text(
          "\n".join(
              [
                  "ray[serve, tune]",
                  "PyGithub",
                  "xgboost_ray",  # has Ray as a dependency
                  "pandas==1.5.3",
                  "typer",
                  "aiofiles",
              ]
          )
      )
  
      # Start a new job on the same cluster using the Summit 2021 requirements.
      with ray.client(f"localhost:{CLIENT_SERVER_PORT}").env(
          {"working_dir": str(tmp_path), "pip": str(requirement_path)}
      ).connect():
  
          @ray.remote
          def test_read():
              return Path("./test").read_text()
  
          assert ray.get(test_read.remote()) == "Hello"
  
          # Check that a task has the job's pip requirements and working_dir.
          @ray.remote
          def test_import():
              import ray  # noqa
              from ray import serve  # noqa
              from ray import tune  # noqa
              import typer  # noqa
              import xgboost_ray  # noqa
  
              return Path("./test").read_text()
  
          assert ray.get(test_import.remote()) == "Hello"
  
          # Check that an actor has the job's pip requirements and working_dir.
          @ray.remote
          class TestActor:
              def test(self):
                  import ray  # noqa
                  from ray import serve  # noqa
                  from ray import tune  # noqa
                  import typer  # noqa
                  import xgboost_ray  # noqa
  
                  return Path("./test").read_text()
  
          a = TestActor.options(
              runtime_env={"pip": str(requirement_path)}
          ).remote()
          assert ray.get(a.test.remote()) == "Hello"
  
          # Check that per-task pip specification works and that the job's
          # working_dir is still inherited.
          @ray.remote
          def test_pip():
              import pip_install_test  # noqa
  
              return Path("./test").read_text()
  
          assert (
              ray.get(
                  test_pip.options(
                      runtime_env={"pip": ["pip-install-test"]}
                  ).remote()
              )
              == "Hello"
          )
  
          # Check that pip_install_test is not in the job's pip requirements.
          with pytest.raises(ray.exceptions.RayTaskError) as excinfo:
              ray.get(test_pip.remote())
          assert "ModuleNotFoundError" in str(excinfo.value)
  
          # Check that per-actor pip specification works and that the job's
          # working_dir is still inherited.
          @ray.remote
          class TestActor:
              def test(self):
                  import pip_install_test  # noqa
  
                  return Path("./test").read_text()
  
          a = TestActor.options(
              runtime_env={"pip": ["pip-install-test"]}
          ).remote()
          assert ray.get(a.test.remote()) == "Hello"
  
imports: import os
  import platform
  from pathlib import Path
  import pytest
  import subprocess
  import sys
  import tempfile
  import time
  from typing import List
  from unittest import mock
  import yaml
  import ray
  from ray.runtime_env import RuntimeEnv
  from ray._private.runtime_env.conda import inject_dependencies, _inject_ray_to_conda_site, _resolve_install_from_source_ray_dependencies, _current_py_version
  from ray._private.runtime_env.conda_utils import get_conda_env_list
  from ray._private.test_utils import run_string_as_driver, run_string_as_driver_nonblocking, wait_for_condition, chdir
  from ray._private.utils import get_conda_env_dir, get_conda_bin_executable, try_to_create_directory
label_desc: Use image node_id 23 for calling a global function with example usage: test_e2e_complex(call_ray_start, tmp_path) and returns: Path, Path, Path, Path, Path, Path, Path, Path