code: def test_controller_restore_dataset_references(
      ray_start_4_cpus_2_gpus_extra, resource_manager_cls
  ):
      """Check that references to Ray Datasets are replaced on resume.
  
      Legacy test: test_trial_runner_3.py::TrialRunnerTest::
          testSearcherCorrectReferencesAfterRestore
      """
  
      class FakeDataset:
          def __init__(self, name):
              self.name = name
  
      config = {
          "param1": {
              "param2": tune.grid_search(
                  [FakeDataset("1"), FakeDataset("2"), FakeDataset("3")]
              ),
          },
          "param4": tune.sample_from(lambda: 1),
          "param5": tune.sample_from(
              lambda spec: spec.config["param1"]["param2"]
          ),
      }
      resolvers = create_resolvers_map()
      config = inject_placeholders(config, resolvers)
  
      def create_searcher():
          search_alg = BasicVariantGenerator()
          experiment_spec = {
              "run": "__fake",
              "stop": {"training_iteration": 2},
              "config": config,
          }
          experiments = [Experiment.from_json("test", experiment_spec)]
          search_alg.add_configurations(experiments)
          return search_alg
  
      searcher = create_searcher()
  
      restored_config = {
          "param1": {
              "param2": tune.grid_search(
                  [FakeDataset("4"), FakeDataset("5"), FakeDataset("6")]
              ),
          },
          "param4": tune.sample_from(lambda: 8),
          "param5": tune.sample_from(
              lambda spec: spec["config"]["param1"]["param2"]
          ),
      }
      replaced_resolvers = create_resolvers_map()
      inject_placeholders(restored_config, replaced_resolvers)
  
      runner = TuneController(
          resource_manager_factory=lambda: resource_manager_cls(),
          reuse_actors=False,
          search_alg=searcher,
          placeholder_resolvers=replaced_resolvers,
          checkpoint_period=-1,
          storage=STORAGE,
      )
  
      while len(runner.get_trials()) < 3 or any(
          trial.status not in {Trial.RUNNING, Trial.TERMINATED}
          for trial in runner.get_trials()
      ):
          runner.step()
  
      assert len(runner.get_trials()) == 3, [
          t.config for t in runner.get_trials()
      ]
      for t in runner.get_trials():
          # Make sure that all the trials carry updated config values.
          assert t.config["param1"]["param2"].name in ["4", "5", "6"]
          assert t.config["param4"] == 8
          assert t.config["param5"].name in ["4", "5", "6"]
  
imports: import os
  from unittest.mock import patch
  import pandas
  import pytest
  import sys
  import ray
  from ray import tune
  from ray.air.execution import FixedResourceManager, PlacementGroupResourceManager
  from ray.train import CheckpointConfig
  from ray.tune import Experiment, PlacementGroupFactory
  from ray.tune.execution.tune_controller import TuneController
  from ray.tune.experiment import Trial
  from ray.tune.impl.placeholder import create_resolvers_map, inject_placeholders
  from ray.tune.search import BasicVariantGenerator
  from ray.rllib.algorithms.callbacks import DefaultCallbacks
  from ray.train.tests.util import mock_storage_context
label_desc: Use image node_id 2 for calling a global function with example usage: test_controller_restore_dataset_references(ray_start_4_cpus_2_gpus_extra, resource_manager_cls) and returns: search_alg