code: def test_fit_can_fail_during_validation(
      train_datasets, val_datasets, val_check_interval, tmpdir
  ):
      size, n_batches = 2, 4
      stop_batch = 1
      n_val_dataloaders = len(val_datasets)
      stop_dataloader = n_val_dataloaders - 1
  
      class TestModel(LightningModule):
          def __init__(self, should_fail):
              super().__init__()
              self.layer = torch.nn.Linear(size, 2)
              self.should_fail = should_fail
  
          def step(self, batch):
              return sum(self.layer(b).sum() for b in batch)
  
          def training_step(self, batch, batch_idx):
              return self.step(batch)
  
          def validation_step(self, batch, batch_idx, dataloader_idx=0):
              if (
                  self.should_fail
                  and dataloader_idx == stop_dataloader
                  and batch_idx == stop_batch
              ):
                  raise CustomException
              return self.step(batch)
  
          def configure_optimizers(self):
              return torch.optim.SGD(self.layer.parameters(), lr=0.1)
  
          def train_dataloader(self):
              return [
                  DataLoader(cls(size, n_batches))
                  for cls in train_datasets
              ]
  
          def val_dataloader(self):
              return [
                  DataLoader(cls(size, n_batches))
                  for cls in val_datasets
              ]
  
      model = TestModel(False)
      trainer = Trainer(
          default_root_dir=tmpdir,
          max_epochs=1,
          val_check_interval=val_check_interval,
          num_sanity_val_steps=0,
          enable_progress_bar=False,
          callbacks=OnExceptionCheckpoint(tmpdir),
      )
      trainer.fit(model)
  
      ckpt_path = os.path.join(tmpdir, "on_exception.ckpt")
      assert not os.path.exists(ckpt_path), "Shouldn't have failed"
      state_dict = trainer.fit_loop.state_dict()
      expected_global_step = trainer.global_step
  
      assert state_dict["epoch_loop.batch_progress"] == {
          "total": {
              "ready": n_batches,
              "started": n_batches,
              "processed": n_batches,
              "completed": n_batches,
          },
          "current": {
              "ready": n_batches,
              "started": n_batches,
              "processed": n_batches,
              "completed": n_batches,
          },
          "is_last_batch": True,
      }
  
      val_per_epoch = int(1 // val_check_interval)
      assert state_dict["epoch_loop.val_loop.batch_progress"] == {
          "total": {
              "ready": n_val_dataloaders * val_per_epoch * n_batches,
              "started": n_val_dataloaders * val_per_epoch * n_batches,
              "processed": n_val_dataloaders
              * val_per_epoch
              * n_batches,
              "completed": n_val_dataloaders
              * val_per_epoch
              * n_batches,
          },
          "current": {
              "ready": n_val_dataloaders * n_batches,
              "started": n_val_dataloaders * n_batches,
              "processed": n_val_dataloaders * n_batches,
              "completed": n_val_dataloaders * n_batches,
          },
          "is_last_batch": True,
      }
  
      model = TestModel(True)
      trainer = Trainer(
          default_root_dir=tmpdir,
          max_epochs=1,
          val_check_interval=val_check_interval,
          num_sanity_val_steps=0,
          enable_progress_bar=False,
          callbacks=OnExceptionCheckpoint(tmpdir),
      )
      with pytest.raises(CustomException):
          # will stop during validation
          trainer.fit(model)
  
      assert os.path.exists(ckpt_path)
      checkpoint = torch.load(ckpt_path)["loops"]["fit_loop"]
  
      per_val_train_batches = int(n_batches * val_check_interval)
      assert checkpoint["epoch_loop.batch_progress"] == {
          "total": {
              "ready": per_val_train_batches,
              "started": per_val_train_batches,
              "processed": per_val_train_batches,
              "completed": per_val_train_batches,
          },
          "current": {
              "ready": per_val_train_batches,
              "started": per_val_train_batches,
              "processed": per_val_train_batches,
              "completed": per_val_train_batches,
          },
          "is_last_batch": val_check_interval == 1,
      }
  
      val_batch_progress = "epoch_loop.val_loop.batch_progress"
      # "nb_": non-breaking
      nb_total_val_batch = stop_dataloader * n_batches
      assert checkpoint[val_batch_progress] == {
          "total": {
              "ready": nb_total_val_batch + stop_batch + 1,
              "started": nb_total_val_batch + stop_batch + 1,
              "processed": nb_total_val_batch + stop_batch,
              "completed": nb_total_val_batch + stop_batch,
          },
          "current": {
              "ready": nb_total_val_batch + stop_batch + 1,
              "started": nb_total_val_batch + stop_batch + 1,
              "processed": nb_total_val_batch + stop_batch,
              "completed": nb_total_val_batch + stop_batch,
          },
          "is_last_batch": False,
      }
  
      model = TestModel(False)
      trainer = Trainer(
          default_root_dir=tmpdir,
          max_epochs=1,
          val_check_interval=val_check_interval,
          enable_progress_bar=False,
      )
      trainer.fit(model, ckpt_path=ckpt_path)
  
      assert trainer.global_step == expected_global_step
  
      state_dict_after_restart = trainer.fit_loop.state_dict()
  
      # should get the same values as in the run that did not fail
      # totals are increased by 1 (the failed batch which never completed)
      expected = state_dict.copy()
  
      assert (
          state_dict_after_restart["epoch_loop.batch_progress"]
          == expected["epoch_loop.batch_progress"]
      )
  
      expected[val_batch_progress]["total"]["ready"] += 1
      expected[val_batch_progress]["total"]["started"] += 1
      assert (
          state_dict_after_restart[val_batch_progress]
          == expected[val_batch_progress]
      )
  
imports: import os
  from copy import deepcopy
  from dataclasses import dataclass
  from typing import Dict, Iterator
  from unittest.mock import ANY, Mock
  import pytest
  import torch
  from lightning.pytorch import LightningModule, Trainer
  from lightning.pytorch.callbacks import Callback, ModelCheckpoint, OnExceptionCheckpoint
  from lightning.pytorch.demos.boring_classes import BoringModel, RandomDataset
  from lightning.pytorch.loops import _Loop
  from lightning.pytorch.loops.progress import _BaseProgress
  from torch.utils.data.dataloader import DataLoader, _MultiProcessingDataLoaderIter
  from tests_pytorch.helpers.runif import RunIf
label_desc: Use image node_id 8 for calling a global function with example usage: test_fit_can_fail_during_validation(train_datasets, val_datasets, val_check_interval, tmpdir) and returns: sum, self, self, torch, unknown, unknown