code: def test_worker_lease_reply_with_resources(ray_start_cluster_enabled):
      cluster = ray_start_cluster_enabled
      cluster.add_node(
          memory=2000 * 1024**2,
          num_cpus=1,
          _system_config={
              "gcs_resource_report_poll_period_ms": 1000000,
              "gcs_actor_scheduling_enabled": True,
          },
      )
      node2 = cluster.add_node(memory=1000 * 1024**2, num_cpus=1)
      ray.init(address=cluster.address)
      cluster.wait_for_nodes()
  
      @ray.remote(memory=1500 * 1024**2, num_cpus=0.01)
      def fun(signal):
          signal.send.remote()
          time.sleep(30)
          return 0
  
      signal = SignalActor.remote()
      fun.remote(signal)
      # Make sure that the `fun` is running.
      ray.get(signal.wait.remote())
  
      @ray.remote(memory=800 * 1024**2, num_cpus=0.01)
      class Foo:
          def method(self):
              return ray._private.worker.global_worker.node.unique_id
  
      foo1 = Foo.remote()
      o1 = foo1.method.remote()
      ready_list, remaining_list = ray.wait([o1], timeout=10)
      # If RequestWorkerLeaseReply carries normal task resources,
      # GCS will then schedule foo1 to node2. Otherwise,
      # GCS would keep trying to schedule foo1 to
      # node1 and getting rejected.
      assert len(ready_list) == 1 and len(remaining_list) == 0
      assert ray.get(o1) == node2.unique_id
  
file_path: ray/python/ray/tests/test_advanced_5.py
imports: import logging
  import sys
  import time
  import numpy
  import pytest
  import ray.cluster_utils
  from ray._private.test_utils import SignalActor, client_test_enabled
label_desc: Use image node_id 5 for calling a global function with example usage: test_worker_lease_reply_with_resources(ray_start_cluster_enabled) and returns: int, ray