code: def test_placement_group_wait(ray_start_cluster, connect_to_client):
      cluster = ray_start_cluster
      [cluster.add_node(num_cpus=2) for _ in range(2)]
      ray.init(address=cluster.address)
      cluster.wait_for_nodes()
  
      with connect_to_client_or_not(connect_to_client):
          # Wait on placement group that cannot be created.
          placement_group = ray.util.placement_group(
              name="name",
              strategy="SPREAD",
              bundles=[
                  {"CPU": 2},
                  {"CPU": 2},
              ],
          )
          ready, unready = ray.wait([placement_group.ready()])
          assert len(unready) == 0
          assert len(ready) == 1
          table = ray.util.placement_group_table(placement_group)
          assert table["state"] == "CREATED"
          pg = ray.get(placement_group.ready())
          assert pg.bundle_specs == placement_group.bundle_specs
          assert pg.id.binary() == placement_group.id.binary()
  
          @ray.remote
          def get_node_id():
              return ray.get_runtime_context().get_node_id()
  
          for i in range(2):
              scheduling_strategy = PlacementGroupSchedulingStrategy(
                  placement_group=placement_group,
                  placement_group_bundle_index=i,
              )
              node_id = ray.get(
                  get_node_id.options(
                      scheduling_strategy=scheduling_strategy
                  ).remote()
              )
              assert node_id == table["bundles_to_node_id"][i]
  
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 3 for calling a global function with example usage: test_placement_group_wait(ray_start_cluster, connect_to_client) and returns: ray