code: def test_callbacks(self, tmp_path: pathlib.Path) -> None:
      checkpoint_dir = tmp_path.joinpath("checkpoint")
      latest_checkpoint = None
      steps_completed = 0
  
      controller = None
  
      def make_workloads1() -> workload.Stream:
          nonlocal controller
          assert controller.trial.counter.trial_startups == 1
  
          yield workload.train_workload(
              1, 1, 0, 4
          ), workload.ignore_workload_response
          assert controller is not None, "controller was never set!"
          assert controller.trial.counter.__dict__ == {
              "trial_startups": 1,
              "validation_steps_started": 0,
              "validation_steps_ended": 0,
              "checkpoints_written": 0,
              "checkpoints_uploaded": 0,
              "training_started_times": 1,
              "training_epochs_started": 2,
              "training_epochs_ended": 2,
              "training_workloads_ended": 1,
              "trial_shutdowns": 0,
          }
  
          yield workload.validation_workload(), workload.ignore_workload_response
          assert controller.trial.counter.__dict__ == {
              "trial_startups": 1,
              "validation_steps_started": 1,
              "validation_steps_ended": 1,
              "checkpoints_written": 0,
              "checkpoints_uploaded": 0,
              "training_started_times": 1,
              "training_epochs_started": 2,
              "training_epochs_ended": 2,
              "training_workloads_ended": 1,
              "trial_shutdowns": 0,
          }
  
          interceptor = workload.WorkloadResponseInterceptor()
          yield from interceptor.send(workload.checkpoint_workload())
          nonlocal latest_checkpoint, steps_completed
          latest_checkpoint = interceptor.metrics_result()["uuid"]
          steps_completed = 1
          assert controller.trial.counter.__dict__ == {
              "trial_startups": 1,
              "validation_steps_started": 1,
              "validation_steps_ended": 1,
              "checkpoints_written": 1,
              "checkpoints_uploaded": 1,
              "training_started_times": 1,
              "training_epochs_started": 2,
              "training_epochs_ended": 2,
              "training_workloads_ended": 1,
              "trial_shutdowns": 0,
          }
  
      hparams1 = dict(self.hparams)
      controller = (
          utils.make_trial_controller_from_trial_implementation(
              trial_class=deepspeed_linear_model.LinearCallbackTrial,
              hparams=hparams1,
              workloads=make_workloads1(),
              checkpoint_dir=str(checkpoint_dir),
              expose_gpus=True,
          )
      )
      controller.run()
      assert controller.trial.counter.trial_shutdowns == 1
  
      # Verify the checkpoint loading callback works.
      def make_workloads2() -> workload.Stream:
          yield workload.train_workload(
              1, 1, 0, 2
          ), workload.ignore_workload_response
  
      controller = (
          utils.make_trial_controller_from_trial_implementation(
              trial_class=deepspeed_linear_model.LinearCallbackTrial,
              hparams=self.hparams,
              workloads=make_workloads2(),
              checkpoint_dir=str(checkpoint_dir),
              latest_checkpoint=latest_checkpoint,
              steps_completed=steps_completed,
              expose_gpus=True,
          )
      )
      controller.run()
      assert controller.trial.counter.__dict__ == {
          # Note: trial_startups will get reset by the loading logic.
          "trial_startups": 1,
          "validation_steps_started": 1,
          "validation_steps_ended": 1,
          # Note: checkpoints_written, checkpoints_uploaded, and trial_shutdowns, cannot be
          # persisted, as they are all updated after checkpointing.
          "checkpoints_written": 0,
          "checkpoints_uploaded": 0,
          "training_started_times": 2,
          "training_epochs_started": 3,
          "training_epochs_ended": 3,
          "training_workloads_ended": 2,
          "trial_shutdowns": 1,
      }
  
imports: import copy
  import json
  import os
  import pathlib
  from typing import Any, Dict, Iterator, Optional
  import pytest
  import torch
  from deepspeed.runtime import config_utils
  import determined
  import determined.pytorch.deepspeed
  from determined import workload
  from tests.experiment import utils
  from tests.experiment.fixtures import deepspeed_linear_model
label_desc: Use image node_id 22 for calling the TestDeepSpeedTrial obj's underlying member method code with example usage: obj.test_callbacks(tmp_path) without return types