[tune] TuneController remaining test migration (1/n) (ray-project#39704)

This PR migrates more tests from test_trial_runner_3.py to the new tune controller.

Signed-off-by: Kai Fricke <kai@anyscale.com>
Signed-off-by: Victor <vctr.y.m@example.com>

python/ray/train/tests/util.py

@@ -25,15 +25,19 @@ def load_dict_checkpoint(checkpoint: Checkpoint) -> Dict[str, Any]:
             return ray_pickle.load(f)
 
 
-def mock_storage_context() -> StorageContext:
+def mock_storage_context(
+    exp_name: str = "exp_name", delete_syncer: bool = True
+) -> StorageContext:
     storage_path = tempfile.mkdtemp()
-    exp_name = "exp_name"
+    exp_name = exp_name
     trial_name = "trial_name"
     storage = StorageContext(
         storage_path=storage_path,
         experiment_dir_name=exp_name,
         trial_dir_name=trial_name,
     )
     storage.storage_local_path = storage_path
+    if delete_syncer:
+        storage.syncer = None
     os.makedirs(os.path.join(storage_path, exp_name, trial_name), exist_ok=True)
     return storage

python/ray/tune/BUILD

@@ -249,6 +249,14 @@ py_test(
     tags = ["team:ml", "exclusive"],
 )
 
+py_test(
+    name = "test_searcher_utils",
+    size = "small",
+    srcs = ["tests/test_searcher_utils.py"],
+    deps = [":tune_lib"],
+    tags = ["team:ml", "exclusive"],
+)
+
 py_test(
     name = "test_searchers",
     size = "large",

python/ray/tune/execution/tune_controller.py

@@ -1810,7 +1810,14 @@ def _process_trial_result(self, trial, result):
             # Cache decision to execute on after the save is processed.
             # This prevents changing the trial's state or kicking off
             # another training step prematurely.
-            self._cached_trial_decisions[trial.trial_id] = decision
+            if not self._cached_trial_decisions.get(trial.trial_id) or decision in {
+                TrialScheduler.PAUSE,
+                TrialScheduler.STOP,
+            }:
+                # If already set, only overwrite if it's a PAUSE or STOP. This is
+                # to avoid that CONTINUE decisions from a training step that resolve
+                # late overwrite PAUSE/STOP decision.
+                self._cached_trial_decisions[trial.trial_id] = decision
             return None
         else:
             self._queue_decision(trial, decision)