def testApproximateEqual(self):
    for dtype in [np.float32, np.double]:
      x = dtype(1)
      y = dtype(1.00009)
      z = False
      with test_util.device(use_gpu=True):
        # Default tolerance is 0.00001
        z_tf = self.evaluate(math_ops.approximate_equal(x, y))
        self.assertAllEqual(z, z_tf)

    for dtype in [np.float32, np.double]:
      x = dtype(1)
      y = dtype(1.000009)
      z = True
      with test_util.device(use_gpu=True):
        # Default tolerance is 0.00001
        z_tf = self.evaluate(math_ops.approximate_equal(x, y))
        self.assertAllEqual(z, z_tf)

    for dtype in [np.float32, np.double]:
      x = np.array([[[[-1, 2.00009999], [-3, 4.01]]]], dtype=dtype)
      y = np.array([[[[-1.001, 2], [-3.00009, 4]]]], dtype=dtype)
      z = np.array([[[[False, True], [True, False]]]], dtype=np.bool)
      with test_util.device(use_gpu=True):
        z_tf = self.evaluate(math_ops.approximate_equal(x, y, tolerance=0.0001))
        self.assertAllEqual(z, z_tf)

 def testAgnosticUsage(self):
   """Graph/eager agnostic usage."""
   # Does create garbage when executing eagerly due to ops.Graph() creation.
   num_training_steps = 10
   checkpoint_directory = self.get_temp_dir()
   checkpoint_prefix = os.path.join(checkpoint_directory, "ckpt")
   for training_continuation in range(3):
     with ops.Graph().as_default(), self.test_session(
         graph=ops.get_default_graph()), test_util.device(use_gpu=True):
       model = MyModel()
       optimizer = adam.AdamOptimizer(0.001)
       root = util.Checkpoint(
           optimizer=optimizer, model=model,
           global_step=training_util.get_or_create_global_step())
       checkpoint_path = checkpoint_management.latest_checkpoint(
           checkpoint_directory)
       status = root.restore(save_path=checkpoint_path)
       input_value = constant_op.constant([[3.]])
       train_fn = functools.partial(
           optimizer.minimize,
           functools.partial(model, input_value),
           global_step=root.global_step)
       if not context.executing_eagerly():
         train_fn = functools.partial(self.evaluate, train_fn())
       status.initialize_or_restore()
       for _ in range(num_training_steps):
         train_fn()
       root.save(file_prefix=checkpoint_prefix)
       self.assertEqual((training_continuation + 1) * num_training_steps,
                        self.evaluate(root.global_step))
       self.assertEqual(training_continuation + 1,
                        self.evaluate(root.save_counter))

 def _compare(self, c, x, y, use_gpu):
   np_ans = np.where(c, x, y)
   with test_util.device(use_gpu=use_gpu):
     out = array_ops.where(c, x, y)
     tf_ans = self.evaluate(out)
   self.assertAllEqual(np_ans, tf_ans)
   self.assertShapeEqual(np_ans, out)

  def testBasicMemory(self):
    """Make sure arguments can be passed correctly."""
    with test_util.device(use_gpu=False):
      a = constant_op.constant(10, name="a")
      b = constant_op.constant(20, name="b")
      c = math_ops.add_n([a, b], name="c")
      d = math_ops.add_n([b, c], name="d")
      train_op = ops.get_collection_ref(ops.GraphKeys.TRAIN_OP)
      train_op.append(d)
      mg = meta_graph.create_meta_graph_def(graph=ops.get_default_graph())

    report = cost_analyzer.GenerateMemoryReport(mg)

    # Print the report to make it easier to debug
    print("{}".format(report))

    # Check the report
    self.assertTrue(
        "Peak usage for device /job:localhost/replica:0/task:0/device:CPU:0: "
        "16 bytes"
        in report)
    self.assertTrue("  a:0 uses 4 bytes" in report)
    self.assertTrue("  b:0 uses 4 bytes" in report)
    self.assertTrue("  c:0 uses 4 bytes" in report)
    self.assertTrue("  d:0 uses 4 bytes" in report)

  def testAcceptsTensor(self):
    tensor = array_ops.ones([10, 10])
    result = math_ops.scalar_mul(3, tensor)
    expected = array_ops.ones([10, 10]) * 3

    with test_util.device(use_gpu=True):
      self.assertAllEqual(self.evaluate(expected), self.evaluate(result))

 def testEagerSingleOutputFloat32(self):
   with test_util.device(use_gpu=True):
     a = array_ops.ones((3, 3), dtype=dtypes.float32)
     x = array_ops.ones((3, 1), dtype=dtypes.float32)
     output = script_ops.eager_py_func(matmul, inp=[a, x], Tout=dtypes.float32)
     ret = self.evaluate(output)
     self.assertAllClose(ret, [[3.0], [3.0], [3.0]])

 def testLoadFromNameBasedSaver(self):
   """Save a name-based checkpoint, load it using the object-based API."""
   with test_util.device(use_gpu=True):
     save_path = self._write_name_based_checkpoint()
     root = self._initialized_model()
     self._set_sentinels(root)
     with self.assertRaises(AssertionError):
       self._check_sentinels(root)
     object_saver = util.TrackableSaver(graph_view.ObjectGraphView(root))
     self._set_sentinels(root)
     status = object_saver.restore(save_path)
     if context.executing_eagerly():
       self._check_sentinels(root)
     if context.executing_eagerly():
       with self.assertRaisesRegexp(AssertionError, "OBJECT_CONFIG_JSON"):
         status.assert_consumed()
     else:
       # When graph building, we haven't read any keys, so we don't know
       # whether the restore will be complete.
       with self.assertRaisesRegexp(AssertionError, "not restored"):
         status.assert_consumed()
     status.run_restore_ops()
     self._check_sentinels(root)
     self._set_sentinels(root)
     status = object_saver.restore(save_path)
     status.initialize_or_restore()
     self._check_sentinels(root)

 def testAgnosticUsage(self):
   """Graph/eager agnostic usage."""
   # Does create garbage when executing eagerly due to ops.Graph() creation.
   num_training_steps = 10
   checkpoint_directory = self.get_temp_dir()
   for training_continuation in range(3):
     with test_util.device(use_gpu=True):
       model = MyModel()
       optimizer = adam.AdamOptimizer(0.001)
       root = checkpointable_utils.Checkpoint(
           optimizer=optimizer, model=model,
           global_step=training_util.get_or_create_global_step())
       manager = checkpoint_management.CheckpointManager(
           root, checkpoint_directory, max_to_keep=1)
       status = root.restore(save_path=manager.latest_checkpoint)
       input_value = constant_op.constant([[3.]])
       train_fn = functools.partial(
           optimizer.minimize,
           functools.partial(model, input_value),
           global_step=root.global_step)
       if not context.executing_eagerly():
         train_fn = functools.partial(self.evaluate, train_fn())
       status.initialize_or_restore()
       for _ in range(num_training_steps):
         train_fn()
       manager.save()
       self.assertEqual((training_continuation + 1) * num_training_steps,
                        self.evaluate(root.global_step))
       self.assertEqual(training_continuation + 1,
                        self.evaluate(root.save_counter))

  def Test(self):
    np_val = np.matrix(a_np_) * np.matrix(b_np_)

    use_gpu = True
    if a_np_.dtype is np.float16 and (
        not test_util.CudaSupportsHalfMatMulAndConv()):
      use_gpu = False
      print("Built without fp16 matmul support for Cuda, running test on CPU.")

    # Transpose and possibly conjugate a_np_ and b_np_ according to the
    # attributes such that tf.matmul(effective_a_np, effective_b_np, **kwargs)
    # results in a valid matrix multiplication and produces the same result as
    # np.matrix(a_np_) * np.matrix(b_np_)
    effective_a_np = _GetTransposedMatrices(a_np_, "a", kwargs_)
    effective_b_np = _GetTransposedMatrices(b_np_, "b", kwargs_)
    with self.cached_session() as sess, test_util.device(use_gpu):
      if use_static_shape_:
        a = constant_op.constant(effective_a_np)
        b = constant_op.constant(effective_b_np)
        res = math_ops.matmul(a, b, **kwargs_)
        tf_val = self.evaluate(res)
      else:
        a = array_ops.placeholder(a_np_.dtype)
        b = array_ops.placeholder(b_np_.dtype)
        res = math_ops.matmul(a, b, **kwargs_)
        tf_val = sess.run(res, feed_dict={a: effective_a_np, b: effective_b_np})

    self.assertAllCloseAccordingToType(
        tf_val,
        np_val,
        float_rtol=2e-5,
        float_atol=2e-5,
        half_rtol=0.2,
        half_atol=0.2)

 def testSmallEntropy(self):
   random_seed.set_random_seed(1618)
   with test_util.device(use_gpu=True):
     # A logit value of -10 corresponds to a probability of ~5e-5.
     logits = constant_op.constant([[-10., 10., -10.], [-10., -10., 10.]])
     num_samples = 1000
     samples = self.evaluate(random_ops.multinomial(logits, num_samples))
     self.assertAllEqual([[1] * num_samples, [2] * num_samples], samples)

 def testSquaredDifference(self):
   for dtype in [np.int32, np.float16]:
     x = np.array([[1, 2, 3], [4, 5, 6]], dtype=dtype)
     y = np.array([-3, -2, -1], dtype=dtype)
     z = (x - y) * (x - y)
     with test_util.device(use_gpu=True):
       z_tf = self.evaluate(math_ops.squared_difference(x, y))
       self.assertAllClose(z, z_tf)

 def _not(self, x, use_gpu=False):
   np_ans = np.logical_not(x)
   with test_util.device(use_gpu=use_gpu):
     out = math_ops.logical_not(ops.convert_to_tensor(x))
     tf_val = self.evaluate(out)
   self.assertEqual(out.dtype, dtypes_lib.bool)
   self.assertAllEqual(np_ans, tf_val)
   self.assertShapeEqual(np_ans, out)

 def _compareConj(self, cplx, use_gpu):
   np_ans = np.conj(cplx)
   with test_util.device(use_gpu=use_gpu):
     inx = ops.convert_to_tensor(cplx)
     tf_conj = math_ops.conj(inx)
     tf_ans = self.evaluate(tf_conj)
   self.assertAllEqual(np_ans, tf_ans)
   self.assertShapeEqual(np_ans, tf_conj)

 def testEagerArrayOutput(self):
   with test_util.device(use_gpu=True):
     a = array_ops.ones((3, 3), dtype=dtypes.float32)
     x = array_ops.ones((3, 1), dtype=dtypes.float32)
     output = script_ops.eager_py_func(
         lambda a, x: [matmul(a, x)], inp=[a, x], Tout=[dtypes.float32])
     ret = self.evaluate(output)
     self.assertAllEqual(ret, [[[3.0], [3.0], [3.0]]])

 def testAcceptsIndexedSlices(self):
   values = constant_op.constant([2, 3, 5, 7, 0, -1], shape=[3, 2])
   indices = constant_op.constant([0, 2, 5])
   x = math_ops.scalar_mul(-3, ops.IndexedSlices(values, indices))
   with test_util.device(use_gpu=True):
     self.assertAllEqual(self.evaluate(x.values),
                         [[-6, -9], [-15, -21], [0, 3]])
     self.assertAllEqual(self.evaluate(x.indices), [0, 2, 5])

 def _testEmpty(self, x, dim, num, expected_shape):
   with test_util.device(use_gpu=True):
     tf_ans = array_ops.split(value=x, num_or_size_splits=num, axis=dim)
     out = self.evaluate(tf_ans)
   self.assertEqual(x.size, 0)
   self.assertEqual(len(out), num)
   for i in range(num):
     self.assertEqual(out[i].shape, expected_shape)
     self.assertEqual(expected_shape, tf_ans[i].get_shape())

 def testComplexSquaredDifference(self):
   for dtype in [np.complex64, np.complex128]:
     x = np.array([[1 + 3j, 2 + 2j, 3 + 1j], [4 - 1j, 5 - 2j, 6 - 3j]],
                  dtype=dtype)
     y = np.array([-3 + 1j, -2 + 2j, -1 + 3j], dtype=dtype)
     z = np.conj(x - y) * (x - y)
     with test_util.device(use_gpu=False):
       z_tf = self.evaluate(math_ops.squared_difference(x, y))
       self.assertAllClose(z, z_tf)

 def _compareBinary(self, x, y, np_func, tf_func, use_gpu=False):
   np_ans = np_func(x, y)
   with test_util.device(use_gpu=use_gpu):
     inx = ops.convert_to_tensor(x)
     iny = ops.convert_to_tensor(y)
     out = tf_func(inx, iny)
     tf_val = self.evaluate(out)
   self.assertEqual(out.dtype, dtypes_lib.bool)
   self.assertAllEqual(np_ans, tf_val)
   self.assertShapeEqual(np_ans, out)

 def testAcceptsRefs(self):
   if context.executing_eagerly():
     var = resource_variable_ops.ResourceVariable(10, name="var")
   else:
     var = variables.Variable(10)
   result = math_ops.scalar_mul(3, var)
   init = variables.global_variables_initializer()
   with test_util.device(use_gpu=True):
     self.evaluate(init)
     self.assertEqual(30, self.evaluate(result))

  def _compareAngle(self, cplx, use_gpu):
    np_angle = np.angle(cplx)

    with test_util.device(use_gpu=use_gpu):
      inx = ops.convert_to_tensor(cplx)
      tf_angle = math_ops.angle(inx)
      tf_angle_val = self.evaluate(tf_angle)

    self.assertAllClose(np_angle, tf_angle_val)
    self.assertShapeEqual(np_angle, tf_angle)