code: def test_load_report(shutdown_only, max_shapes):
      resource1 = "A"
      resource2 = "B"
      cluster = ray.init(
          num_cpus=1,
          resources={resource1: 1},
          _system_config={
              "max_resource_shapes_per_load_report": max_shapes,
          },
      )
  
      global_state_accessor = make_global_state_accessor(cluster)
  
      @ray.remote
      def sleep():
          time.sleep(1000)
  
      sleep.remote()
      for _ in range(3):
          sleep.remote()
          sleep.options(resources={resource1: 1}).remote()
          sleep.options(resources={resource2: 1}).remote()
  
      class Checker:
          def __init__(self):
              self.report = None
  
          def check_load_report(self):
              message = global_state_accessor.get_all_resource_usage()
              if message is None:
                  return False
  
              resource_usage = (
                  gcs_utils.ResourceUsageBatchData.FromString(message)
              )
              self.report = (
                  resource_usage.resource_load_by_shape.resource_demands
              )
              if max_shapes == 0:
                  return True
              elif max_shapes == 2:
                  return len(self.report) >= 2
              else:
                  return len(self.report) >= 3
  
      # Wait for load information to arrive.
      checker = Checker()
      wait_for_condition(checker.check_load_report)
  
      # Check that we respect the max shapes limit.
      if max_shapes != -1:
          assert len(checker.report) <= max_shapes
  
      print(checker.report)
  
      if max_shapes > 0:
          # Check that we differentiate between infeasible and ready tasks.
          for demand in checker.report:
              if resource2 in demand.shape:
                  assert demand.num_infeasible_requests_queued > 0
                  assert demand.num_ready_requests_queued == 0
              else:
                  assert demand.num_ready_requests_queued > 0
                  assert demand.num_infeasible_requests_queued == 0
      global_state_accessor.disconnect()
  
imports: import os
  import time
  import pytest
  import ray
  import ray._private.gcs_utils
  import ray._private.ray_constants
  from ray._raylet import GcsClient
  from ray.core.generated import autoscaler_pb2
  from ray._private.test_utils import convert_actor_state, make_global_state_accessor, wait_for_condition
label_desc: Use image node_id 10 for calling a global function with example usage: test_load_report(shutdown_only, max_shapes) and returns: False, True, unknown, unknown