code: def test_resources_changing(
      ray_start_4_cpus_2_gpus_extra, resource_manager_cls
  ):
      """Checks that resource requirements can be changed on fly.
  
      Legacy test: test_trial_runner.py::TrialRunnerTest::testChangeResources
      """
  
      class ChangingScheduler(FIFOScheduler):
          def on_trial_result(self, tune_controller, trial, result):
              if result["training_iteration"] == 1:
                  # NOTE: This is a hack to get around the new pausing logic,
                  # which doesn't set the trial status to PAUSED immediately.
                  orig_status = trial.status
                  trial.set_status(Trial.PAUSED)
                  trial.update_resources(dict(cpu=4, gpu=0))
                  trial.set_status(orig_status)
                  return TrialScheduler.PAUSE
              return TrialScheduler.NOOP
  
      scheduler = ChangingScheduler()
      runner = TuneController(
          resource_manager_factory=lambda: resource_manager_cls(),
          scheduler=scheduler,
          storage=STORAGE,
      )
      kwargs = {
          "stopping_criterion": {"training_iteration": 2},
          "placement_group_factory": PlacementGroupFactory(
              [{"CPU": 2, "GPU": 0}]
          ),
          "storage": STORAGE,
      }
      trials = [Trial("__fake", **kwargs)]
      for t in trials:
          runner.add_trial(t)
  
      while not trials[0].status == Trial.RUNNING:
          runner.step()
  
      assert trials[0].status == Trial.RUNNING
      assert (
          runner._actor_manager.get_live_actors_resources().get("CPU")
          == 2
      )
  
      with pytest.raises(ValueError):
          trials[0].update_resources(dict(cpu=4, gpu=0))
  
      while trials[0].status == Trial.RUNNING:
          runner.step()
  
      assert trials[0].status == Trial.PAUSED
  
      while not trials[0].status == Trial.RUNNING:
          runner.step()
  
      assert (
          runner._actor_manager.get_live_actors_resources().get("CPU")
          == 4
      )
  
      runner.step()
  
imports: import os
  import time
  from collections import Counter
  import pytest
  import sys
  import ray
  from ray import tune
  from ray.air.execution import FixedResourceManager, PlacementGroupResourceManager
  from ray.tune import PlacementGroupFactory, TuneError
  from ray.tune.execution.tune_controller import TuneController
  from ray.tune.experiment import Trial
  from ray.tune.schedulers import FIFOScheduler, TrialScheduler
  from ray.tune.search import BasicVariantGenerator
  from ray.tune.utils.mock import TrialStatusSnapshot, TrialStatusSnapshotTaker
  from ray.train.tests.util import mock_storage_context
label_desc: Use image node_id 5 for calling a global function with example usage: test_resources_changing(ray_start_4_cpus_2_gpus_extra, resource_manager_cls) and returns: TrialScheduler, TrialScheduler