code: def test_checkpoint_force_with_num_to_keep(
      ray_start_4_cpus_2_gpus_extra, resource_manager_cls, tmp_path
  ):
      """Test that cloud syncing is forced if one of the trials has made more
      than num_to_keep checkpoints since last sync.
  
      Legacy test: test_trial_runner_3.py::TrialRunnerTest::
          testCloudCheckpointForceWithNumToKeep
      """
      storage = mock_storage_context(delete_syncer=False)
      # Needed to avoid infinite recursion error on CI runners
      storage.syncer.__getstate__ = lambda *a, **kw: {}
  
      with mock.patch.dict(
          os.environ,
          {
              "TUNE_WARN_EXCESSIVE_EXPERIMENT_CHECKPOINT_SYNC_THRESHOLD_S": "2"
          },
      ), mock.patch.object(storage.syncer, "sync_up") as sync_up:
          num_to_keep = 2
          checkpoint_config = CheckpointConfig(
              num_to_keep=num_to_keep, checkpoint_frequency=1
          )
  
          runner = TuneController(
              resource_manager_factory=lambda: resource_manager_cls(),
              storage=storage,
              checkpoint_period=100,  # only rely on force syncing
              trial_checkpoint_config=checkpoint_config,
          )
  
          class CheckpointingTrial(Trial):
              def should_checkpoint(self):
                  return True
  
              def get_json_state(self):
                  return "", ""
  
          trial = CheckpointingTrial(
              "__fake",
              checkpoint_config=checkpoint_config,
              stopping_criterion={"training_iteration": 10},
              storage=storage,
          )
          runner.add_trial(trial)
  
          # also check if the warning is printed
          buffer = []
          from ray.tune.execution.experiment_state import logger
  
          with mock.patch.object(
              logger, "warning", lambda x: buffer.append(x)
          ):
              while not runner.is_finished():
                  runner.step()
          assert any(
              "syncing has been triggered multiple" in x for x in buffer
          )
  
          # We should sync 6 times:
          # The first checkpoint happens when the experiment starts,
          # since no checkpoints have happened yet
          # (This corresponds to the new_trial event in the runner loop)
          # Then, every num_to_keep=2 checkpoints, we should perform a forced checkpoint
          # which results in 5 more checkpoints (running for 10 iterations),
          # giving a total of 6
          assert sync_up.call_count == 6
  
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 12 for calling a global function with example usage: test_checkpoint_force_with_num_to_keep(ray_start_4_cpus_2_gpus_extra, resource_manager_cls, tmp_path) and returns: True, str,  str