code: def test_trainer_checkpoint_callback_1(
      model_and_optimizers: TrainerParameters, driver, device
  ):
      for version in [0, 1]:
          for only_state_dict in [True, False]:
              try:
                  path = Path.cwd().joinpath(f"test_model_checkpoint")
                  path.mkdir(exist_ok=True, parents=True)
  
                  if version == 0:
                      callbacks = [
                          CheckpointCallback(
                              folder=path,
                              every_n_epochs=7,
                              every_n_batches=123,
                              last=False,
                              on_exceptions=None,
                              topk=0,
                              monitor=None,
                              only_state_dict=only_state_dict,
                              save_object="trainer",
                          )
                      ]
                  elif version == 1:
                      callbacks = [
                          CheckpointCallback(
                              folder=path,
                              every_n_epochs=None,
                              every_n_batches=None,
                              last=True,
                              on_exceptions=None,
                              topk=2,
                              monitor="acc",
                              only_state_dict=only_state_dict,
                              save_object="trainer",
                          )
                      ]
  
                  trainer = Trainer(
                      model=model_and_optimizers.model,
                      driver=driver,
                      device=device,
                      optimizers=model_and_optimizers.optimizers,
                      train_dataloader=model_and_optimizers.train_dataloader,
                      evaluate_dataloaders=model_and_optimizers.evaluate_dataloaders,
                      input_mapping=model_and_optimizers.input_mapping,
                      output_mapping=model_and_optimizers.output_mapping,
                      metrics=model_and_optimizers.metrics,
                      n_epochs=10,
                      callbacks=callbacks,
                      output_from_new_proc="all",
                  )
  
                  trainer.run()
  
                  all_saved_model_paths = {
                      w.name: w
                      for w in path.joinpath(
                          os.environ[FASTNLP_LAUNCH_TIME]
                      ).iterdir()
                  }
                  # 检查生成保存模型文件的数量是不是正确的；
                  if version == 0:
                      if not isinstance(device, list):
                          assert (
                              "trainer-epoch_7" in all_saved_model_paths
                          )
                          assert (
                              "trainer-epoch_4-batch_123"
                              in all_saved_model_paths
                          )
  
                          epoch_save_path = all_saved_model_paths[
                              "trainer-epoch_7"
                          ]
                          step_save_path = all_saved_model_paths[
                              "trainer-epoch_4-batch_123"
                          ]
  
                          assert len(all_saved_model_paths) == 3
                      # ddp 下的文件名不同，因为同样的数据，ddp 用了更少的步数跑完；
                      else:
                          assert (
                              "trainer-epoch_7" in all_saved_model_paths
                          )
                          assert (
                              "trainer-epoch_9-batch_123"
                              in all_saved_model_paths
                          )
  
                          epoch_save_path = all_saved_model_paths[
                              "trainer-epoch_7"
                          ]
                          step_save_path = all_saved_model_paths[
                              "trainer-epoch_9-batch_123"
                          ]
  
                          assert len(all_saved_model_paths) == 2
                      all_state_dicts = [
                          epoch_save_path,
                          step_save_path,
                      ]
  
                  elif version == 1:
                      pattern = re.compile(
                          "trainer-epoch_[0-9]+-batch_[0-9]+-[a-zA-Z#]+_[0-9]*.?[0-9]*"
                      )
  
                      # all_saved_model_paths = {w.name: w for w in path.joinpath(os.environ[FASTNLP_LAUNCH_TIME]).iterdir()}
                      if not isinstance(device, list):
                          assert "trainer-last" in all_saved_model_paths
                          aLL_topk_folders = []
                          for each_folder_name in all_saved_model_paths:
                              each_folder_name = pattern.findall(
                                  each_folder_name
                              )
                              if len(each_folder_name) != 0:
                                  aLL_topk_folders.append(
                                      each_folder_name[0]
                                  )
                          assert len(aLL_topk_folders) == 2
  
                          last_save_path = all_saved_model_paths[
                              "trainer-last"
                          ]
                          topk_save_path = all_saved_model_paths[
                              aLL_topk_folders[0]
                          ]
  
                          assert len(all_saved_model_paths) == 3
                      # ddp 下的文件名不同，因为同样的数据，ddp 用了更少的步数跑完；
                      else:
                          assert "trainer-last" in all_saved_model_paths
  
                          aLL_topk_folders = []
                          for each_folder_name in all_saved_model_paths:
                              each_folder_name = pattern.findall(
                                  each_folder_name
                              )
                              if len(each_folder_name) != 0:
                                  aLL_topk_folders.append(
                                      each_folder_name[0]
                                  )
                          assert len(aLL_topk_folders) == 2
  
                          last_save_path = all_saved_model_paths[
                              "trainer-last"
                          ]
                          topk_save_path = all_saved_model_paths[
                              aLL_topk_folders[0]
                          ]
  
                          assert len(all_saved_model_paths) == 3
  
                      all_state_dicts = [last_save_path, topk_save_path]
  
                  for folder in all_state_dicts:
                      trainer = Trainer(
                          model=model_and_optimizers.model,
                          driver=driver,
                          device=device,
                          optimizers=model_and_optimizers.optimizers,
                          train_dataloader=model_and_optimizers.train_dataloader,
                          evaluate_dataloaders=model_and_optimizers.evaluate_dataloaders,
                          input_mapping=model_and_optimizers.input_mapping,
                          output_mapping=model_and_optimizers.output_mapping,
                          metrics=model_and_optimizers.metrics,
                          n_epochs=13,
                          output_from_new_proc="all",
                      )
                      trainer.load_checkpoint(
                          folder, only_state_dict=only_state_dict
                      )
  
                      trainer.run()
                      trainer.driver.barrier()
  
              finally:
                  rank_zero_rm(path)
  
      if dist.is_initialized():
          dist.destroy_process_group()
  
imports: import os
  import pytest
  from typing import Any
  from dataclasses import dataclass
  from pathlib import Path
  import re
  import time
  from fastNLP.core.callbacks.checkpoint_callback import CheckpointCallback
  from fastNLP.core.controllers.trainer import Trainer
  from fastNLP import Evaluator
  from fastNLP.envs import FASTNLP_LAUNCH_TIME, FASTNLP_DISTRIBUTED_CHECK
  from tests.helpers.utils import magic_argv_env_context
  from fastNLP.envs.distributed import rank_zero_rm
  from tests.helpers.models.torch_model import TorchNormalModel_Classification_1
  from tests.helpers.datasets.torch_data import TorchArgMaxDataset
  from tests.helpers.utils import Capturing
  from fastNLP.core.log import logger
  from fastNLP.envs.imports import _NEED_IMPORT_TORCH
label_desc: Use image node_id 4 for calling a global function with example usage: test_trainer_checkpoint_callback_1(model_and_optimizers, driver, device) without return types