code: def test_torch_batch_process_preemption(
      mock_reduce_metrics: unittest.mock.MagicMock,
      mock_synchronize_and_checkpoint: unittest.mock.MagicMock,
      mock_initialize_default_inference_context: unittest.mock.MagicMock,
  ) -> None:
      with test_util.set_mock_cluster_info(DEFAULT_ADDRS, 0, 1):
          # Mock core context to set preemption result
          mock_initialize_default_inference_context.return_value = (
              _get_core_context(should_preempt_results=[False, True])
          )
          index_dataset = IndexData(100)
  
          MyProcessorCLS = unittest.mock.Mock()
          my_processor_instance = unittest.mock.Mock()
          MyProcessorCLS.return_value = my_processor_instance
  
          experimental.torch_batch_process(
              dataset=index_dataset,
              batch_processor_cls=MyProcessorCLS,
              batch_size=10,
              checkpoint_interval=2,
          )
  
          # Without preemption, we should iterate for 10 batches (100 / 10).
          # We check preemption every two batches, and at the second check, preemption is true
          # Thus, we expect to iterate for 4 batches total
  
          assert my_processor_instance.process_batch.call_count == 4
          # on_finish should not be called as the job was preempted before completion
          assert my_processor_instance.on_finish.call_count == 0
          # Before preempting, we run reduce metrics once
          # If there is no metrics reducer, this call is a no-op
          assert mock_reduce_metrics.call_count == 1
  
imports: import math
  import unittest.mock
  from typing import Any, Dict, List, Optional
  import pytest
  import torch
  from torch.utils import data
  from determined import core, pytorch
  from determined.pytorch import experimental
  from tests.launch import test_util
label_desc: Use image node_id 10 for calling a global function with example usage: test_torch_batch_process_preemption(mock_reduce_metrics, mock_synchronize_and_checkpoint, mock_initialize_default_inference_context) without return types