code: def test_wait_cluster(ray_start_cluster_enabled):
      cluster = ray_start_cluster_enabled
      cluster.add_node(num_cpus=1, resources={"RemoteResource": 1})
      cluster.add_node(num_cpus=1, resources={"RemoteResource": 1})
      ray.init(address=cluster.address)
  
      @ray.remote(resources={"RemoteResource": 1})
      def f():
          return
  
      # Make sure we have enough workers on the remote nodes to execute some
      # tasks.
      tasks = [f.remote() for _ in range(10)]
      start = time.time()
      ray.get(tasks)
      end = time.time()
  
      # Submit some more tasks that can only be executed on the remote nodes.
      tasks = [f.remote() for _ in range(10)]
      # Sleep for a bit to let the tasks finish.
      time.sleep((end - start) * 2)
      _, unready = ray.wait(tasks, num_returns=len(tasks), timeout=0)
      # All remote tasks should have finished.
      assert len(unready) == 0
  
imports: import json
  import logging
  import os
  import sys
  import time
  import numpy
  import pytest
  import ray._private.profiling
  import ray.cluster_utils
  from ray._private.internal_api import memory_summary
  from ray._private.test_utils import client_test_enabled, wait_for_condition
  from ray.exceptions import ObjectFreedError
label_desc: Use image node_id 6 for calling a global function with example usage: test_wait_cluster(ray_start_cluster_enabled) without return types