docstr: Reduce the given `inputs` and verify the output matches `expect`.
  
  Args:
    inputs: a list of `Tensor` or `IndexedSlices`, where i-th value will be
      fed to i-th replica.
    expect: a `Tensor` or `IndexedSlices`. This should be the expected value
      for one replica.
    options: a `RunOpotions` instance.
docstr_tok: ["Reduce","the","given","`","inputs","`","and","verify","the","output","matches","`","expect","`",".","Args",":","inputs",":","a","list","of","`","Tensor","`","or","`","IndexedSlices","`",",","where","i-th","value","will","be","fed","to","i-th","replica",".","expect",":","a","`","Tensor","`","or","`","IndexedSlices","`",".","This","should","be","the","expected","value","for","one","replica",".","options",":","a","`","RunOpotions","`","instance","."]
code: def reduce_and_verify(self, inputs, expect, options):
      """Reduce the given `inputs` and verify the output matches `expect`.
  
      Args:
        inputs: a list of `Tensor` or `IndexedSlices`, where i-th value will be
          fed to i-th replica.
        expect: a `Tensor` or `IndexedSlices`. This should be the expected value
          for one replica.
        options: a `RunOpotions` instance.
      """
  
      def replica_fn():
          CollectiveReplicaLauncher._prefer_unique_instance_key = (
              options.prefer_unique_instance_key
          )
          collective, devices, pid = self.make_collective(
              options.num_processes, options.gpus_per_process
          )
  
          def reduce_fn():
              value_fn = lambda device_idx: inputs[
                  pid * len(devices) + device_idx
              ]
              per_replica_value = make_per_replica_value(
                  value_fn, devices
              )
              reduced_values = collective.reduce(
                  options.reduce_op,
                  per_replica_value,
                  per_replica_value,
                  options.communication_options,
              )
              if options.gpus_per_process > 1:
                  self.assertIsInstance(
                      reduced_values, value_lib.Mirrored
                  )
              reduced_values = self.as_list(reduced_values)
              self.assertAllEqual(
                  devices, [v.device for v in reduced_values]
              )
              return [ops.convert_to_tensor(v) for v in reduced_values]
  
          per_replica_expect = [ops.convert_to_tensor(expect)] * len(
              devices
          )
  
          if "eager" in options.mode:
              got = reduce_fn()
              self.assertAllClose(got, per_replica_expect)
  
          if "func_graph" in options.mode:
              got = def_function.function(reduce_fn)()
              self.assertAllClose(got, per_replica_expect)
  
      get_global_mpr(options.num_processes).run(replica_fn)
  
imports: import collections
  import os
  import threading
  import time
  from absl.testing import parameterized
  from tensorflow.core.protobuf import config_pb2
  from tensorflow.core.protobuf import tensorflow_server_pb2
  from tensorflow.python.distribute import cluster_resolver
  from tensorflow.python.distribute import collective_util
  from tensorflow.python.distribute import combinations
  from tensorflow.python.distribute import cross_device_ops
  from tensorflow.python.distribute import cross_device_utils
  from tensorflow.python.distribute import device_util
  from tensorflow.python.distribute import multi_process_runner
  from tensorflow.python.distribute import multi_worker_test_base
  from tensorflow.python.distribute import reduce_util
  from tensorflow.python.distribute import test_util
  from tensorflow.python.distribute import values
  from tensorflow.python.eager import context
  from tensorflow.python.eager import def_function
  from tensorflow.python.eager import test
  from tensorflow.python.framework import constant_op
  from tensorflow.python.framework import dtypes
  from tensorflow.python.framework import errors
  from tensorflow.python.framework import indexed_slices
  from tensorflow.python.framework import ops
  from tensorflow.python.framework import tensor
  from tensorflow.python.ops import array_ops
  from tensorflow.python.ops import collective_ops
  from tensorflow.python.ops import cond
  from tensorflow.python.ops import math_ops
  from tensorflow.python.util import nest
label_desc: Use image node_id 4 for calling the CollectiveOpsTest obj's underlying member method code with example usage: obj.reduce_and_verify(inputs, expect, options) and returns: unknown