def test_callbacks_state_fit_ckpt_path(tmp_path):
    """Test that resuming from a checkpoint restores callbacks that persist state."""
    dm = ClassifDataModule()
    model = ClassificationModel()
    callback_capture = CaptureCallbacksBeforeTraining()

    def get_trainer_args():
        checkpoint = ModelCheckpoint(dirpath=tmp_path, monitor="val_loss", save_last=True)
        assert checkpoint.best_model_path == ""
        assert checkpoint.best_model_score is None
        return {
            "default_root_dir": tmp_path,
            "limit_train_batches": 1,
            "limit_val_batches": 2,
            "max_epochs": 1,
            "logger": False,
            "callbacks": [checkpoint, callback_capture],
        }

    # initial training
    trainer = Trainer(**get_trainer_args())
    trainer.fit(model, datamodule=dm)

    callbacks_before_resume = deepcopy(trainer.callbacks)

    # resumed training
    trainer = Trainer(**get_trainer_args())
    trainer.fit(model, datamodule=dm, ckpt_path=str(tmp_path / "last.ckpt"))

    assert len(callbacks_before_resume) == len(callback_capture.callbacks)

    for before, after in zip(callbacks_before_resume, callback_capture.callbacks):
        if isinstance(before, ModelCheckpoint):
            for attribute in (
                "best_model_path",
                "best_model_score",
                "best_k_models",
                "kth_best_model_path",
                "kth_value",
                "last_model_path",
            ):
                assert getattr(before, attribute) == getattr(after, attribute), f"{attribute}"
Test that resuming from a checkpoint restores callbacks that persist state.
tests/tests_pytorch/models/test_restore.py