code: def test_named_placement_group(ray_start_cluster):
      cluster = ray_start_cluster
      for _ in range(2):
          cluster.add_node(num_cpus=3)
      cluster.wait_for_nodes()
      info = ray.init(
          address=cluster.address, namespace="default_test_namespace"
      )
      global_placement_group_name = "named_placement_group"
  
      # Create a detached placement group with name.
      driver_code = f"""
  import ray
  
  ray.init(address="{info["address"]}", namespace="default_test_namespace")
  
  pg = ray.util.placement_group(
          [{{"CPU": 1}} for _ in range(2)],
          strategy="STRICT_SPREAD",
          name="{global_placement_group_name}",
          lifetime="detached")
  ray.get(pg.ready())
  
  ray.shutdown()
      """
  
      run_string_as_driver(driver_code)
  
      # Wait until the driver is reported as dead by GCS.
      def is_job_done():
          jobs = ray._private.state.jobs()
          for job in jobs:
              if job["IsDead"]:
                  return True
          return False
  
      wait_for_condition(is_job_done)
  
      @ray.remote(num_cpus=1)
      class Actor:
          def ping(self):
              return "pong"
  
      # Get the named placement group and schedule a actor.
      placement_group = ray.util.get_placement_group(
          global_placement_group_name
      )
      assert placement_group is not None
      assert placement_group.wait(5)
      actor = Actor.options(
          scheduling_strategy=PlacementGroupSchedulingStrategy(
              placement_group=placement_group,
              placement_group_bundle_index=0,
          )
      ).remote()
  
      ray.get(actor.ping.remote())
  
      # Create another placement group and make sure its creation will failed.
      error_creation_count = 0
      try:
          ray.util.placement_group(
              [{"CPU": 1} for _ in range(2)],
              strategy="STRICT_SPREAD",
              name=global_placement_group_name,
          )
      except RaySystemError:
          error_creation_count += 1
      assert error_creation_count == 1
  
      # Remove a named placement group and make sure the second creation
      # will successful.
      ray.util.remove_placement_group(placement_group)
      same_name_pg = ray.util.placement_group(
          [{"CPU": 1} for _ in range(2)],
          strategy="STRICT_SPREAD",
          name=global_placement_group_name,
      )
      assert same_name_pg.wait(10)
  
      # Get a named placement group with a name that doesn't exist
      # and make sure it will raise ValueError correctly.
      error_count = 0
      try:
          ray.util.get_placement_group("inexistent_pg")
      except ValueError:
          error_count = error_count + 1
      assert error_count == 1
  
imports: import sys
  import time
  import pytest
  import ray
  import ray._private.gcs_utils
  import ray.cluster_utils
  import ray.experimental.internal_kv
  from ray._private.ray_constants import DEBUG_AUTOSCALING_ERROR, DEBUG_AUTOSCALING_STATUS
  from ray.autoscaler._private.constants import AUTOSCALER_UPDATE_INTERVAL_S
  from ray._private.test_utils import convert_actor_state, generate_system_config_map, is_placement_group_removed, kill_actor_and_wait_for_failure, reset_autoscaler_v2_enabled_cache, run_string_as_driver, wait_for_condition
  from ray.autoscaler._private.commands import debug_status
  from ray.exceptions import RaySystemError
  from ray.util.client.ray_client_helpers import connect_to_client_or_not
  from ray.util.placement_group import placement_group, remove_placement_group
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
label_desc: Use image node_id 7 for calling a global function with example usage: test_named_placement_group(ray_start_cluster) and returns: False, str, True