code: def test_placement_group_scheduling_strategy(
      ray_start_cluster, connect_to_client
  ):
      cluster = ray_start_cluster
      cluster.add_node(num_cpus=8, resources={"head": 1})
      cluster.add_node(num_cpus=8, num_gpus=8, resources={"worker": 1})
      cluster.wait_for_nodes()
  
      ray.init(address=cluster.address)
      pg = ray.util.placement_group(
          bundles=[{"CPU": 1, "GPU": 1}, {"CPU": 1, "GPU": 1}]
      )
      ray.get(pg.ready())
  
      with connect_to_client_or_not(connect_to_client):
  
          @ray.remote(scheduling_strategy="DEFAULT")
          def get_node_id_1():
              return ray._private.worker.global_worker.current_node_id
  
          worker_node_id = ray.get(
              get_node_id_1.options(resources={"worker": 1}).remote()
          )
  
          assert (
              ray.get(
                  get_node_id_1.options(
                      num_cpus=1,
                      scheduling_strategy=PlacementGroupSchedulingStrategy(
                          placement_group=pg
                      ),
                  ).remote()
              )
              == worker_node_id
          )
  
          @ray.remote(
              num_cpus=1,
              scheduling_strategy=PlacementGroupSchedulingStrategy(
                  placement_group=pg
              ),
          )
          def get_node_id_2():
              return ray._private.worker.global_worker.current_node_id
  
          assert ray.get(get_node_id_2.remote()) == worker_node_id
  
          @ray.remote(
              num_cpus=1,
              scheduling_strategy=PlacementGroupSchedulingStrategy(
                  placement_group=pg
              ),
          )
          class Actor1:
              def get_node_id(self):
                  return (
                      ray._private.worker.global_worker.current_node_id
                  )
  
          actor1 = Actor1.remote()
          assert ray.get(actor1.get_node_id.remote()) == worker_node_id
  
          @ray.remote
          class Actor2:
              def get_node_id(self):
                  return (
                      ray._private.worker.global_worker.current_node_id
                  )
  
          actor2 = Actor2.options(
              scheduling_strategy=PlacementGroupSchedulingStrategy(
                  placement_group=pg
              )
          ).remote()
          assert ray.get(actor2.get_node_id.remote()) == worker_node_id
  
      with pytest.raises(ValueError):
  
          @ray.remote(
              scheduling_strategy=PlacementGroupSchedulingStrategy(
                  placement_group=pg
              )
          )
          def func():
              return 0
  
          func.options(placement_group=pg).remote()
  
      with pytest.raises(ValueError):
  
          @ray.remote
          def func():
              return 0
  
          func.options(scheduling_strategy="XXX").remote()
  
imports: import os
  import platform
  import sys
  import time
  import numpy
  import pytest
  import ray
  import ray._private.gcs_utils
  import ray.experimental.internal_kv
  from ray._private.test_utils import make_global_state_accessor, wait_for_condition, get_metric_check_condition, MetricSamplePattern
  from ray.util.client.ray_client_helpers import connect_to_client_or_not
  from ray.util.placement_group import placement_group
  from ray.util.scheduling_strategies import NodeAffinitySchedulingStrategy, PlacementGroupSchedulingStrategy
  from ray._private.test_utils import SignalActor
label_desc: Use image node_id 4 for calling a global function with example usage: test_placement_group_scheduling_strategy(ray_start_cluster, connect_to_client) and returns: ray, ray, int, int, ray, ray