code: def test_mini_integration(ray_start_cluster, connect_to_client):
      # Create bundles as many as number of gpus in the cluster.
      # Do some random work and make sure all resources are properly recovered.
  
      cluster = ray_start_cluster
  
      num_nodes = 5
      per_bundle_gpus = 2
      gpu_per_node = 4
      total_gpus = num_nodes * per_bundle_gpus * gpu_per_node
      per_node_gpus = per_bundle_gpus * gpu_per_node
  
      bundles_per_pg = 2
      total_num_pg = total_gpus // (bundles_per_pg * per_bundle_gpus)
  
      [
          cluster.add_node(
              num_cpus=2, num_gpus=per_bundle_gpus * gpu_per_node
          )
          for _ in range(num_nodes)
      ]
      cluster.wait_for_nodes()
      ray.init(address=cluster.address)
  
      with connect_to_client_or_not(connect_to_client):
  
          @ray.remote(num_cpus=0, num_gpus=1)
          def random_tasks():
              import random
              import time
  
              sleep_time = random.uniform(0.1, 0.2)
              time.sleep(sleep_time)
              return True
  
          pgs = []
          pg_tasks = []
          # total bundle gpu usage = bundles_per_pg*total_num_pg*per_bundle_gpus
          # Note this is half of total
          for index in range(total_num_pg):
              pgs.append(
                  ray.util.placement_group(
                      name=f"name{index}",
                      strategy="PACK",
                      bundles=[
                          {"GPU": per_bundle_gpus}
                          for _ in range(bundles_per_pg)
                      ],
                  )
              )
  
          # Schedule tasks.
          for i in range(total_num_pg):
              pg = pgs[i]
              pg_tasks.append(
                  [
                      random_tasks.options(
                          scheduling_strategy=PlacementGroupSchedulingStrategy(
                              placement_group=pg,
                              placement_group_bundle_index=bundle_index,
                          )
                      ).remote()
                      for bundle_index in range(bundles_per_pg)
                  ]
              )
  
          # Make sure tasks are done and we remove placement groups.
          num_removed_pg = 0
          pg_indexes = [2, 3, 1, 7, 8, 9, 0, 6, 4, 5]
          while num_removed_pg < total_num_pg:
              index = pg_indexes[num_removed_pg]
              pg = pgs[index]
              assert all(ray.get(pg_tasks[index]))
              ray.util.remove_placement_group(pg)
              num_removed_pg += 1
  
          @ray.remote(num_cpus=2, num_gpus=per_node_gpus)
          class A:
              def ping(self):
                  return True
  
          # Make sure all resources are properly returned by scheduling
          # actors that take up all existing resources.
          actors = [A.remote() for _ in range(num_nodes)]
          assert all(ray.get([a.ping.remote() for a in actors]))
  
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 6 for calling a global function with example usage: test_mini_integration(ray_start_cluster, connect_to_client) and returns: True, True