code: def test_placement_group_task_resource_ids(
      ray_start_cluster, connect_to_client
  ):
      @ray.remote(num_cpus=1)
      def f():
          return ray.get_runtime_context().get_assigned_resources()
  
      cluster = ray_start_cluster
      num_nodes = 1
      for _ in range(num_nodes):
          cluster.add_node(num_cpus=4)
      ray.init(address=cluster.address)
  
      with connect_to_client_or_not(connect_to_client):
          g1 = ray.util.placement_group([{"CPU": 2}])
          o1 = f.options(
              scheduling_strategy=PlacementGroupSchedulingStrategy(
                  placement_group=g1
              )
          ).remote()
          resources = ray.get(o1)
          assert resources == {"CPU": 1}
  
          # Now retry with a bundle index constraint.
          o1 = f.options(
              scheduling_strategy=PlacementGroupSchedulingStrategy(
                  placement_group=g1, placement_group_bundle_index=0
              )
          ).remote()
          resources = ray.get(o1)
          assert resources == {"CPU": 1}
  
          placement_group_assert_no_leak([g1])
  
imports: import sys
  import warnings
  import os
  import pytest
  import ray
  from ray._private.utils import get_ray_doc_version
  import ray.cluster_utils
  from ray._private.test_utils import placement_group_assert_no_leak
  from ray._private.test_utils import skip_flaky_core_test_premerge
  from ray.util.client.ray_client_helpers import connect_to_client_or_not
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
label_desc: Use image node_id 9 for calling a global function with example usage: test_placement_group_task_resource_ids(ray_start_cluster, connect_to_client) and returns: ray