class_docstr: This class is for testing directly via get_regression_trainer
  
  It mixes in `TrainerIntegrationCommon` which already has a lot of helper validation methods
  which we can re-use here.
  
  Important: this class' setup can only work with a single gpu because it runs within the current
  pytest worker. For multi-gpu tests use TestDeepSpeedWithLauncher.
  
  Note: if any of the tests of this class get run there will be at least one gpu occupied by them
  until this pytest worker exits. This is because the gpu memory allocated by the cuda-kernels
  won't be released until this pytest worker exits.
  
  This may appear as some run-away tests if you watch `nvidia-smi` while other tests that fork new
  processes are run. So there will be one or two "stale" processes reported in `nvidia-smi`. This
  is not a bug.
class_docstr_tok: ["This","class","is","for","testing","directly","via","get_regression_trainer","It","mixes","in","`","TrainerIntegrationCommon","`","which","already","has","a","lot","of","helper","validation","methods","which","we","can","re-use","here",".","Important",":","this","class","'","setup","can","only","work","with","a","single","gpu","because","it","runs","within","the","current","pytest","worker",".","For","multi-gpu","tests","use","TestDeepSpeedWithLauncher",".","Note",":","if","any","of","the","tests","of","this","class","get","run","there","will","be","at","least","one","gpu","occupied","by","them","until","this","pytest","worker","exits",".","This","is","because","the","gpu","memory","allocated","by","the","cuda-kernels","wo","n't","be","released","until","this","pytest","worker","exits",".","This","may","appear","as","some","run-away","tests","if","you","watch","`","nvidia-smi","`","while","other","tests","that","fork","new","processes","are","run",".","So","there","will","be","one","or","two","``","stale","''","processes","reported","in","`","nvidia-smi","`",".","This","is","not","a","bug","."]
code: def test_can_resume_training_normal(
      self, stage, dtype, optim, scheduler
  ):
      # adapted from TrainerIntegrationTest.test_can_resume_training
      # test normal resume for each stage separately, error-handling is tested in a different test
  
      # ToDo: Currently, hf_optim + hf_scheduler resumes with the correct states and
      # also has same losses for few steps but then slowly diverges. Need to figure it out.
      if optim == HF_OPTIM and scheduler == HF_SCHEDULER:
          return
  
      output_dir = self.get_auto_remove_tmp_dir("./xxx", after=False)
      ds_config_dict = self.get_config_dict(stage)
      if dtype == FP16:
          ds_config_dict["fp16"][
              "initial_scale_power"
          ] = 1  # force optimizer on the first step
      # XXX:
      if stage == ZERO3:
          ds_config_dict["zero_optimization"][
              "stage3_gather_16bit_weights_on_model_save"
          ] = True
  
      if optim == HF_OPTIM:
          del ds_config_dict["optimizer"]
  
      if scheduler == HF_SCHEDULER:
          del ds_config_dict["scheduler"]
  
      kwargs = {
          "output_dir": output_dir,
          "train_len": 128,
          "save_steps": 5,
          "learning_rate": 0.1,
          "deepspeed": ds_config_dict,
      }
      kwargs[dtype] = True
  
      with mockenv_context(**self.dist_env_1_gpu):
          trainer = get_regression_trainer(**kwargs)
          trainer.train()
          (a, b) = trainer.model.a.item(), trainer.model.b.item()
          state = dataclasses.asdict(trainer.state)
  
          checkpoint = os.path.join(output_dir, "checkpoint-5")
  
          # Reinitialize trainer
          trainer = get_regression_trainer(**kwargs)
  
          trainer.train(resume_from_checkpoint=checkpoint)
          (a1, b1) = trainer.model.a.item(), trainer.model.b.item()
          state1 = dataclasses.asdict(trainer.state)
          self.assertEqual(a, a1)
          self.assertEqual(b, b1)
          self.check_trainer_state_are_the_same(state, state1)
  
          # Now check with a later checkpoint that it also works when we span over one epoch
          checkpoint = os.path.join(output_dir, "checkpoint-15")
  
          # Reinitialize trainer and load model
          trainer = get_regression_trainer(**kwargs)
  
          trainer.train(resume_from_checkpoint=checkpoint)
          (a1, b1) = trainer.model.a.item(), trainer.model.b.item()
          state1 = dataclasses.asdict(trainer.state)
          self.assertEqual(a, a1)
          self.assertEqual(b, b1)
          self.check_trainer_state_are_the_same(state, state1)
  
imports: import dataclasses
  import io
  import itertools
  import json
  import os
  import unittest
  from copy import deepcopy
  from functools import partial
  import datasets
  from parameterized import parameterized
  import tests.trainer.test_trainer
  from tests.trainer.test_trainer import TrainerIntegrationCommon
  from transformers import AutoModel, TrainingArguments, is_torch_available, logging
  from transformers.integrations.deepspeed import HfDeepSpeedConfig, is_deepspeed_available, unset_hf_deepspeed_config
  from transformers.testing_utils import CaptureLogger, CaptureStd, CaptureStderr, LoggingLevel, TestCasePlus, backend_device_count, execute_subprocess_async, mockenv_context, require_deepspeed, require_optuna, require_torch_accelerator, require_torch_multi_accelerator, slow, torch_device
  from transformers.trainer_utils import get_last_checkpoint, set_seed
  from transformers.utils import SAFE_WEIGHTS_NAME, is_torch_bf16_available_on_device
label_desc: Use image node_id 14 for calling the TrainerIntegrationDeepSpeed obj's underlying member method code with example usage: obj.test_can_resume_training_normal(stage, dtype, optim, scheduler) without return types