# Copyright 2015 The TensorFlow Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ==============================================================================
"""Tests for tensorflow.python.framework.ops."""

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import gc
import threading
import weakref

from tensorflow.core.framework import attr_value_pb2
from tensorflow.core.protobuf import config_pb2
from tensorflow.python.client import session
from tensorflow.python.eager import context
from tensorflow.python.eager import function as eager_function
from tensorflow.python.framework import common_shapes
from tensorflow.python.framework import constant_op
from tensorflow.python.framework import device as pydev
from tensorflow.python.framework import dtypes
from tensorflow.python.framework import errors
from tensorflow.python.framework import function
from tensorflow.python.framework import ops
from tensorflow.python.framework import sparse_tensor
from tensorflow.python.framework import tensor_shape
from tensorflow.python.framework import tensor_util
from tensorflow.python.framework import test_ops
from tensorflow.python.fr
amework import test_util
from tensorflow.python.framework import versions
from tensorflow.python.ops import array_ops
from tensorflow.python.ops import control_flow_ops
from tensorflow.python.ops import math_ops
from tensorflow.python.ops import resource_variable_ops
from tensorflow.python.ops import reso
urces
from tensorflow.python.ops import variable_scope
from tensorflow.python.ops import variables
import tensorflow.python.ops.gradients  # pylint: disable=unused-import
from tensorflow.python.platform import googletest
from tensorflow.python.util import compat

ops._set_call_cpp_shape_fn(common_shapes.call_cpp_shape_fn)


class ResourceTest(test_util.TensorFlowTestCase):

  def testBuildGraph(self):
    with self.test_session():
      pt = test_ops.stub_resource_handle_op(container="a", shared_name="b")
      test_ops.resource_create_op(pt).run()

  def testInitialize(self):
    with self.test_session():
      handle = test_ops.stub_resource_handle_op(container="a", shared_name="b")
      resources.register_resource(
          handle=handle,
          create_op=test_ops.resource_create_op(handle),
          is_initialized_op=test_ops.resource_initialized_op(handle))
      self.assertEquals(
          len(
              resources.report_uninitialized_resources(
                  resources.shared_resources()).eval()), 1)
      resources.initialize_resources(resources.shared_resources()).run()
      self.assertEquals(
          len(
              resources.report_uninitialized_resources(
                  resources.shared_resources()).eval()), 0)


class TensorAndShapeTest(test_util.TensorFlowTestCase):

  def testShape(self):
    op = ops.Operation(
        ops._NodeDef("FloatOutput", "myop"), ops.Graph(), [], [dtypes.float32])
    t = op.outputs[0]
    self.assertEqual(tensor_shape.unknown_shape(), t.get_shape())
    t.set_shape([1, 2, 3])
    self.assertEqual([1, 2, 3], t.get_shape())

  def testIterable(self):
    op = ops.Operation(
        ops._NodeDef("FloatOutput", "myop"), ops.Graph(), [], [dtypes.float32])
    t = op.outputs[0]
    self.assertTrue(isinstance(t, ops.Tensor))
    with self.assertRaisesRegexp(TypeError, "iter"):
      for _ in t:
        pass

  def testAddShape(self):
    with self.test_session():
      a = array_ops.zeros([2, 3])
      b = array_ops.ones([1, 3])
      c = a + b
      self.assertEqual([2, 3], c.shape)

  def testUnknownDim(self):
    with self.test_session():
      a = array_ops.placeholder(dtype=dtypes.float32, shape=[2, None, 3])
      b = array_ops.placeholder(dtype=dtypes.float32, shape=[2, None, 3])
      c = a + b
      self.assertEqual([2, None, 3], c.shape.as_list())

  def testUnknownShape(self):
    with self.test_session():
      a = array_ops.placeholder(dtype=dtypes.float32, shape=None)
      b = array_ops.ones([1, 3])
      c = a + b
      self.assertEqual(tensor_shape.unknown_shape(), c.shape)

  def testScalarShape(self):
    with self.test_session():
      a = array_ops.placeholder(dtype=dtypes.float32, shape=[])
      b = array_ops.ones([])
      c = a + b
      self.assertEqual(tensor_shape.scalar(), c.shape)

  def testShapeFunctionError(self):
    with self.test_session():
      a = array_ops.ones([1, 2, 3])
      b = array_ops.ones([4, 5, 6])
      with self.assertRaisesRegexp(
          ValueError,
          r"Dimensions must be equal, but are 2 and 5 for 'add' \(op: 'Add'\) "
          r"with input shapes: \[1,2,3\], \[4,5,6\]."):
        _ = a + b


