code: def test_acquire_return_resources(ray_start_4_cpus):
      """Tests that acquiring and returning resources works.
  
      - At the start, no resources should be ready (no PG scheduled)
      - Request resources for 2 CPUs
      - (wait until they are ready)
      - Assert that these 2 CPUs are available to be acquired
      - Acquire
      - Assert that there are no 2 CPU resources available anymore
      - Free resources
      - Assert that the 2 CPU resources are still not available (no new request)
          - This is also tested in includes test_request_cancel_resources
      """
      manager = PlacementGroupResourceManager(update_interval_s=0)
      assert not manager.has_resources_ready(REQUEST_2_CPU)
  
      # Request PG
      manager.request_resources(REQUEST_2_CPU)
  
      # Wait until ready
      ray.wait(manager.get_resource_futures(), num_returns=1)
  
      assert manager.has_resources_ready(REQUEST_2_CPU)
  
      # PG exists
      pg_states = _count_pg_states()
      assert pg_states["CREATED"] == 1
      assert pg_states["REMOVED"] == 0
  
      # Acquire PG
      acquired = manager.acquire_resources(REQUEST_2_CPU)
  
      assert not manager.has_resources_ready(REQUEST_2_CPU)
  
      # Free resources
      manager.free_resources(acquired)
  
      assert not manager.has_resources_ready(REQUEST_2_CPU)
  
      # PG still exists
      pg_states = _count_pg_states()
      assert pg_states["CREATED"] == 0
      assert pg_states["REMOVED"] == 1
  
file_path: ray/python/ray/air/tests/execution/test_resource_manager_placement_group.py
imports: import time
  from collections import Counter
  import pytest
  import ray
  from ray.air.execution.resources.placement_group import PlacementGroupResourceManager
  from ray.air.execution.resources.request import ResourceRequest
label_desc: Use image node_id 4 for calling a global function with example usage: test_acquire_return_resources(ray_start_4_cpus) without return types