code: def test_demand_report_for_node_affinity_scheduling_strategy(
      monkeypatch, shutdown_only
  ):
      from ray.cluster_utils import AutoscalingCluster
  
      cluster = AutoscalingCluster(
          head_resources={"CPU": 0},
          worker_node_types={
              "cpu_node": {
                  "resources": {
                      "CPU": 1,
                      "object_store_memory": 1024 * 1024 * 1024,
                  },
                  "node_config": {},
                  "min_workers": 1,
                  "max_workers": 1,
              },
          },
      )
  
      cluster.start()
      info = ray.init(address="auto")
  
      @ray.remote(num_cpus=1)
      def f(sleep_s):
          time.sleep(sleep_s)
          return ray.get_runtime_context().get_node_id()
  
      worker_node_id = ray.get(f.remote(0))
  
      tasks = []
      tasks.append(f.remote(10000))
      # This is not reported since there is feasible node.
      tasks.append(
          f.options(
              scheduling_strategy=NodeAffinitySchedulingStrategy(
                  worker_node_id, soft=False
              )
          ).remote(0)
      )
      # This is reported since there is no feasible node and soft is True.
      tasks.append(
          f.options(
              num_gpus=1,
              scheduling_strategy=NodeAffinitySchedulingStrategy(
                  ray.NodeID.from_random().hex(), soft=True
              ),
          ).remote(0)
      )
  
      global_state_accessor = make_global_state_accessor(info)
  
      def check_resource_demand():
          message = global_state_accessor.get_all_resource_usage()
          if message is None:
              return False
  
          resource_usage = gcs_utils.ResourceUsageBatchData.FromString(
              message
          )
          aggregate_resource_load = (
              resource_usage.resource_load_by_shape.resource_demands
          )
  
          if len(aggregate_resource_load) != 1:
              return False
  
          if (
              aggregate_resource_load[0].num_infeasible_requests_queued
              != 1
          ):
              return False
  
          if aggregate_resource_load[0].shape != {
              "CPU": 1.0,
              "GPU": 1.0,
          }:
              return False
  
          return True
  
      wait_for_condition(check_resource_demand, 20)
      cluster.shutdown()
  
imports: import os
  import platform
  import sys
  import time
  import numpy
  import pytest
  import ray
  import ray._private.gcs_utils
  import ray.experimental.internal_kv
  from ray._private.test_utils import make_global_state_accessor, wait_for_condition, get_metric_check_condition, MetricSamplePattern
  from ray.util.client.ray_client_helpers import connect_to_client_or_not
  from ray.util.placement_group import placement_group
  from ray.util.scheduling_strategies import NodeAffinitySchedulingStrategy, PlacementGroupSchedulingStrategy
  from ray._private.test_utils import SignalActor
label_desc: Use image node_id 9 for calling a global function with example usage: test_demand_report_for_node_affinity_scheduling_strategy(monkeypatch, shutdown_only) and returns: ray, True, False, False, False, False