code: def test_task_nested(shutdown_only):
      info = ray.init(num_cpus=2, **METRIC_CONFIG)
  
      driver = """
  import ray
  import time
  
  ray.init("auto")
  
  @ray.remote(num_cpus=0)
  def wrapper():
      @ray.remote
      def f():
          time.sleep(999)
  
      ray.get([f.remote() for _ in range(10)])
  
  w = wrapper.remote()
  ray.get(w)
  """
      proc = run_string_as_driver_nonblocking(driver)
  
      expected = {
          "RUNNING": 2.0,
          "RUNNING_IN_RAY_GET": 1.0,
          "PENDING_NODE_ASSIGNMENT": 8.0,
      }
      wait_for_condition(
          lambda: tasks_by_state(info) == expected,
          timeout=20,
          retry_interval_ms=2000,
      )
      assert tasks_by_name_and_state(info) == {
          ("wrapper", "RUNNING_IN_RAY_GET"): 1.0,
          ("f", "RUNNING"): 2.0,
          ("f", "PENDING_NODE_ASSIGNMENT"): 8.0,
      }
      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 7 for calling a global function with example usage: test_task_nested(shutdown_only) without return types