code: def test_search_alg_finishes(
      ray_start_4_cpus_2_gpus_extra, resource_manager_cls
  ):
      """Empty SearchAlg changing state in `next_trials` does not crash.
  
      The search algorithm changes to ``finished`` mid-run. This should not
      affect processing of the experiment.
  
      Legacy test: test_trial_runner_3.py::TrialRunnerTest::testSearchAlgFinishes
      """
      os.environ["TUNE_MAX_PENDING_TRIALS_PG"] = "1"
  
      class FinishFastAlg(_MockSuggestionAlgorithm):
          _index = 0
  
          def next_trial(self):
              spec = self._experiment.spec
              trial = None
              if self._index < spec["num_samples"]:
                  trial = Trial(
                      spec.get("run"),
                      stopping_criterion=spec.get("stop"),
                      storage=spec.get("storage"),
                  )
              self._index += 1
  
              if self._index > 4:
                  self.set_finished()
  
              return trial
  
          def suggest(self, trial_id):
              return {}
  
      experiment_spec = {
          "run": "__fake",
          "num_samples": 2,
          "stop": {"training_iteration": 1},
      }
      searcher = FinishFastAlg()
      experiments = [Experiment.from_json("test", experiment_spec)]
      searcher.add_configurations(experiments)
  
      runner = TestTuneController(
          resource_manager_factory=lambda: resource_manager_cls(),
          search_alg=searcher,
      )
  
      assert not runner.is_finished()
  
      while len(runner.get_trials()) < 2:
          runner.step()  # Launch 2 runs
  
      assert not searcher.is_finished()
      assert not runner.is_finished()
  
      searcher_finished_before = False
      while not runner.is_finished():
          runner.step()
          searcher_finished_before = searcher.is_finished()
  
      # searcher_finished_before will be True if the searcher was finished before
      # the controller.
      assert searcher_finished_before
  
imports: import os
  import pickle
  from collections import Counter
  import pytest
  import sys
  import ray
  from ray.air.constants import TRAINING_ITERATION
  from ray.air.execution import FixedResourceManager, PlacementGroupResourceManager
  from ray.tune import Experiment, PlacementGroupFactory
  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 Searcher, ConcurrencyLimiter, Repeater, SearchGenerator
  from ray.tune.search._mock import _MockSuggestionAlgorithm
  from ray.train.tests.util import mock_storage_context
label_desc: Use image node_id 6 for calling a global function with example usage: test_search_alg_finishes(ray_start_4_cpus_2_gpus_extra, resource_manager_cls) and returns: trial, dict