code: def test_collect_rank_zero_results(
      trainer_fn, fake_node_rank, fake_local_rank, tmpdir
  ):
      """Tests that the spawn strategy transfers the new weights to the main process and deletes the temporary file."""
      model = Mock(wraps=BoringModel(), spec=BoringModel)
      fake_global_rank = 2 * fake_node_rank + fake_local_rank
  
      cluster_environment = Mock(spec=ClusterEnvironment)
      cluster_environment.world_size.return_value = 4
      cluster_environment.node_rank.return_value = fake_node_rank
      cluster_environment.local_rank.return_value = fake_local_rank
      cluster_environment.global_rank.return_value = fake_global_rank
  
      strategy = DDPStrategy(
          cluster_environment=cluster_environment, start_method="spawn"
      )
      strategy._local_rank = fake_local_rank
  
      launcher = _MultiProcessingLauncher(strategy=strategy)
      trainer = Trainer(
          accelerator="cpu", default_root_dir=tmpdir, strategy=strategy
      )
  
      assert strategy.node_rank == fake_node_rank
      assert strategy.local_rank == fake_local_rank
      assert strategy.global_rank == fake_global_rank
  
      trainer.strategy.connect(model)
      trainer.state.fn = (
          trainer_fn  # pretend we are in a particular trainer state
      )
  
      spawn_output = launcher._collect_rank_zero_results(trainer, {})
  
      model.state_dict.assert_called_once()
      is_fitting = trainer_fn == TrainerFn.FITTING
      if strategy.local_rank == 0:
          # on local rank 0 (each node), we expect a temp checkpoint (when fitting)
          assert not is_fitting or spawn_output.weights_path.endswith(
              ".temp.ckpt"
          )
          assert not is_fitting or os.path.isfile(
              spawn_output.weights_path
          )
          assert is_fitting or spawn_output.weights_path is None
      else:
          # all other ranks don't have outputs (rank 0 needs to handle the output)
          assert spawn_output is None
  
imports: import os
  from multiprocessing import Process
  from unittest import mock
  from unittest.mock import ANY, Mock, call, patch
  import pytest
  import torch
  from lightning.fabric.plugins import ClusterEnvironment
  from lightning.pytorch import Trainer
  from lightning.pytorch.demos.boring_classes import BoringModel
  from lightning.pytorch.strategies import DDPStrategy
  from lightning.pytorch.strategies.launchers.multiprocessing import _GlobalStateSnapshot, _MultiProcessingLauncher
  from lightning.pytorch.trainer.states import TrainerFn
  from tests_pytorch.helpers.runif import RunIf
label_desc: Use image node_id 5 for calling a global function with example usage: test_collect_rank_zero_results(trainer_fn, fake_node_rank, fake_local_rank, tmpdir) without return types