code: def test_pull_manager_stats(shutdown_only):
      info = ray.init(
          num_cpus=2, object_store_memory=100_000_000, **METRIC_CONFIG
      )
  
      driver = """
  import ray
  import time
  import numpy as np
  
  ray.init("auto")
  
  # Spill a lot of 10MiB objects. The object store is 100MiB, so pull manager will
  # only be able to pull ~9 total into memory at once, including running tasks.
  buf = []
  for _ in range(100):
      buf.append(ray.put(np.ones(10 * 1024 * 1024, dtype=np.uint8)))
  
  @ray.remote
  def f(x):
      time.sleep(999)
  
  ray.get([f.remote(x) for x in buf])"""
  
      proc = run_string_as_driver_nonblocking(driver)
  
      # This test is non-deterministic since pull bundles can sometimes end up fallback
      # allocated. This leads to slightly more objects pulled than you'd expect.
      def close_to_expected(stats):
          assert len(stats) == 3, stats
          assert stats["RUNNING"] == 2, stats
          assert 7 <= stats["PENDING_NODE_ASSIGNMENT"] <= 17, stats
          assert 81 <= stats["PENDING_OBJ_STORE_MEM_AVAIL"] <= 91, stats
          assert sum(stats.values()) == 100, stats
          return True
  
      wait_for_condition(
          lambda: close_to_expected(tasks_by_state(info)),
          timeout=20,
          retry_interval_ms=500,
      )
      proc.kill()
  
imports: from collections import defaultdict
  import sys
  import os
  import copy
  import pytest
  import ray
  from ray._private.metrics_agent import RAY_WORKER_TIMEOUT_S
  from ray._private.test_utils import raw_metrics, run_string_as_driver, run_string_as_driver_nonblocking, wait_for_condition
label_desc: Use image node_id 17 for calling a global function with example usage: test_pull_manager_stats(shutdown_only) and returns: True