code: def test_reduce_with_collectives_matches_reduce_without_collectives(
      self, reduce_op, distributed_op, skip_on_eager, axes
  ):
      if skip_on_eager and (tf.executing_eagerly() and TF_MODE):
          self.skipTest("Not supported in Eager.")
      x = tf.reshape(
          tf.range(test_lib.NUM_DEVICES * 6.0) / 5.0,
          [test_lib.NUM_DEVICES, 3, 2],
      )
  
      def run(x):
          return distributed_op(x, axis=pos_axes, named_axis=named_axes)
  
      def distributed_run(x):
          return self.per_replica_to_tensor(
              self.strategy_run(run, (self.shard_values(x),))
          )
  
      if axes is None:
          pos_axes = list(range(2))
          named_axes = [self.axis_name]
      else:
          axes = [axes] if not isinstance(axes, list) else axes
          pos_axes = [x - 1 for x in axes if x != 0]
          named_axes = [self.axis_name] if 0 in axes else []
      reduce_out = reduce_op(x, axis=axes)
      dist_out = distributed_run(x)
      # If we reduce over the 0 dimension, it will still be present in the
      # distributed op
      if axes is None or 0 in axes:
          for i in range(test_lib.NUM_DEVICES):
              self.assertAllClose(reduce_out, dist_out[i])
      else:
          self.assertAllClose(reduce_out, dist_out)
  
imports: import functools
  import itertools
  from absl.testing import parameterized
  import tensorflow.compat.v2
  from tensorflow_probability.python.distributions import bernoulli
  from tensorflow_probability.python.distributions import normal
  from tensorflow_probability.python.internal import distribute_lib
  from tensorflow_probability.python.internal import distribute_test_lib
  from tensorflow_probability.python.internal import samplers
  from tensorflow_probability.python.internal import test_util
  from tensorflow_probability.python.math import gradient
label_desc: Use image node_id 3 for calling the CollectiveTest obj's underlying member method code with example usage: obj.test_reduce_with_collectives_matches_reduce_without_collectives(reduce_op, distributed_op, skip_on_eager, axes) and returns: distributed_op, self