def test_checkpoint_repeated_strategy_extended(tmp_path):
    """This test validates checkpoint can be called several times without increasing internally its global step if
    nothing run."""

    class ExtendedBoringModel(BoringModel):
        def validation_step(self, batch, batch_idx):
            loss = self.step(batch)
            self.log("val_loss", loss)
            return {"val_loss": loss}

    def assert_trainer_init(trainer):
        assert trainer.global_step == 0
        assert trainer.current_epoch == 0

    def get_last_checkpoint(ckpt_dir):
        last = sorted(ckpt_dir.iterdir())[-1]
        return str(last)

    def assert_checkpoint_content(ckpt_dir):
        chk = pl_load(get_last_checkpoint(ckpt_dir))
        # `-1` because this checkpoint is saved `on_train_epoch_end` which is considered part of the epoch so the
        # `current_epoch` count has not been increased yet
        assert chk["epoch"] == epochs - 1
        assert chk["global_step"] == 4

    def assert_checkpoint_log_dir(idx):
        lightning_logs = tmp_path / "lightning_logs"
        actual = [d.name for d in sorted(lightning_logs.iterdir())]
        assert actual == [f"version_{i}" for i in range(idx + 1)]
        actual = [d.name for d in sorted(ckpt_dir.iterdir())]
        assert len(actual) == epochs, actual

    ckpt_dir = tmp_path / "checkpoints"
    checkpoint_cb = ModelCheckpoint(dirpath=ckpt_dir, save_top_k=-1)
    epochs = 2
    limit_train_batches = 2
    trainer_config = {
        "default_root_dir": tmp_path,
        "max_epochs": epochs,
        "limit_train_batches": limit_train_batches,
        "limit_val_batches": 3,
        "limit_test_batches": 4,
        "callbacks": [checkpoint_cb],
        "logger": TensorBoardLogger(tmp_path),
    }
    trainer = Trainer(**trainer_config)
    assert_trainer_init(trainer)

    model = ExtendedBoringModel()
    trainer.fit(model)
    assert trainer.global_step == epochs * limit_train_batches
    assert trainer.current_epoch == epochs
    assert_checkpoint_log_dir(0)
    assert_checkpoint_content(ckpt_dir)

    trainer.validate(model)
    assert trainer.current_epoch == epochs

    trainer.test(model)
    assert trainer.current_epoch == epochs

    for idx in range(1, 5):
        chk = get_last_checkpoint(ckpt_dir)
        assert_checkpoint_content(ckpt_dir)

        # load from checkpoint
        trainer_config["logger"] = TensorBoardLogger(tmp_path)
        trainer = pl.Trainer(**trainer_config)
        assert_trainer_init(trainer)

        model = ExtendedBoringModel()

        trainer.test(model)
        assert_trainer_init(trainer)

        trainer.fit(model, ckpt_path=chk)
        assert trainer.global_step == epochs * limit_train_batches
        assert trainer.current_epoch == epochs
        assert trainer.fit_loop.epoch_progress.current.processed == epochs

        trainer.validate(model)
        assert trainer.global_step == epochs * limit_train_batches
        assert trainer.current_epoch == epochs
        assert trainer.fit_loop.epoch_progress.current.processed == epochs

        trainer.fit(model)
        assert trainer.global_step == epochs * limit_train_batches
        assert trainer.current_epoch == epochs
        assert trainer.fit_loop.epoch_progress.current.processed == epochs
        assert_checkpoint_log_dir(idx)
This test validates checkpoint can be called several times without increasing internally its global step if
    nothing run.
tests/tests_pytorch/checkpointing/test_model_checkpoint.py