code: def _class_test(
      rank: int,
      worldsize: int,
      preds: TEXT_METRIC_INPUT,
      targets: TEXT_METRIC_INPUT,
      metric_class: Metric,
      ref_metric: Callable,
      dist_sync_on_step: bool,
      metric_args: Optional[dict] = None,
      check_dist_sync_on_step: bool = True,
      check_batch: bool = True,
      atol: float = 1e-8,
      device: str = "cpu",
      fragment_kwargs: bool = False,
      check_scriptable: bool = True,
      key: Optional[str] = None,
      ignore_order: Optional[bool] = None,
      **kwargs_update: Any,
  ):
      """Comparison between class metric and reference metric.
  
      Args:
          rank: rank of current process
          worldsize: number of processes
          preds: Sequence of predicted tokens or predicted sentences
          targets: Sequence of target tokens or target sentences
          metric_class: metric class that should be tested
          ref_metric: callable function that is used for comparison
          dist_sync_on_step: bool, if true will synchronize metric state across
              processes at each ``forward()``
          metric_args: dict with additional arguments used for class initialization
          check_dist_sync_on_step: bool, if true will check if the metric is also correctly
              calculated per batch per device (and not just at the end)
          check_batch: bool, if true will check if the metric is also correctly
              calculated across devices for each batch (and not just at the end)
          atol: absolute tolerance used for comparison of results
          device: determine which device to run on, either 'cuda' or 'cpu'
          fragment_kwargs: whether tensors in kwargs should be divided as `preds` and `targets` among processes
          check_scriptable: bool indicating if metric should also be tested if it can be scripted
          key: The key passed onto the `_assert_allclose` to compare the respective metric from the Dict output against
              the ref_metric.
          ignore_order: Ignore order of prediction across processes when DDP is used.
          kwargs_update: Additional keyword arguments that will be passed with preds and
              targets when running update on the metric.
  
      """
      if not metric_args:
          metric_args = {}
  
      # Instantiate metric
      metric = metric_class(
          dist_sync_on_step=dist_sync_on_step, **metric_args
      )
  
      # check that the metric is scriptable
      if check_scriptable:
          torch.jit.script(metric)
  
      # move to device
      metric = metric.to(device)
      kwargs_update = {
          k: v.to(device) if isinstance(v, Tensor) else v
          for k, v in kwargs_update.items()
      }
  
      # verify metrics work after being loaded from pickled state
      pickled_metric = pickle.dumps(metric)
      metric = pickle.loads(pickled_metric)
  
      for i in range(rank, NUM_BATCHES, worldsize):
          batch_kwargs_update = {
              k: v[i] if isinstance(v, Tensor) else v
              for k, v in kwargs_update.items()
          }
          batch_result = metric(
              preds[i], targets[i], **batch_kwargs_update
          )
  
          if (
              metric.dist_sync_on_step
              and check_dist_sync_on_step
              and rank == 0
          ):
              # Concatenation of Sequence of strings
              ddp_preds = type(preds)()
              ddp_targets = type(targets)()
              for r in range(worldsize):
                  ddp_preds = ddp_preds + preds[i + r]
                  ddp_targets = ddp_targets + targets[i + r]
              ddp_kwargs_upd = {
                  k: torch.cat(
                      [v[i + r] for r in range(worldsize)]
                  ).cpu()
                  if isinstance(v, Tensor)
                  else v
                  for k, v in (
                      kwargs_update
                      if fragment_kwargs
                      else batch_kwargs_update
                  ).items()
              }
  
              sk_batch_result = ref_metric(
                  ddp_preds, ddp_targets, **ddp_kwargs_upd
              )
              if ignore_order:
                  _assert_all_close_regardless_of_order(
                      batch_result, sk_batch_result, atol=atol, key=key
                  )
              else:
                  _assert_allclose(
                      batch_result, sk_batch_result, atol=atol, key=key
                  )
  
          elif check_batch and not metric.dist_sync_on_step:
              batch_kwargs_update = {
                  k: v.cpu() if isinstance(v, Tensor) else v
                  for k, v in (
                      batch_kwargs_update
                      if fragment_kwargs
                      else kwargs_update
                  ).items()
              }
              sk_batch_result = ref_metric(
                  preds[i], targets[i], **batch_kwargs_update
              )
              if ignore_order:
                  _assert_all_close_regardless_of_order(
                      batch_result, sk_batch_result, atol=atol, key=key
                  )
              else:
                  _assert_allclose(
                      batch_result, sk_batch_result, atol=atol, key=key
                  )
  
      # check that metrics are hashable
      assert hash(metric)
  
      # check on all batches on all ranks
      result = metric.compute()
      _assert_tensor(result, key=key)
  
      # Concatenation of Sequence of strings
      total_preds = type(preds)()
      total_targets = type(targets)()
      for i in range(NUM_BATCHES):
          total_preds = total_preds + preds[i]
          total_targets = total_targets + targets[i]
      total_kwargs_update = {
          k: torch.cat([v[i] for i in range(NUM_BATCHES)]).cpu()
          if isinstance(v, Tensor)
          else v
          for k, v in kwargs_update.items()
      }
      sk_result = ref_metric(
          total_preds, total_targets, **total_kwargs_update
      )
      # assert after aggregation
      if ignore_order:
          _assert_all_close_regardless_of_order(
              result, sk_result, atol=atol, key=key
          )
      else:
          _assert_allclose(result, sk_result, atol=atol, key=key)
  
imports: import pickle
  import sys
  from functools import partial, wraps
  from typing import Any, Callable, Dict, Optional, Sequence, Union
  import numpy
  import pytest
  import torch
  from torch import Tensor
  from torchmetrics import Metric
  from unittests import NUM_PROCESSES
  from unittests.helpers.testers import MetricTester, _assert_allclose, _assert_requires_grad, _assert_tensor
label_desc: Use image node_id 2 for calling a global function with example usage: _class_test(rank, worldsize, preds, targets, metric_class, ref_metric, dist_sync_on_step, metric_args, check_dist_sync_on_step, check_batch, atol, device, fragment_kwargs, check_scriptable, key, ignore_order) without return types