code: def test_pull_manager_at_capacity_reports(ray_start_cluster):
      cluster = ray_start_cluster
      cluster.add_node(num_cpus=0, object_store_memory=int(1e8))
      ray.init(address=cluster.address)
      cluster.add_node(num_cpus=1, object_store_memory=int(1e8))
  
      object_size = int(1e7)
      refs = []
      for _ in range(20):
          refs.append(ray.put(np.zeros(object_size, dtype=np.uint8)))
  
      def fetches_queued():
          return "fetches queued" in memory_summary(stats_only=True)
  
      assert not fetches_queued()
  
      @ray.remote
      def f(s, ref):
          ray.get(s.wait.remote())
  
      signal = SignalActor.remote()
      xs = [f.remote(signal, ref) for ref in refs]
      wait_for_condition(fetches_queued)
  
      signal.send.remote()
      ray.get(xs)
      wait_for_condition(lambda: not fetches_queued())
  
imports: import collections
  import logging
  import platform
  import subprocess
  import sys
  import time
  import unittest
  import numpy
  import pytest
  import ray
  import ray.cluster_utils
  import ray.util.accelerators
  from ray._private.internal_api import memory_summary
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
  from ray._private.test_utils import Semaphore, SignalActor, object_memory_usage, get_metric_check_condition, wait_for_condition, MetricSamplePattern
label_desc: Use image node_id 14 for calling a global function with example usage: test_pull_manager_at_capacity_reports(ray_start_cluster) and returns: unknown