code: def _validate(
      cls,
      actual_output: torch.Tensor,
      expected_output: torch.Tensor,
      input: torch.Tensor,
      world_size: int,
  ) -> None:
      torch.testing.assert_close(
          actual_output.cpu().detach(), expected_output.cpu().detach()
      )
      torch.testing.assert_close(
          # pyre-fixme[16]: Optional type has no attribute `cpu`.
          input.grad.cpu().detach(),
          torch.ones(input.size()),
      )
  
imports: import itertools
  import random
  import unittest
  from typing import Dict, Generator, Iterable, List, Optional, Tuple, TypeVar, Union
  import hypothesis.strategies
  import torch
  import torch.distributed
  from hypothesis import given, settings
  from torchrec.distributed.dist_data import _get_recat, KJTAllToAll, KJTAllToAllSplitsAwaitable, PooledEmbeddingsAllGather, PooledEmbeddingsAllToAll, PooledEmbeddingsReduceScatter, SequenceEmbeddingsAllToAll, VariableBatchPooledEmbeddingsAllToAll
  from torchrec.distributed.fbgemm_qcomm_codec import CommType, get_qcomm_codecs, QCommsConfig
  from torchrec.distributed.test_utils.multi_process import MultiProcessTestBase
  from torchrec.sparse.jagged_tensor import KeyedJaggedTensor
label_desc: Use image node_id 1 for calling the PooledEmbeddingsAllGatherTest obj's underlying member method code with example usage: obj._validate(cls, actual_output, expected_output, input, world_size) without return types