code: def testAssignSignature(
      self, distribution, synchronization, aggregation
  ):
      # This test verifies assign*() can be called in the same way as normal
      # variables.
      with distribution.scope():
          v = variables_lib.Variable(
              0.0,
              synchronization=synchronization,
              aggregation=aggregation,
          )
  
          def assign():
              one = constant_op.constant(1.0)
              v.assign(one, True, "assign", False)
              # TODO(b/154017756): SyncOnReadVariable.assign() doesn't support passing
              # value as a keyword argument.
              v.assign(
                  one, use_locking=True, name="assign", read_value=False
              )
              v.assign_add(one, True, "assign", False)
              v.assign_add(
                  one, use_locking=True, name="assign", read_value=False
              )
              v.assign_sub(one, True, "assign", False)
              v.assign_sub(
                  one, use_locking=True, name="assign", read_value=False
              )
              # Return something for graph mode to fetch.
              return constant_op.constant(1)
  
          self.evaluate(variables_lib.global_variables_initializer())
          if not (
              synchronization
              == variables_lib.VariableSynchronization.ON_READ
              and aggregation == variables_lib.VariableAggregation.SUM
          ):
              self.evaluate(
                  distribution.experimental_local_results(assign())
              )
          if not (
              isinstance(
                  distribution.extended, tpu_strategy.TPUExtended
              )
              and context.executing_eagerly()
          ):
              self.evaluate(
                  distribution.experimental_local_results(
                      distribution.run(assign)
                  )
              )
  
imports: import copy
  import os
  from absl.testing import parameterized
  from tensorflow.python.checkpoint import checkpoint
  from tensorflow.python.checkpoint import checkpoint_options
  from tensorflow.python.distribute import collective_all_reduce_strategy
  from tensorflow.python.distribute import combinations
  from tensorflow.python.distribute import distribute_lib
  from tensorflow.python.distribute import distribute_utils
  from tensorflow.python.distribute import packed_distributed_variable
  from tensorflow.python.distribute import parameter_server_strategy
  from tensorflow.python.distribute import ps_values
  from tensorflow.python.distribute import strategy_combinations
  from tensorflow.python.distribute import test_util
  from tensorflow.python.distribute import tpu_strategy
  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 indexed_slices
  from tensorflow.python.framework import ops
  from tensorflow.python.framework import tensor_shape
  from tensorflow.python.ops import array_ops
  from tensorflow.python.ops import check_ops
  from tensorflow.python.ops import control_flow_assert
  from tensorflow.python.ops import math_ops
  from tensorflow.python.ops import variable_scope
  from tensorflow.python.ops import variables
  from tensorflow.python.saved_model import save
  from tensorflow.python.saved_model import save_context
  from tensorflow.python.saved_model import save_options
  from tensorflow.python.types import core
label_desc: Use image node_id 9 for calling the DistributedVariableTest obj's underlying member method code with example usage: obj.testAssignSignature(distribution, synchronization, aggregation) and returns: constant_op