code: def test_checkpoint_num_to_keep(
      ray_start_4_cpus_2_gpus_extra, resource_manager_cls, tmp_path
  ):
      """Test that only num_to_keep checkpoints are kept.
  
      This should also hold true when the experiment is resumed.
  
      Legacy test: test_trial_runner_2.py::TrialRunnerTest::testPauseResumeCheckpointCount
      """
      trial = Trial(
          "__fake",
          checkpoint_config=CheckpointConfig(num_to_keep=2),
          storage=STORAGE,
      )
      trial.init_local_path()
  
      def write_checkpoint(trial: Trial, index: int):
          checkpoint_dir = (
              tmp_path / StorageContext._make_checkpoint_dir_name(index)
          )
          checkpoint_dir.mkdir(parents=True, exist_ok=True)
          result = {"training_iteration": index}
          with open(os.path.join(checkpoint_dir, "cp.json"), "w") as f:
              json.dump(result, f)
  
          checkpoint = Checkpoint.from_directory(checkpoint_dir)
          return _TrainingResult(checkpoint=checkpoint, metrics=result)
  
      def get_checkpoint_dirs(trial: Trial):
          return [
              d
              for d in os.listdir(tmp_path)
              if d.startswith("checkpoint_")
          ]
  
      runner = TuneController(
          resource_manager_factory=lambda: resource_manager_cls(),
          storage=STORAGE,
      )
  
      runner.add_trial(trial)
  
      # Write 1 checkpoint
      result = write_checkpoint(trial, 1)
      runner._on_saving_result(trial, result)
  
      # Expect 1 checkpoint
      cp_dirs = get_checkpoint_dirs(trial)
      assert len(cp_dirs) == 1, f"Checkpoint dirs: {cp_dirs}"
  
      # Write second checkpoint
      result = write_checkpoint(trial, 2)
      runner._on_saving_result(trial, result)
  
      # Expect 2 checkpoints
      cp_dirs = get_checkpoint_dirs(trial)
      assert len(cp_dirs) == 2, f"Checkpoint dirs: {cp_dirs}"
  
      # Write third checkpoint
      result = write_checkpoint(trial, 3)
      runner._on_saving_result(trial, result)
  
      # Expect 2 checkpoints because num_to_keep = 2
      cp_dirs = get_checkpoint_dirs(trial)
      assert len(cp_dirs) == 2, f"Checkpoint dirs: {cp_dirs}"
  
      # Re-instantiate trial runner and resume
      runner.checkpoint(force=True)
      runner = TuneController(
          resource_manager_factory=lambda: resource_manager_cls(),
          storage=STORAGE,
      )
      runner.resume()
  
      trial = runner.get_trials()[0]
  
      # Write fourth checkpoint
      result = write_checkpoint(trial, 4)
      runner._on_saving_result(trial, result)
  
      # Expect 2 checkpoints because num_to_keep = 2
      cp_dirs = get_checkpoint_dirs(trial)
      assert len(cp_dirs) == 2, f"Checkpoint dirs: {cp_dirs}"
  
      # Write fifth checkpoint
      result = write_checkpoint(trial, 5)
      runner._on_saving_result(trial, result)
  
      # Expect 2 checkpoints because num_to_keep = 2
      cp_dirs = get_checkpoint_dirs(trial)
      assert len(cp_dirs) == 2, f"Checkpoint dirs: {cp_dirs}"
  
      # Checkpoints before restore should be deleted
      assert "checkpoint_000004" in cp_dirs
      assert "checkpoint_000005" in cp_dirs
  
      assert "checkpoint_000002" not in cp_dirs
      assert "checkpoint_000003" not in cp_dirs
  
imports: import json
  import logging
  import os
  import tempfile
  from unittest import mock
  import pytest
  import sys
  import time
  from functools import partial
  import ray
  from freezegun import freeze_time
  from ray.train import CheckpointConfig
  from ray.air.execution import FixedResourceManager, PlacementGroupResourceManager
  from ray.air.constants import TRAINING_ITERATION
  from ray.train import Checkpoint
  from ray.train._internal.session import _TrainingResult
  from ray.train._internal.storage import StorageContext
  from ray.tune import PlacementGroupFactory
  from ray.tune.execution.tune_controller import TuneController
  from ray.tune.experiment import Trial
  from ray.tune.result import DONE
  from ray.tune.schedulers import FIFOScheduler
  from ray.tune.search import BasicVariantGenerator
  from ray.train.tests.util import mock_storage_context
  from ray.tune.tests.tune_test_util import TrialResultObserver
label_desc: Use image node_id 6 for calling a global function with example usage: test_checkpoint_num_to_keep(ray_start_4_cpus_2_gpus_extra, resource_manager_cls, tmp_path) and returns: _TrainingResult, unknown