code: def test_basic(serve_instance):
      """Test the basic workflow: multiple replicas with their own PGs."""
  
      @serve.deployment(
          num_replicas=2,
          placement_group_bundles=[{"CPU": 1}, {"CPU": 0.1}],
      )
      class D:
          def get_pg(self) -> PlacementGroup:
              return get_current_placement_group()
  
      h = serve.run(D.bind(), name="pg_test")
  
      # Verify that each replica has its own placement group with the correct config.
      assert len(get_all_live_placement_group_names()) == 2
      unique_pgs = set(
          ray.get(
              [
                  h.get_pg.remote()._to_object_ref_sync()
                  for _ in range(20)
              ]
          )
      )
      assert len(unique_pgs) == 2
      for pg in unique_pgs:
          assert _get_pg_strategy(pg) == "PACK"
          assert pg.bundle_specs == [{"CPU": 1}, {"CPU": 0.1}]
  
      # Verify that all placement groups are deleted when the deployment is deleted.
      serve.delete("pg_test")
      assert len(get_all_live_placement_group_names()) == 0
  
file_path: ray/python/ray/serve/tests/test_replica_placement_group.py
imports: import os
  import sys
  import pytest
  import ray
  from ray import serve
  from ray._private.test_utils import wait_for_condition
  from ray.serve._private.utils import get_all_live_placement_group_names
  from ray.serve.context import _get_global_client
  from ray.util.placement_group import PlacementGroup, get_current_placement_group
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
label_desc: Use image node_id 2 for calling a global function with example usage: test_basic(serve_instance) and returns: get_current_placement_group