def test_deepspeed_multi_save_same_filepath(tmp_path):
    """Test that verifies that deepspeed saves only latest checkpoint in the specified path and deletes the old sharded
    checkpoints."""

    class CustomModel(BoringModel):
        def training_step(self, *args, **kwargs):
            self.log("grank", self.global_rank)
            return super().training_step(*args, **kwargs)

    model = CustomModel()
    trainer = Trainer(
        default_root_dir=tmp_path,
        strategy="deepspeed",
        accelerator="gpu",
        devices=2,
        callbacks=[ModelCheckpoint(filename="{epoch}_{step}_{grank}", save_top_k=1)],
        limit_train_batches=1,
        limit_val_batches=0,
        num_sanity_val_steps=0,
        max_epochs=2,
        enable_progress_bar=False,
        enable_model_summary=False,
    )
    trainer.fit(model)

    filepath = "epoch=1_step=2_grank=0.0.ckpt"
    expected = {filepath}
    assert expected == set(os.listdir(trainer.checkpoint_callback.dirpath))

    ckpt_path = os.path.join(trainer.checkpoint_callback.dirpath, filepath)
    expected = {"latest", "zero_to_fp32.py", "checkpoint"}
    assert expected == set(os.listdir(ckpt_path))
Test that verifies that deepspeed saves only latest checkpoint in the specified path and deletes the old sharded
    checkpoints.
tests/tests_pytorch/strategies/test_deepspeed.py