code: def test_checkpoint_plugin_called(tmpdir):
      """Ensure that the custom checkpoint IO plugin and torch checkpoint IO plugin is called when saving/loading."""
      checkpoint_plugin = CustomCheckpointIO()
      checkpoint_plugin = MagicMock(
          wraps=checkpoint_plugin, spec=CustomCheckpointIO
      )
  
      ck = ModelCheckpoint(dirpath=tmpdir, save_last=True)
  
      model = BoringModel()
      trainer = Trainer(
          default_root_dir=tmpdir,
          accelerator="cpu",
          strategy=SingleDeviceStrategy(
              "cpu", checkpoint_io=checkpoint_plugin
          ),
          callbacks=ck,
          max_epochs=2,
          limit_train_batches=1,
          limit_val_batches=0,
          limit_test_batches=1,
      )
      trainer.fit(model)
  
      ckpt_files = {fn.name for fn in Path(tmpdir).glob("*.ckpt")}
      assert ckpt_files == {"epoch=1-step=2.ckpt", "last.ckpt"}
      assert (
          trainer.checkpoint_callback.best_model_path
          == tmpdir / "epoch=1-step=2.ckpt"
      )
      assert (
          trainer.checkpoint_callback.last_model_path
          == tmpdir / "last.ckpt"
      )
      assert checkpoint_plugin.save_checkpoint.call_count == 4
      assert checkpoint_plugin.remove_checkpoint.call_count == 1
  
      trainer.test(model, ckpt_path=ck.last_model_path)
      checkpoint_plugin.load_checkpoint.assert_called_with(
          tmpdir / "last.ckpt"
      )
  
      checkpoint_plugin.reset_mock()
      ck = ModelCheckpoint(dirpath=tmpdir, save_last=True)
  
      model = BoringModel()
      trainer = Trainer(
          default_root_dir=tmpdir,
          accelerator="cpu",
          strategy=SingleDeviceStrategy("cpu"),
          plugins=[checkpoint_plugin],
          callbacks=ck,
          max_epochs=2,
          limit_train_batches=1,
          limit_val_batches=0,
          limit_test_batches=1,
      )
      trainer.fit(model)
  
      ckpt_files = {fn.name for fn in Path(tmpdir).glob("*.ckpt")}
      assert ckpt_files == {
          "epoch=1-step=2.ckpt",
          "last.ckpt",
          "epoch=1-step=2-v1.ckpt",
          "last-v1.ckpt",
      }
      assert (
          trainer.checkpoint_callback.best_model_path
          == tmpdir / "epoch=1-step=2-v1.ckpt"
      )
      assert (
          trainer.checkpoint_callback.last_model_path
          == tmpdir / "last-v1.ckpt"
      )
      assert checkpoint_plugin.save_checkpoint.call_count == 4
      assert checkpoint_plugin.remove_checkpoint.call_count == 1
  
      trainer.test(model, ckpt_path=ck.last_model_path)
      checkpoint_plugin.load_checkpoint.assert_called_once()
      checkpoint_plugin.load_checkpoint.assert_called_with(
          tmpdir / "last-v1.ckpt"
      )
  
file_path: pytorch-lightning/tests/tests_pytorch/plugins/test_checkpoint_io_plugin.py
imports: import os
  from pathlib import Path
  from typing import Any, Dict, Optional
  from unittest.mock import MagicMock, Mock
  import torch
  from lightning.fabric.plugins import CheckpointIO, TorchCheckpointIO
  from lightning.fabric.utilities.types import _PATH
  from lightning.pytorch import Trainer
  from lightning.pytorch.callbacks import ModelCheckpoint
  from lightning.pytorch.demos.boring_classes import BoringModel
  from lightning.pytorch.plugins.io.async_plugin import AsyncCheckpointIO
  from lightning.pytorch.strategies import SingleDeviceStrategy
label_desc: Use image node_id 1 for calling a global function with example usage: test_checkpoint_plugin_called(tmpdir) without return types