code: def test_multi_node(start_cluster, option: str, source: str):
      """Tests that the working_dir is propagated across multi-node clusters."""
      NUM_NODES = 3
      cluster, address = start_cluster
      for i in range(NUM_NODES - 1):  # Head node already added.
          cluster.add_node(
              num_cpus=1,
              runtime_env_dir_name=f"node_{i}_runtime_resources",
          )
  
      if option == "working_dir":
          if source.endswith(".whl"):
              pytest.skip(".whl working dir is not supported")
          ray.init(address, runtime_env={"working_dir": source})
      elif option == "py_modules":
          if source not in REMOTE_URIS and source != S3_WHL_PACKAGE_URI:
              source = str(Path(source) / "test_module")
          ray.init(address, runtime_env={"py_modules": [source]})
  
      @ray.remote(num_cpus=1)
      class A:
          def check_and_get_node_id(self):
              import test_module
  
              test_module.one()
              return ray.get_runtime_context().get_node_id()
  
      num_cpus = int(ray.available_resources()["CPU"])
      actors = [A.remote() for _ in range(num_cpus)]
      object_refs = [a.check_and_get_node_id.remote() for a in actors]
      assert len(set(ray.get(object_refs))) == NUM_NODES
  
file_path: ray/python/ray/tests/test_runtime_env_working_dir_remote_uri.py
imports: from pathlib import Path
  import sys
  import pytest
  from pytest_lazyfixture import lazy_fixture
  import ray
label_desc: Use image node_id 2 for calling a global function with example usage: test_multi_node(start_cluster, option, source) and returns: ray