code: def test_blocks_to_batches(block_size, drop_last):
      num_blocks = 5
      block_iter = block_generator(
          num_rows=block_size, num_blocks=num_blocks
      )
  
      batch_size = 3
      batch_iter = list(
          blocks_to_batches(
              block_iter, batch_size=batch_size, drop_last=drop_last
          )
      )
  
      if drop_last:
          for batch in batch_iter:
              assert len(batch.data) == batch_size
      else:
          full_batches = 0
          leftover_batches = 0
  
          dataset_size = block_size * num_blocks
          for batch in batch_iter:
              if len(batch.data) == batch_size:
                  full_batches += 1
              if len(batch.data) == (dataset_size % batch_size):
                  leftover_batches += 1
  
          assert leftover_batches == 1
          assert full_batches == (dataset_size // batch_size)
  
      assert [batch.batch_idx for batch in batch_iter] == list(
          range(len(batch_iter))
      )
  
imports: import threading
  import time
  import numpy
  import pandas
  import pyarrow
  import pytest
  import ray
  from ray.data._internal.block_batching.interfaces import Batch
  from ray.data._internal.block_batching.util import _calculate_ref_hits, blocks_to_batches, collate, finalize_batches, format_batches, resolve_block_refs
  from ray.data._internal.util import Queue, make_async_gen
label_desc: Use image node_id 3 for calling a global function with example usage: test_blocks_to_batches(block_size, drop_last) without return types