def test_model_checkpoint_no_extraneous_invocations(tmp_path):
    """Test to ensure that the model callback saves the checkpoints only once in distributed mode."""
    model = LogInTwoMethods()
    num_epochs = 4
    model_checkpoint = ModelCheckpointTestInvocations(monitor="early_stop_on", expected_count=num_epochs, save_top_k=-1)
    trainer = Trainer(
        strategy="ddp_spawn",
        accelerator="cpu",
        devices=2,
        default_root_dir=tmp_path,
        callbacks=[model_checkpoint],
        max_epochs=num_epochs,
    )
    trainer.fit(model)
    assert trainer.state.finished, f"Training failed with {trainer.state}"
Test to ensure that the model callback saves the checkpoints only once in distributed mode.
tests/tests_pytorch/checkpointing/test_model_checkpoint.py