code: def test_tuner_restore_from_moved_experiment_path(
      ray_start_2_cpus, tmp_path, use_tune_run
  ):
      """Check that restoring a Tuner from a moved experiment directory works."""
      # Create a fail_marker dummy file that causes the first Tune run to fail and
      # the second run to succeed
      os.environ["RAY_AIR_LOCAL_CACHE_DIR"] = str(
          tmp_path / "local_dir"
      )
      fail_marker = tmp_path / "fail_marker"
      fail_marker.write_text("", encoding="utf-8")
  
      old_storage_path = tmp_path / "ray_results"
      old_exp_name = "exp_dir"
  
      new_storage_path = tmp_path / "new_ray_results"
      new_exp_name = "new_exp_dir"
  
      # Initial training run (that errors out in the middle)
      num_to_keep = 2
      tuner = Tuner(
          _train_fn_sometimes_failing,
          tune_config=TuneConfig(
              num_samples=1,
          ),
          run_config=RunConfig(
              name=old_exp_name,
              storage_path=str(old_storage_path),
              checkpoint_config=CheckpointConfig(
                  num_to_keep=num_to_keep
              ),
          ),
          param_space={
              "failing_hanging": (fail_marker, None),
          },
      )
      tuner.fit()
  
      # Move experiment from `tmp_path/ray_results/exp_dir`
      # to `tmp_path/moved_ray_results/new_exp_dir`, changing both `storage_path` and
      # the experiment `name`
      shutil.move(str(old_storage_path), str(new_storage_path))
      os.rename(
          str(new_storage_path / old_exp_name),
          str(new_storage_path / new_exp_name),
      )
  
      # Check that the results can be read from the new location.
      restore_path = str(new_storage_path / new_exp_name)
      results = ResultGrid(ExperimentAnalysis(restore_path))
  
      # TODO(justinvyu): [populate_exception] for storage_path != None
      # assert len(results.errors) == 1
      training_iteration = results[0].metrics["training_iteration"]
      assert (
          training_iteration == 1
      ), f"Should only have 1 train.report before erroring, got {training_iteration}"
      assert results[0].checkpoint.path.endswith("checkpoint_000000")
      assert "new_exp_dir" in results[0].checkpoint.path
  
      del tuner
      # Remove fail_marker so that the restored Tuner doesn't error again
      fail_marker.unlink()
  
      # Restore from moved experiment directory location, and launch resumed training
      if use_tune_run:
          analysis = tune.run(
              _train_fn_sometimes_failing,
              name=new_exp_name,
              storage_path=str(new_storage_path),
              resume="AUTO+ERRORED",
          )
          results = ResultGrid(analysis)
      else:
          tuner = Tuner.restore(
              restore_path,
              trainable=_train_fn_sometimes_failing,
              resume_errored=True,
          )
          results = tuner.fit()
  
      assert len(results.errors) == 0
  
      # Check that we restored iter=1, then made 2 calls to train.report -> iter=3
      training_iteration = results[0].metrics["training_iteration"]
      assert training_iteration == 3, training_iteration
  
      # Make sure that checkpoints are loaded properly
      assert results[0].checkpoint
      assert len(results[0].best_checkpoints) == num_to_keep
      checkpoint_dirs = [
          path
          for path in os.listdir(results[0].path)
          if path.startswith("checkpoint_")
      ]
      assert sorted(checkpoint_dirs) == [
          "checkpoint_000001",
          "checkpoint_000002",
      ]
  
      # Make sure that we did not create a logdir in the old location
      assert not old_storage_path.exists()
  
imports: import json
  import logging
  import os
  import shutil
  import time
  import unittest
  import pytest
  import ray
  from ray import train, tune
  import ray.cloudpickle
  from ray.train import Checkpoint, CheckpointConfig, FailureConfig, RunConfig, ScalingConfig
  from ray.air._internal.uri_utils import URI
  from ray.train.data_parallel_trainer import DataParallelTrainer
  from ray.train._internal.storage import get_fs_and_path, _download_from_fs_path, _upload_to_fs_path
  from ray.tune import Callback, Trainable
  from ray.tune.analysis import ExperimentAnalysis
  from ray.tune.execution.experiment_state import _find_newest_experiment_checkpoint
  from ray.tune.experiment import Trial
  from ray.tune.result_grid import ResultGrid
  from ray.tune.schedulers.async_hyperband import ASHAScheduler
  from ray.tune.search.optuna import OptunaSearch
  from ray.tune.tune_config import TuneConfig
  from ray.tune.tuner import Tuner
  from ray.train.tests.util import create_dict_checkpoint, load_dict_checkpoint
label_desc: Use image node_id 21 for calling a global function with example usage: test_tuner_restore_from_moved_experiment_path(ray_start_2_cpus, tmp_path, use_tune_run) without return types