code: def test_idle_node_blocked(ray_start_cluster):
      cluster = ray_start_cluster
      cluster.add_node(num_cpus=1)
      ray.init(address=cluster.address)
  
      stub = _autoscaler_state_service_stub()
  
      # We don't have node id from `add_node` unfortunately.
      def nodes_up():
          nodes = list_nodes()
          assert len(nodes) == 1
          return True
  
      wait_for_condition(nodes_up)
  
      head_node_id = get_node_ids()
  
      def verify_cluster_idle():
          state = get_cluster_resource_state(stub)
          assert_node_states(
              state,
              [
                  ExpectedNodeState(
                      head_node_id,
                      NodeStatus.IDLE,
                      lambda idle_ms: idle_ms > 0,
                  ),
              ],
          )
          return True
  
      wait_for_condition(verify_cluster_idle)
  
      # Unschedulable
      @ray.remote(num_cpus=10000)
      def f():
          pass
  
      # Schedule a task running
      @ray.remote(num_cpus=1)
      def g():
          ray.get(f.remote())
  
      t = g.remote()
  
      def verify_cluster_busy():
          state = get_cluster_resource_state(stub)
          assert_node_states(
              state,
              [
                  ExpectedNodeState(
                      head_node_id,
                      NodeStatus.RUNNING,
                      lambda idle_ms: idle_ms == 0,
                  ),
              ],
          )
          return True
  
      wait_for_condition(verify_cluster_busy)
      for _ in range(10):
          time.sleep(0.5)
          verify_cluster_busy()
  
      # Kill the task
      ray.cancel(t, force=True)
      wait_for_condition(verify_cluster_idle)
  
imports: import os
  import sys
  import time
  from dataclasses import dataclass
  from typing import Callable, List, Optional, Tuple
  import pytest
  import ray
  import ray._private.ray_constants
  from ray._private.test_utils import wait_for_condition
  from ray.autoscaler.v2.schema import ClusterStatus, LaunchRequest, NodeInfo, ResourceRequestByCount
  from ray.autoscaler.v2.sdk import get_cluster_status, request_cluster_resources
  from ray.autoscaler.v2.tests.util import get_available_resources, get_cluster_resource_state, get_total_resources, report_autoscaling_state
  from ray.core.generated import autoscaler_pb2, autoscaler_pb2_grpc
  from ray.core.generated.autoscaler_pb2 import ClusterResourceState, NodeStatus
  from ray.util.state.api import list_nodes
label_desc: Use image node_id 13 for calling a global function with example usage: test_idle_node_blocked(ray_start_cluster) and returns: True, True, True