code: def test_capture_child_tasks(ray_start_cluster, connect_to_client):
      cluster = ray_start_cluster
      total_num_tasks = 4
      for _ in range(2):
          cluster.add_node(
              num_cpus=total_num_tasks, num_gpus=total_num_tasks
          )
      ray.init(address=cluster.address)
  
      with connect_to_client_or_not(connect_to_client):
          pg = ray.util.placement_group(
              [
                  {
                      "CPU": 2,
                      "GPU": 2,
                  },
                  {
                      "CPU": 2,
                      "GPU": 2,
                  },
              ],
              strategy="STRICT_PACK",
          )
          ray.get(pg.ready())
  
          # If get_current_placement_group is used when the current worker/driver
          # doesn't belong to any of placement group, it should return None.
          assert get_current_placement_group() is None
  
          # Test if tasks capture child tasks.
          @ray.remote
          def task():
              return get_current_placement_group()
  
          @ray.remote
          def create_nested_task(child_cpu, child_gpu, set_none=False):
              assert get_current_placement_group() is not None
              kwargs = {
                  "num_cpus": child_cpu,
                  "num_gpus": child_gpu,
              }
              if set_none:
                  kwargs["placement_group"] = None
              return ray.get(
                  [task.options(**kwargs).remote() for _ in range(3)]
              )
  
          t = create_nested_task.options(
              num_cpus=1,
              num_gpus=0,
              scheduling_strategy=PlacementGroupSchedulingStrategy(
                  placement_group=pg,
                  placement_group_capture_child_tasks=True,
              ),
          ).remote(1, 0)
          pgs = ray.get(t)
          # Every task should have current placement group because they
          # should be implicitly captured by default.
          assert None not in pgs
  
          t1 = create_nested_task.options(
              num_cpus=1,
              num_gpus=0,
              scheduling_strategy=PlacementGroupSchedulingStrategy(
                  placement_group=pg,
                  placement_group_capture_child_tasks=True,
              ),
          ).remote(1, 0, True)
          pgs = ray.get(t1)
          # Every task should have no placement group since it's set to None.
          # should be implicitly captured by default.
          assert set(pgs) == {None}
  
          # Test if tasks don't capture child tasks when the option is off.
          t2 = create_nested_task.options(
              num_cpus=0,
              num_gpus=1,
              scheduling_strategy=PlacementGroupSchedulingStrategy(
                  placement_group=pg
              ),
          ).remote(0, 1)
          pgs = ray.get(t2)
          # All placement groups should be None since we don't capture child
          # tasks.
          assert not all(pgs)
  
imports: import sys
  import time
  import pytest
  import ray
  import ray._private.gcs_utils
  import ray.cluster_utils
  from ray._private.test_utils import convert_actor_state, generate_system_config_map, get_error_message, get_other_nodes, kill_actor_and_wait_for_failure, placement_group_assert_no_leak, run_string_as_driver, wait_for_condition
  from ray.util.client.ray_client_helpers import connect_to_client_or_not
  from ray.util.placement_group import get_current_placement_group
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
label_desc: Use image node_id 8 for calling a global function with example usage: test_capture_child_tasks(ray_start_cluster, connect_to_client) and returns: get_current_placement_group, ray