code: def test_placement_group_max_cpu_frac(
      ray_start_cluster, multi_bundle, even_pack, scheduling_strategy
  ):
      cluster = ray_start_cluster
      cluster.add_node(num_cpus=4)
      cluster.wait_for_nodes()
      ray.init(address=cluster.address)
  
      if multi_bundle:
          bundles = [{"CPU": 1}] * 3
      else:
          bundles = [{"CPU": 3}]
  
      # Input validation - max_cpu_fraction_per_node must be between 0 and 1.
      with pytest.raises(ValueError):
          ray.util.placement_group(
              bundles, _max_cpu_fraction_per_node=-1
          )
      with pytest.raises(ValueError):
          ray.util.placement_group(
              bundles, _max_cpu_fraction_per_node=2
          )
  
      pg = ray.util.placement_group(
          bundles,
          strategy=scheduling_strategy,
          _max_cpu_fraction_per_node=0.5,
      )
  
      # Placement group will never be scheduled since it would violate the max CPU
      # fraction reservation.
      with pytest.raises(ray.exceptions.GetTimeoutError):
          ray.get(pg.ready(), timeout=5)
  
      # Add new node with enough CPU cores to scheduled placement group bundle while
      # adhering to the max CPU fraction constraint.
      if even_pack:
          num_cpus = 6
      else:
          num_cpus = 8
      cluster.add_node(num_cpus=num_cpus)
      cluster.wait_for_nodes()
      # The placement group should be schedulable so this shouldn't raise.
      ray.get(pg.ready(), timeout=5)
  
file_path: ray/python/ray/tests/test_placement_group_5.py
imports: import sys
  import pytest
  import ray
  from ray.tests.test_placement_group import are_pairwise_unique
  from ray.util.client.ray_client_helpers import connect_to_client_or_not
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
label_desc: Use image node_id 2 for calling a global function with example usage: test_placement_group_max_cpu_frac(ray_start_cluster, multi_bundle, even_pack, scheduling_strategy) without return types