code: def test_scheduled_task_no_pending_demand(mode):
      # So that head node will need to dispatch tasks to worker node.
      num_head_cpu = 0 if mode == "multi_node" else 1
  
      cluster = AutoscalingCluster(
          head_resources={"CPU": num_head_cpu},
          worker_node_types={
              "type-1": {
                  "resources": {"CPU": 1},
                  "node_config": {},
                  "min_workers": 0,
                  "max_workers": 1,
              },
          },
      )
  
      driver_script = """
  import time
  import ray
  @ray.remote(num_cpus=1)
  def foo():
    return True
  
  ray.init("auto")
  
  while True:
      assert(ray.get(foo.remote()))
  """
  
      try:
          cluster.start()
          ray.init("auto")
          gcs_address = ray.get_runtime_context().gcs_address
  
          run_string_as_driver_nonblocking(driver_script)
  
          def tasks_run():
              tasks = list_tasks()
  
              # Waiting til the driver in the run_string_as_driver_nonblocking is running
              assert len(tasks) > 0
  
              return True
  
          wait_for_condition(tasks_run)
  
          for _ in range(30):
              # verify no pending task + with resource used.
              status = get_cluster_status(gcs_address)
              has_task_demand = (
                  len(status.resource_demands.ray_task_actor_demand) > 0
              )
              has_task_usage = False
  
              for usage in status.cluster_resource_usage:
                  if usage.resource_name == "CPU":
                      has_task_usage = usage.used > 0
              print(status.cluster_resource_usage)
              print(status.resource_demands.ray_task_actor_demand)
              assert not (has_task_demand and has_task_usage), status
              time.sleep(0.1)
      finally:
          # TODO(rickyx): refactor into a fixture for autoscaling cluster.
          ray.shutdown()
          cluster.shutdown()
  
file_path: ray/python/ray/autoscaler/v2/tests/test_e2e.py
imports: import os
  import sys
  import time
  from typing import Dict
  import pytest
  import ray
  from ray._private.resource_spec import HEAD_NODE_RESOURCE_NAME
  from ray._private.test_utils import run_string_as_driver_nonblocking, wait_for_condition
  from ray.autoscaler.v2.sdk import get_cluster_status
  from ray.cluster_utils import AutoscalingCluster
  from ray.util.placement_group import placement_group, remove_placement_group
  from ray.util.state.api import list_placement_groups, list_tasks
label_desc: Use image node_id 3 for calling a global function with example usage: test_scheduled_task_no_pending_demand(mode) and returns: True