class IndexedSlicesTest(test_util.TensorFlowTestCase):

  def testToTensor(self):
    with self.test_session():
      values = constant_op.constant([2, 3, 5, 7], shape=[2, 2])
      indices = constant_op.constant([0, 2])
      dense_shape = constant_op.constant([3, 2])
      x = ops.IndexedSlices(values, indices, dense_shape)
      tensor = ops.convert_to_tensor(x, name="tensor")
      self.assertAllEqual(tensor.eval(), [[2, 3], [0, 0], [5, 7]])

  def testNegation(self):
    with self.test_session():
      values = constant_op.constant([2, 3, 5, 7], shape=[2, 2])
      indices = constant_op.constant([0, 2])
      x = -ops.IndexedSlices(values, indices)
      self.assertAllEqual(x.values.eval(), [[-2, -3], [-5, -7]])
      self.assertAllEqual(x.indices.eval(), [0, 2])

  def testScalarMul(self):
    with self.test_session():
      values = constant_op.constant([2, 3, 5, 7], shape=[2, 2])
      indices = constant_op.constant([0, 2])
      x = math_ops.scalar_mul(-2, ops.IndexedSlices(values, indices))
      self.assertAllEqual(x.values.eval(), [[-4, -6], [-10, -14]])
      self.assertAllEqual(x.indices.eval(), [0, 2])


class NodeDefConstructorTest(test_util.TensorFlowTestCase):

  def testNoArgs(self):
    nodedef = ops._NodeDef("None", "bar")
    self.assertProtoEquals("op: 'None' name: 'bar'", nodedef)

  def testArgs(self):
    nodedef = ops._NodeDef("foo", "bar", device="/device:baz:*")
    self.assertProtoEquals("op:'foo' name:'bar' device:'/device:baz:*'",
                           nodedef)
    nodedef = ops._NodeDef("foo", "bar", device=pydev.DeviceSpec(job="j"))
    self.assertProtoEquals("op:'foo' name:'bar' device:'/job:j'", nodedef)


def _apply_op(g, *args, **kwargs):
  op = g.create_op(*args, **kwargs)
  if len(op.outputs) == 1:
    return op.outputs[0]
  else:
    return op.outputs


class OperationTest(test_util.TensorFlowTestCase):

  def testNoInputs(self):
    op = test_ops.float_output_string_output(name="myop").a.op
    self.assertEqual(2, len(op.values()))
    self.assertEqual(0, len(op.inputs))
    self.assertEqual("myop", op.name)

    float_t, label_str_t = op.values()
    self.assertEqual(dtypes.float32, float_t.dtype)
    self.assertEqual(op, float_t.op)
    self.assertEqual(0, float_t._value_index)
    self.assertEqual(0, len(float_t.consumers()))
    self.assertEqual("myop", float_t._as_node_def_input())

    self.assertEqual(dtypes.string, label_str_t.dtype)
    self.assertEqual(op, label_str_t.op)
    self.assertEqual(1, label_str_t._value_index)
    self.assertEqual(0, len(label_str_t.consumers()))
    self.assertEqual("myop:1", label_str_t._as_node_def_input())

    self.assertProtoEquals("op:'FloatOutputStringOutput' name:'myop'",
                           op.node_def)

  def testNoOutputs(self):
    op1 = test_ops.float_output(name="myop1").op
    float_t, = op1.values()
    op2 = test_ops.float_input(float_t, name="myop2")
    self.assertEqual(0, len(op2.values()))
    self.assertEqual