code: def test_tuner(
      monkeypatch,
      tmp_path,
      trainable,
      storage_path_type,
      checkpoint_config: train.CheckpointConfig,
  ):
      """End-to-end test that the new persistence mode works with the Tuner API.
      This test covers many `storage_path_type` options:
      - storage_path=None --> save locally to the default local path (e.g., ~/ray_results)
      - storage_path="nfs" --> save locally to a fake NFS path
      - storage_path="cloud" --> save to a mock S3 bucket
      - storage_path="custom_fs" --> save to a custom pyarrow filesystem
          - The custom fs is a local filesystem that appends a path prefix to every path.
  
      This is the expected output at the storage path:
  
      {storage_path}/{exp_name}
      ├── tuner.pkl                   <- Driver artifacts (global experiment state)
      ├── basic-variant-state.json
      ├── experiment_state.json
      ├── train_fn_a2b9e_00000_0_...
      │   ├── artifact-iter=0.txt     <- Trial artifacts
      │   ├── ...
      │   ├── checkpoint_000000       <- Trial checkpoints
      │   │   └── checkpoint.pkl
      │   ├── ...
      │   ├── events.out.tfevents...  <- Driver artifacts (trial results)
      │   ├── params.json
      │   ├── params.pkl
      │   ├── progress.csv
      │   └── result.json
      └── train_fn_a2b9e_00001_1_...
          └── ...                     <- Same as above
      """
      # Set the cache dir to some temp directory
      LOCAL_CACHE_DIR = tmp_path / "ray_results"
      monkeypatch.setenv(
          "RAY_AIR_LOCAL_CACHE_DIR", str(LOCAL_CACHE_DIR)
      )
  
      exp_name = "simple_persistence_test"
  
      with _resolve_storage_type(storage_path_type, tmp_path) as (
          storage_path,
          storage_filesystem,
      ):
          tuner = tune.Tuner(
              trainable,
              param_space={
                  "num_iterations": TestConstants.NUM_ITERATIONS,
                  "fail_iters": [2, 4],
                  # NOTE: This param is only used in the ClassTrainable.
                  "save_checkpoint_as_dict": tune.grid_search(
                      [True, False]
                  ),
                  "tmp_path": tmp_path,
              },
              run_config=train.RunConfig(
                  storage_path=storage_path,
                  storage_filesystem=storage_filesystem,
                  name=exp_name,
                  verbose=0,
                  failure_config=train.FailureConfig(max_failures=1),
                  checkpoint_config=checkpoint_config,
                  sync_config=train.SyncConfig(sync_artifacts=True),
              ),
              # 2 samples (from the grid search). Run 1 at at time to test actor reuse
              tune_config=tune.TuneConfig(
                  num_samples=1, max_concurrent_trials=1
              ),
          )
          result_grid = tuner.fit()
          assert result_grid.errors
  
          if storage_path:
              shutil.rmtree(LOCAL_CACHE_DIR, ignore_errors=True)
  
          restored_tuner = tune.Tuner.restore(
              path=str(
                  URI(storage_path or str(LOCAL_CACHE_DIR)) / exp_name
              ),
              trainable=trainable,
              storage_filesystem=storage_filesystem,
              resume_errored=True,
          )
          result_grid = restored_tuner.fit()
          assert not result_grid.errors
  
          local_inspect_dir, storage_fs_path = _get_local_inspect_dir(
              root_local_path=tmp_path,
              storage_path=storage_path,
              storage_local_path=LOCAL_CACHE_DIR,
              storage_filesystem=storage_filesystem,
          )
  
      # First, check that the ResultGrid returns the correct paths.
      print(result_grid)
      experiment_fs_path = result_grid.experiment_path
      assert isinstance(
          result_grid.filesystem, pyarrow.fs.FileSystem
      ), result_grid
      assert experiment_fs_path == os.path.join(
          storage_fs_path, exp_name
      )
      assert len(result_grid) == TestConstants.NUM_TRIALS
      for result in result_grid:
          trial_fs_path = result.path
          assert isinstance(
              result.filesystem, pyarrow.fs.FileSystem
          ), result
          assert trial_fs_path.startswith(experiment_fs_path)
          for checkpoint, _ in result.best_checkpoints:
              assert checkpoint.path.startswith(trial_fs_path)
  
      # Next, inspect the storage path contents.
      _assert_storage_contents(
          local_inspect_dir,
          exp_name,
          checkpoint_config,
          trainable_name=trainable.__name__,
          test_trainer=False,
      )
  
imports: import logging
  import os
  import pickle
  import re
  import shutil
  import tempfile
  import time
  from contextlib import contextmanager
  from pathlib import Path
  from typing import List, Optional, Tuple
  import pyarrow.fs
  import pytest
  import ray
  from ray import train, tune
  from ray._private.test_utils import simulate_storage
  from ray.air._internal.uri_utils import URI
  from ray.air.constants import EXPR_RESULT_FILE
  from ray.train._checkpoint import Checkpoint
  from ray.train._internal.storage import StorageContext, _delete_fs_path, _download_from_fs_path
  from ray.train.base_trainer import TrainingFailedError
  from ray.train.data_parallel_trainer import DataParallelTrainer
  from ray.tune.trainable.trainable import _DICT_CHECKPOINT_FILE_NAME
label_desc: Use image node_id 14 for calling a global function with example usage: test_tuner(monkeypatch, tmp_path, trainable, storage_path_type, checkpoint_config) without return types