code: def testCallbackSteps(self):
      trials = [
          Trial("__fake", trial_id="one"),
          Trial("__fake", trial_id="two"),
      ]
      for t in trials:
          self.trial_runner.add_trial(t)
  
      self.executor.next_future_result = _ExecutorEvent(
          event_type=_ExecutorEventType.PG_READY
      )
      self.trial_runner.step()
  
      # Trial 1 has been started
      self.assertEqual(
          self.callback.state["trial_start"]["iteration"], 0
      )
      self.assertEqual(
          self.callback.state["trial_start"]["trial"].trial_id, "one"
      )
  
      # All these events haven't happened, yet
      self.assertTrue(
          all(
              k not in self.callback.state
              for k in [
                  "trial_restore",
                  "trial_save",
                  "trial_result",
                  "trial_complete",
                  "trial_fail",
                  "experiment_end",
              ]
          )
      )
  
      self.executor.next_future_result = _ExecutorEvent(
          event_type=_ExecutorEventType.PG_READY
      )
      self.trial_runner.step()
  
      # Iteration not increased yet
      self.assertEqual(
          self.callback.state["step_begin"]["iteration"], 1
      )
  
      # Iteration increased
      self.assertEqual(self.callback.state["step_end"]["iteration"], 2)
  
      # Second trial has been just started
      self.assertEqual(
          self.callback.state["trial_start"]["iteration"], 1
      )
      self.assertEqual(
          self.callback.state["trial_start"]["trial"].trial_id, "two"
      )
  
      # Just a placeholder object ref for cp.value.
      cp = _TrackedCheckpoint(
          dir_or_data=ray.put(1),
          storage_mode=CheckpointStorage.PERSISTENT,
          metrics={TRAINING_ITERATION: 0},
      )
      trials[0].temporary_state.saving_to = cp
  
      # Let the first trial save a checkpoint
      self.executor.next_future_result = _ExecutorEvent(
          event_type=_ExecutorEventType.SAVING_RESULT,
          trial=trials[0],
          result={_ExecutorEvent.KEY_FUTURE_RESULT: "__checkpoint"},
      )
      self.trial_runner.step()
      self.assertEqual(
          self.callback.state["trial_save"]["iteration"], 2
      )
      self.assertEqual(
          self.callback.state["trial_save"]["trial"].trial_id, "one"
      )
  
      # Let the second trial send a result
      result = {TRAINING_ITERATION: 1, "metric": 800, "done": False}
      self.executor.next_future_result = _ExecutorEvent(
          event_type=_ExecutorEventType.TRAINING_RESULT,
          trial=trials[1],
          result={_ExecutorEvent.KEY_FUTURE_RESULT: result},
      )
      self.assertTrue(not trials[1].has_reported_at_least_once)
      self.trial_runner.step()
      self.assertEqual(
          self.callback.state["trial_result"]["iteration"], 3
      )
      self.assertEqual(
          self.callback.state["trial_result"]["trial"].trial_id, "two"
      )
      self.assertEqual(
          self.callback.state["trial_result"]["result"]["metric"], 800
      )
      self.assertEqual(trials[1].last_result["metric"], 800)
  
      # Let the second trial restore from a checkpoint
      trials[1].temporary_state.restoring_from = cp
      self.executor.next_future_result = _ExecutorEvent(
          event_type=_ExecutorEventType.RESTORING_RESULT,
          trial=trials[1],
          result={_ExecutorEvent.KEY_FUTURE_RESULT: None},
      )
      self.trial_runner.step()
      self.assertEqual(
          self.callback.state["trial_restore"]["iteration"], 4
      )
      self.assertEqual(
          self.callback.state["trial_restore"]["trial"].trial_id, "two"
      )
  
      # Let the second trial finish
      trials[1].temporary_state.restoring_from = None
      self.executor.next_future_result = _ExecutorEvent(
          event_type=_ExecutorEventType.TRAINING_RESULT,
          trial=trials[1],
          result={
              _ExecutorEvent.KEY_FUTURE_RESULT: {
                  TRAINING_ITERATION: 2,
                  "metric": 900,
                  "done": True,
              }
          },
      )
      self.trial_runner.step()
      self.assertEqual(
          self.callback.state["trial_complete"]["iteration"], 5
      )
      self.assertEqual(
          self.callback.state["trial_complete"]["trial"].trial_id, "two"
      )
  
      # Let the first trial error
      self.executor.next_future_result = _ExecutorEvent(
          event_type=_ExecutorEventType.TRAINING_RESULT,
          trial=trials[0],
          result={_ExecutorEvent.KEY_EXCEPTION: Exception()},
      )
      self.trial_runner.step()
      self.assertEqual(
          self.callback.state["trial_fail"]["iteration"], 6
      )
      self.assertEqual(
          self.callback.state["trial_fail"]["trial"].trial_id, "one"
      )
  
imports: import os
  import shutil
  import sys
  import tempfile
  import time
  import unittest
  from unittest.mock import patch
  from collections import OrderedDict
  import ray
  from ray import tune
  from ray.air._internal.checkpoint_manager import _TrackedCheckpoint, CheckpointStorage
  from ray.air.constants import TRAINING_ITERATION
  from ray.rllib import _register_all
  from ray.tune.execution.ray_trial_executor import _ExecutorEvent, _ExecutorEventType, RayTrialExecutor
  from ray.tune.callback import warnings
  from ray.tune.experiment import Trial
  from ray.tune.execution.trial_runner import TrialRunner
  from ray.tune import Callback
  from ray.tune.experiment import Experiment
label_desc: Use image node_id 3 for calling the TrialRunnerCallbacks obj's underlying member method code with example usage: obj.testCallbackSteps() without return types