code: def rec_metric_value_test_helper(
      target_clazz: Type[RecMetric],
      target_compute_mode: RecComputeMode,
      test_clazz: Optional[Type[TestMetric]],
      fused_update_limit: int,
      compute_on_all_ranks: bool,
      should_validate_update: bool,
      world_size: int,
      my_rank: int,
      task_names: List[str],
      batch_size: int = BATCH_SIZE,
      nsteps: int = NSTEPS,
      batch_window_size: int = BATCH_WINDOW_SIZE,
      is_time_dependent: bool = False,
      time_dependent_metric: Optional[
          Dict[Type[RecMetric], str]
      ] = None,
      n_classes: Optional[int] = None,
      zero_weights: bool = False,
      **kwargs: Any,
  ) -> Tuple[
      Dict[str, torch.Tensor], Tuple[Dict[str, torch.Tensor], ...]
  ]:
      tasks = gen_test_tasks(task_names)
      model_outs = []
      for _ in range(nsteps):
          weight_value: Optional[torch.Tensor] = None
          if zero_weights:
              weight_value = torch.zeros(batch_size)
  
          _model_outs = [
              gen_test_batch(
                  label_name=task.label_name,
                  prediction_name=task.prediction_name,
                  weight_name=task.weight_name,
                  batch_size=batch_size,
                  n_classes=n_classes,
                  weight_value=weight_value,
              )
              for task in tasks
          ]
          model_outs.append(
              {k: v for d in _model_outs for k, v in d.items()}
          )
  
      def get_target_rec_metric_value(
          model_outs: List[Dict[str, torch.Tensor]],
          tasks: List[RecTaskInfo],
          timestamps: Optional[List[float]] = None,
          time_mock: Optional[Mock] = None,
          **kwargs: Any,
      ) -> Dict[str, torch.Tensor]:
          window_size = world_size * batch_size * batch_window_size
          if n_classes:
              kwargs["number_of_classes"] = n_classes
          if zero_weights:
              kwargs["allow_missing_label_with_zero_weight"] = True
  
          target_metric_obj = target_clazz(
              world_size=world_size,
              my_rank=my_rank,
              batch_size=batch_size,
              tasks=tasks,
              compute_mode=target_compute_mode,
              window_size=window_size,
              fused_update_limit=fused_update_limit,
              compute_on_all_ranks=compute_on_all_ranks,
              should_validate_update=should_validate_update,
              **kwargs,
          )
          for i in range(nsteps):
              (
                  labels,
                  predictions,
                  weights,
                  _,
              ) = parse_task_model_outputs(tasks, model_outs[i])
              if (
                  target_compute_mode
                  == RecComputeMode.FUSED_TASKS_COMPUTATION
              ):
                  labels = torch.stack(list(labels.values()))
                  predictions = torch.stack(list(predictions.values()))
                  weights = torch.stack(list(weights.values()))
  
              if timestamps is not None:
                  time_mock.return_value = timestamps[i]
              target_metric_obj.update(
                  predictions=predictions,
                  labels=labels,
                  weights=weights,
              )
          result_metrics = target_metric_obj.compute()
          result_metrics.update(target_metric_obj.local_compute())
          return result_metrics
  
      def get_test_rec_metric_value(
          model_outs: List[Dict[str, torch.Tensor]],
          tasks: List[RecTaskInfo],
          timestamps: Optional[List[float]] = None,
      ) -> TestRecMetricOutput:
          test_metrics: TestRecMetricOutput = ({}, {}, {}, {})
          if test_clazz is not None:
              # pyre-ignore[45]: Cannot instantiate abstract class `TestMetric`.
              test_metric_obj = test_clazz(world_size, tasks)
              test_metrics = test_metric_obj.compute(
                  model_outs, nsteps, batch_window_size, timestamps
              )
          return test_metrics
  
      if is_time_dependent:
          timestamps: Optional[List[float]] = (
              gen_test_timestamps(nsteps) if is_time_dependent else None
          )
          assert time_dependent_metric is not None  # avoid typing issue
          time_dependent_target_clazz_path = time_dependent_metric[
              target_clazz
          ]
          with patch(
              time_dependent_target_clazz_path + ".time.monotonic"
          ) as time_mock:
              result_metrics = get_target_rec_metric_value(
                  model_outs, tasks, timestamps, time_mock, **kwargs
              )
          test_metrics = get_test_rec_metric_value(
              model_outs, tasks, timestamps
          )
      else:
          result_metrics = get_target_rec_metric_value(
              model_outs, tasks, **kwargs
          )
          test_metrics = get_test_rec_metric_value(model_outs, tasks)
  
      return result_metrics, test_metrics
  
imports: import abc
  import os
  import random
  import tempfile
  import uuid
  from typing import Any, Callable, Dict, List, Optional, Tuple, Type
  from unittest.mock import Mock, patch
  import torch
  import torch.distributed
  import torch.distributed.launcher
  from torchrec.metrics.auc import AUCMetric
  from torchrec.metrics.auprc import AUPRCMetric
  from torchrec.metrics.model_utils import parse_task_model_outputs
  from torchrec.metrics.rec_metric import RecComputeMode, RecMetric, RecTaskInfo
label_desc: Use image node_id 4 for calling a global function with example usage: rec_metric_value_test_helper(target_clazz, target_compute_mode, test_clazz, fused_update_limit, compute_on_all_ranks, should_validate_update, world_size, my_rank, task_names, batch_size, nsteps, batch_window_size, is_time_dependent, time_dependent_metric, n_classes, zero_weights) and returns: result_metrics,  test_metrics, result_metrics, test_metrics