code: def test_gpu_requests(ray_start_4_cpus_4_gpus_4_extra, tmp_path):
      def get_visible_devices_for_workers():
          return [file.read_text() for file in tmp_path.glob("*")]
  
      class CudaTestBackend(Backend):
          share_cuda_visible_devices = True
  
      class CudaTestConfig(BackendConfig):
          @property
          def backend_cls(self):
              return CudaTestBackend
  
      def get_resources():
          cuda_visible_devices = os.environ["CUDA_VISIBLE_DEVICES"]
          world_rank = train.get_context().get_world_rank()
          (tmp_path / f"{world_rank}").write_text(cuda_visible_devices)
          train.report(dict(devices=cuda_visible_devices))
  
      # 0 GPUs will be requested and should not raise an error.
      trainer = DataParallelTrainer(
          get_resources,
          backend_config=CudaTestConfig(),
          scaling_config=ScalingConfig(num_workers=2, use_gpu=False),
      )
      trainer.fit()
      assert get_visible_devices_for_workers() == ["", ""]
  
      # 1 GPU will be requested and should not raise an error.
      trainer = DataParallelTrainer(
          get_resources,
          backend_config=CudaTestConfig(),
          scaling_config=ScalingConfig(num_workers=2, use_gpu=True),
      )
      trainer.fit()
      visible_devices = get_visible_devices_for_workers()
      # Sort the cuda visible devices to have exact match with expected result.
      visible_devices = [
          ",".join(sorted(r.split(","))) for r in visible_devices
      ]
      assert visible_devices == ["0,1", "0,1"]
  
      # Partial GPUs should not raise an error.
      trainer = DataParallelTrainer(
          get_resources,
          backend_config=CudaTestConfig(),
          scaling_config=ScalingConfig(
              num_workers=2,
              use_gpu=True,
              resources_per_worker={"GPU": 0.1},
          ),
      )
      trainer.fit()
      visible_devices = get_visible_devices_for_workers()
      assert visible_devices == ["0", "0"]
  
      # Multiple GPUs should not raise an error.
      trainer = DataParallelTrainer(
          get_resources,
          backend_config=CudaTestConfig(),
          scaling_config=ScalingConfig(
              num_workers=2,
              use_gpu=True,
              resources_per_worker={"GPU": 2},
          ),
      )
      trainer.fit()
      visible_devices = get_visible_devices_for_workers()
      # Sort the cuda visible devices to have exact match with expected result.
      visible_devices = [
          ",".join(sorted(r.split(","))) for r in visible_devices
      ]
      assert visible_devices == ["0,1,2,3", "0,1,2,3"]
  
imports: import os
  import time
  from unittest.mock import patch
  import pytest
  import ray
  from ray import train, tune
  from ray.train import RunConfig, ScalingConfig
  from ray.train._internal.backend_executor import BackendExecutor
  from ray.train._internal.worker_group import WorkerGroup
  from ray.train.backend import Backend, BackendConfig
  from ray.train.data_parallel_trainer import DataParallelTrainer
  from ray.train.tests.util import create_dict_checkpoint, load_dict_checkpoint
  from ray.tune.callback import Callback
  from ray.tune.tune_config import TuneConfig
  from ray.tune.tuner import Tuner
label_desc: Use image node_id 16 for calling a global function with example usage: test_gpu_requests(ray_start_4_cpus_4_gpus_4_extra, tmp_path) and returns: unknown, CudaTestBackend