path: root/tensorflow/python/debug/wrappers/local_cli_wrapper_test.py

# Copyright 2016 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.
# ==============================================================================
"""Unit tests for local command-line-interface debug wrapper session."""
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import os
import shutil
import tempfile

from tensorflow.core.protobuf import config_pb2
from tensorflow.python.client import session
from tensorflow.python.debug.cli import cli_shared
from tensorflow.python.debug.cli import debugger_cli_common
from tensorflow.python.debug.wrappers import local_cli_wrapper
from tensorflow.python.framework import constant_op
from tensorflow.python.framework import dtypes
from tensorflow.python.framework import errors
from tensorflow.python.framework import ops
from tensorflow.python.framework import test_util
from tensorflow.python.ops import array_ops
from tensorflow.python.ops import control_flow_ops
from tensorflow.python.ops import math_ops
# Import resource_variable_ops for the variables-to-tensor implicit conversion.
from tensorflow.python.ops import resource_variable_ops  # pylint: disable=unused-import
from tensorflow.python.ops import sparse_ops
from tensorflow.python.ops import state_ops
from tensorflow.python.ops import variables
from tensorflow.python.platform import googletest


class LocalCLIDebuggerWrapperSessionForTest(
    local_cli_wrapper.LocalCLIDebugWrapperSession):
  """Subclasses the wrapper class for testing.

  Overrides its CLI-related methods for headless testing environments.
  Inserts observer variables for assertions.
  """

  def __init__(self,
               command_sequence,
               sess,
               dump_root=None):
    """Constructor of the for-test subclass.

    Args:
      command_sequence: (list of list of str) A list of command arguments,
        including the command prefix, each element of the list is such as:
        ["run", "-n"],
        ["print_feed", "input:0"].
      sess: See the doc string of LocalCLIDebugWrapperSession.__init__.
      dump_root: See the doc string of LocalCLIDebugWrapperSession.__init__.
    """

    local_cli_wrapper.LocalCLIDebugWrapperSession.__init__(
        self, sess, dump_root=dump_root, log_usage=False)

    self._command_sequence = command_sequence
    self._command_pointer = 0

    # Observer variables.
    self.observers = {
        "debug_dumps": [],
        "tf_errors": [],
        "run_start_cli_run_numbers": [],
        "run_end_cli_run_numbers": [],
        "print_feed_responses": [],
        "profiler_py_graphs": [],
        "profiler_run_metadata": [],
    }

  def _prep_cli_for_run_start(self):
    pass

  def _prep_debug_cli_for_run_end(self, debug_dump, tf_error, passed_filter):
    self.observers["debug_dumps"].append(debug_dump)
    self.observers["tf_errors"].append(tf_error)

  def _prep_profile_cli_for_run_end(self, py_graph, run_metadata):
    self.observers["profiler_py_graphs"].append(py_graph)
    self.observers["profiler_run_metadata"].append(run_metadata)

  def _launch_cli(self):
    if self._is_run_start:
      self.observers["run_start_cli_run_numbers"].append(self._run_call_count)
    else:
      self.observers["run_end_cli_run_numbers"].append(self._run_call_count)

    while True:
      command = self._command_sequence[self._command_pointer]
      self._command_pointer += 1

      try:
        if command[0] == "run":
          self._run_handler(command[1:])
        elif command[0] == "print_feed":
          self.observers["print_feed_responses"].append(
              self._print_feed_handler(command[1:]))
        else:
          raise ValueError("Unrecognized command prefix: %s" % command[0])
      except debugger_cli_common.CommandLineExit as e:
        return e.exit_token


class LocalCLIDebugWrapperSessionTest(test_util.TensorFlowTestCase):

  def setUp(self):
    self._tmp_dir = tempfile.mktemp()

    self.v = variables.Variable(10.0, name="v")
    self.w = variables.Variable(21.0, name="w")
    self.delta = constant_op.constant(1.0, name="delta")
    self.inc_v = state_ops.assign_add(self.v, self.delta, name="inc_v")

    self.w_int = control_flow_ops.with_dependencies(
        [self.inc_v],
        math_ops.cast(self.w, dtypes.int32, name="w_int_inner"),
        name="w_int_outer")

    self.ph = array_ops.placeholder(dtypes.float32, name="ph")
    self.xph = array_ops.transpose(self.ph, name="xph")
    self.m = constant_op.constant(
        [[0.0, 1.0, 2.0], [-4.0, -1.0, 0.0]], dtype=dtypes.float32, name="m")
    self.y = math_ops.matmul(self.m, self.xph, name="y")

    self.sparse_ph = array_ops.sparse_placeholder(
        dtypes.float32, shape=([5, 5]), name="sparse_placeholder")
    self.sparse_add = sparse_ops.sparse_add(self.sparse_ph, self.sparse_ph)

    self.sess = session.Session()

    # Initialize variable.
    self.sess.run(variables.global_variables_initializer())

  def tearDown(self):
    ops.reset_default_graph()
    if os.path.isdir(self._tmp_dir):
      shutil.rmtree(self._tmp_dir)

  def testConstructWrapper(self):
    local_cli_wrapper.LocalCLIDebugWrapperSession(
        session.Session(), log_usage=False)

  def testConstructWrapperWithExistingEmptyDumpRoot(self):
    os.mkdir(self._tmp_dir)
    self.assertTrue(os.path.isdir(self._tmp_dir))

    local_cli_wrapper.LocalCLIDebugWrapperSession(
        session.Session(), dump_root=self._tmp_dir, log_usage=False)

  def testConstructWrapperWithExistingNonEmptyDumpRoot(self):
    os.mkdir(self._tmp_dir)
    dir_path = os.path.join(self._tmp_dir, "foo")
    os.mkdir(dir_path)
    self.assertTrue(os.path.isdir(dir_path))

    with self.assertRaisesRegexp(
        ValueError, "dump_root path points to a non-empty directory"):
      local_cli_wrapper.LocalCLIDebugWrapperSession(
          session.Session(), dump_root=self._tmp_dir, log_usage=False)

  def testConstructWrapperWithExistingFileDumpRoot(self):
    os.mkdir(self._tmp_dir)
    file_path = os.path.join(self._tmp_dir, "foo")
    open(file_path, "a").close()  # Create the file
    self.assertTrue(os.path.isfile(file_path))
    with self.assertRaisesRegexp(ValueError, "dump_root path points to a file"):
      local_cli_wrapper.LocalCLIDebugWrapperSession(
          session.Session(), dump_root=file_path, log_usage=False)

  def testRunsUnderDebugMode(self):
    wrapped_sess = LocalCLIDebuggerWrapperSessionForTest(
        [["run"], ["run"], ["run"]], self.sess, dump_root=self._tmp_dir)

    # run under debug mode twice.
    wrapped_sess.run(self.inc_v)
    wrapped_sess.run(self.inc_v)

    # Verify that the assign_add op did take effect.
    self.assertAllClose(12.0, self.sess.run(self.v))

    # Assert correct run call numbers for which the CLI has been launched at
    # run-start and run-end.
    self.assertEqual([1], wrapped_sess.observers["run_start_cli_run_numbers"])
    self.assertEqual([1, 2], wrapped_sess.observers["run_end_cli_run_numbers"])

    # Verify that the dumps have been generated and picked up during run-end.
    self.assertEqual(2, len(wrapped_sess.observers["debug_dumps"]))

    # Verify that the TensorFlow runtime errors are picked up and in this case,
    # they should be both None.
    self.assertEqual([None, None], wrapped_sess.observers["tf_errors"])

  def testRunsWithEmptyStringDumpRootWorks(self):
    wrapped_sess = LocalCLIDebuggerWrapperSessionForTest(
        [["run"], ["run"]], self.sess, dump_root="")

    # run under debug mode.
    wrapped_sess.run(self.inc_v)

    self.assertAllClose(11.0, self.sess.run(self.v))

  def testRunInfoOutputAtRunEndIsCorrect(self):
    wrapped_sess = LocalCLIDebuggerWrapperSessionForTest(
        [["run"], ["run"], ["run"]], self.sess, dump_root=self._tmp_dir)

    wrapped_sess.run(self.inc_v)
    run_info_output = wrapped_sess._run_info_handler([])

    tfdbg_logo = cli_shared.get_tfdbg_logo()

    # The run_info output in the first run() call should contain the tfdbg logo.
    self.assertEqual(tfdbg_logo.lines,
                     run_info_output.lines[:len(tfdbg_logo.lines)])
    menu = run_info_output.annotations[debugger_cli_common.MAIN_MENU_KEY]
    self.assertIn("list_tensors", menu.captions())

    wrapped_sess.run(self.inc_v)
    run_info_output = wrapped_sess._run_info_handler([])

    # The run_info output in the second run() call should NOT contain the logo.
    self.assertNotEqual(tfdbg_logo.lines,
                        run_info_output.lines[:len(tfdbg_logo.lines)])
    menu = run_info_output.annotations[debugger_cli_common.MAIN_MENU_KEY]
    self.assertIn("list_tensors", menu.captions())

  def testRunsUnderNonDebugMode(self):
    wrapped_sess = LocalCLIDebuggerWrapperSessionForTest(
        [["run", "-n"], ["run", "-n"], ["run", "-n"]],
        self.sess, dump_root=self._tmp_dir)

    # run three times.
    wrapped_sess.run(self.inc_v)
    wrapped_sess.run(self.inc_v)
    wrapped_sess.run(self.inc_v)

    self.assertAllClose(13.0, self.sess.run(self.v))

    self.assertEqual([1, 2, 3],
                     wrapped_sess.observers["run_start_cli_run_numbers"])
    self.assertEqual([], wrapped_sess.observers["run_end_cli_run_numbers"])

  def testRunningWithSparsePlaceholderFeedWorks(self):
    wrapped_sess = LocalCLIDebuggerWrapperSessionForTest(
        [["run"], ["run"]], self.sess, dump_root=self._tmp_dir)

    sparse_feed = ([[0, 1], [0, 2]], [10.0, 20.0])
    sparse_result = wrapped_sess.run(
        self.sparse_add, feed_dict={self.sparse_ph: sparse_feed})
    self.assertAllEqual([[0, 1], [0, 2]], sparse_result.indices)
    self.assertAllClose([20.0, 40.0], sparse_result.values)

  def testRunsUnderNonDebugThenDebugMode(self):
    # Do two NON_DEBUG_RUNs, followed by DEBUG_RUNs.
    wrapped_sess = LocalCLIDebuggerWrapperSessionForTest(
        [["run", "-n"], ["run", "-n"], ["run"], ["run"]],
        self.sess, dump_root=self._tmp_dir)

    # run three times.
    wrapped_sess.run(self.inc_v)
    wrapped_sess.run(self.inc_v)
    wrapped_sess.run(self.inc_v)

    self.assertAllClose(13.0, self.sess.run(self.v))

    self.assertEqual([1, 2, 3],
                     wrapped_sess.observers["run_start_cli_run_numbers"])

    # Here, the CLI should have been launched only under the third run,
    # because the first and second runs are NON_DEBUG.
    self.assertEqual([3], wrapped_sess.observers["run_end_cli_run_numbers"])
    self.assertEqual(1, len(wrapped_sess.observers["debug_dumps"]))
    self.assertEqual([None], wrapped_sess.observers["tf_errors"])

  def testRunMultipleTimesWithinLimit(self):
    wrapped_sess = LocalCLIDebuggerWrapperSessionForTest(
        [["run", "-t", "3"], ["run"]],
        self.sess, dump_root=self._tmp_dir)

    # run three times.
    wrapped_sess.run(self.inc_v)
    wrapped_sess.run(self.inc_v)
    wrapped_sess.run(self.inc_v)

    self.assertAllClose(13.0, self.sess.run(self.v))

    self.assertEqual([1], wrapped_sess.observers["run_start_cli_run_numbers"])
    self.assertEqual([3], wrapped_sess.observers["run_end_cli_run_numbers"])
    self.assertEqual(1, len(wrapped_sess.observers["debug_dumps"]))
    self.assertEqual([None], wrapped_sess.observers["tf_errors"])

  def testRunMultipleTimesOverLimit(self):
    wrapped_sess = LocalCLIDebuggerWrapperSessionForTest(
        [["run", "-t", "3"]], self.sess, dump_root=self._tmp_dir)

    # run twice, which is less than the number of times specified by the
    # command.
    wrapped_sess.run(self.inc_v)
    wrapped_sess.run(self.inc_v)

    self.assertAllClose(12.0, self.sess.run(self.v))

    self.assertEqual([1], wrapped_sess.observers["run_start_cli_run_numbers"])
    self.assertEqual([], wrapped_sess.observers["run_end_cli_run_numbers"])
    self.assertEqual(0, len(wrapped_sess.observers["debug_dumps"]))
    self.assertEqual([], wrapped_sess.observers["tf_errors"])

  def testRunMixingDebugModeAndMultpleTimes(self):
    wrapped_sess = LocalCLIDebuggerWrapperSessionForTest(
        [["run", "-n"], ["run", "-t", "2"], ["run"], ["run"]],
        self.sess, dump_root=self._tmp_dir)

    # run four times.
    wrapped_sess.run(self.inc_v)
    wrapped_sess.run(self.inc_v)
    wrapped_sess.run(self.inc_v)
    wrapped_sess.run(self.inc_v)

    self.assertAllClose(14.0, self.sess.run(self.v))

    self.assertEqual([1, 2],
                     wrapped_sess.observers["run_start_cli_run_numbers"])
    self.assertEqual([3, 4], wrapped_sess.observers["run_end_cli_run_numbers"])
    self.assertEqual(2, len(wrapped_sess.observers["debug_dumps"]))
    self.assertEqual([None, None], wrapped_sess.observers["tf_errors"])

  def testDebuggingMakeCallableTensorRunnerWorks(self):
    wrapped_sess = LocalCLIDebuggerWrapperSessionForTest(
        [["run"], ["run"]], self.sess, dump_root=self._tmp_dir)
    v = variables.Variable(42)
    tensor_runner = wrapped_sess.make_callable(v)
    self.sess.run(v.initializer)

    self.assertAllClose(42, tensor_runner())
    self.assertEqual(1, len(wrapped_sess.observers["debug_dumps"]))

  def testDebuggingMakeCallableTensorRunnerWithCustomRunOptionsWorks(self):
    wrapped_sess = LocalCLIDebuggerWrapperSessionForTest(
        [["run"], ["run"]], self.sess, dump_root=self._tmp_dir)
    a = constant_op.constant(42)
    tensor_runner = wrapped_sess.make_callable(a)

    run_options = config_pb2.RunOptions(
        trace_level=config_pb2.RunOptions.FULL_TRACE)
    run_metadata = config_pb2.RunMetadata()
    self.assertAllClose(
        42, tensor_runner(options=run_options, run_metadata=run_metadata))
    self.assertEqual(1, len(wrapped_sess.observers["debug_dumps"]))
    self.assertGreater(len(run_metadata.step_stats.dev_stats), 0)

  def testDebuggingMakeCallableOperationRunnerWorks(self):
    wrapped_sess = LocalCLIDebuggerWrapperSessionForTest(
        [["run"], ["run"]], self.sess, dump_root=self._tmp_dir)
    v = variables.Variable(10.0)
    inc_v = state_ops.assign_add(v, 1.0)
    op_runner = wrapped_sess.make_callable(inc_v.op)
    self.sess.run(v.initializer)

    op_runner()
    self.assertEqual(1, len(wrapped_sess.observers["debug_dumps"]))
    self.assertEqual(11.0, self.sess.run(v))

  def testDebuggingMakeCallableRunnerWithFeedListWorks(self):
    wrapped_sess = LocalCLIDebuggerWrapperSessionForTest(
        [["run"], ["run"]], self.sess, dump_root=self._tmp_dir)
    ph1 = array_ops.placeholder(dtypes.float32)
    ph2 = array_ops.placeholder(dtypes.float32)
    a = math_ops.add(ph1, ph2)
    tensor_runner = wrapped_sess.make_callable(a, feed_list=[ph1, ph2])

    self.assertAllClose(42.0, tensor_runner(41.0, 1.0))
    self.assertEqual(1, len(wrapped_sess.observers["debug_dumps"]))

  def testRuntimeErrorShouldBeCaught(self):
    wrapped_sess = LocalCLIDebuggerWrapperSessionForTest(
        [["run"], ["run"]], self.sess, dump_root=self._tmp_dir)

    # Do a run that should lead to an TensorFlow runtime error.
    wrapped_sess.run(self.y, feed_dict={self.ph: [[0.0], [1.0], [2.0]]})

    self.assertEqual([1], wrapped_sess.observers["run_start_cli_run_numbers"])
    self.assertEqual([1], wrapped_sess.observers["run_end_cli_run_numbers"])
    self.assertEqual(1, len(wrapped_sess.observers["debug_dumps"]))

    # Verify that the runtime error is caught by the wrapped session properly.
    self.assertEqual(1, len(wrapped_sess.observers["tf_errors"]))
    tf_error = wrapped_sess.observers["tf_errors"][0]
    self.assertEqual("y", tf_error.op.name)

  def testRuntimeErrorBeforeGraphExecutionIsRaised(self):
    # Use an impossible device name to cause an error before graph execution.
    with ops.device("/gpu:1337"):
      w = variables.Variable([1.0] * 10, name="w")

    wrapped_sess = LocalCLIDebuggerWrapperSessionForTest(
        [["run"]], self.sess, dump_root=self._tmp_dir)
    with self.assertRaisesRegexp(errors.OpError, r".*[Dd]evice.*1337.*"):
      wrapped_sess.run(w)

  def testRunTillFilterPassesShouldLaunchCLIAtCorrectRun(self):
    wrapped_sess = LocalCLIDebuggerWrapperSessionForTest(
        [["run", "-f", "v_greater_than_twelve"],
         ["run", "-f", "v_greater_than_twelve"],
         ["run"]],
        self.sess,
        dump_root=self._tmp_dir)

    def v_greater_than_twelve(datum, tensor):
      return datum.node_name == "v" and tensor > 12.0

    wrapped_sess.add_tensor_filter("v_greater_than_twelve",
                                   v_greater_than_twelve)

    # run five times.
    wrapped_sess.run(self.inc_v)
    wrapped_sess.run(self.inc_v)
    wrapped_sess.run(self.inc_v)
    wrapped_sess.run(self.inc_v)
    wrapped_sess.run(self.inc_v)

    self.assertAllClose(15.0, self.sess.run(self.v))

    self.assertEqual([1], wrapped_sess.observers["run_start_cli_run_numbers"])

    # run-end CLI should NOT have been launched for run #2 and #3, because only
    # starting from run #4 v becomes greater than 12.0.
    self.assertEqual([4, 5], wrapped_sess.observers["run_end_cli_run_numbers"])

    self.assertEqual(2, len(wrapped_sess.observers["debug_dumps"]))
    self.assertEqual([None, None], wrapped_sess.observers["tf_errors"])

  def testRunsUnderDebugModeWithWatchFnFilteringNodeNames(self):
    wrapped_sess = LocalCLIDebuggerWrapperSessionForTest(
        [["run", "--node_name_filter", "inc.*"],
         ["run", "--node_name_filter", "delta"],
         ["run"]],
        self.sess, dump_root=self._tmp_dir)

    # run under debug mode twice.
    wrapped_sess.run(self.inc_v)
    wrapped_sess.run(self.inc_v)

    # Verify that the assign_add op did take effect.
    self.assertAllClose(12.0, self.sess.run(self.v))

    # Verify that the dumps have been generated and picked up during run-end.
    self.assertEqual(2, len(wrapped_sess.observers["debug_dumps"]))

    dumps = wrapped_sess.observers["debug_dumps"][0]
    self.assertEqual(1, dumps.size)
    self.assertEqual("inc_v", dumps.dumped_tensor_data[0].node_name)

    dumps = wrapped_sess.observers["debug_dumps"][1]
    self.assertEqual(1, dumps.size)
    self.assertEqual("delta", dumps.dumped_tensor_data[0].node_name)

  def testRunsUnderDebugModeWithWatchFnFilteringOpTypes(self):
    wrapped_sess = LocalCLIDebuggerWrapperSessionForTest(
        [["run", "--node_name_filter", "delta"],
         ["run", "--op_type_filter", "AssignAdd"],
         ["run"]],
        self.sess, dump_root=self._tmp_dir)

    # run under debug mode twice.
    wrapped_sess.run(self.inc_v)
    wrapped_sess.run(self.inc_v)

    # Verify that the assign_add op did take effect.
    self.assertAllClose(12.0, self.sess.run(self.v))

    # Verify that the dumps have been generated and picked up during run-end.
    self.assertEqual(2, len(wrapped_sess.observers["debug_dumps"]))

    dumps = wrapped_sess.observers["debug_dumps"][0]
    self.assertEqual(1, dumps.size)
    self.assertEqual("delta", dumps.dumped_tensor_data[0].node_name)

    dumps = wrapped_sess.observers["debug_dumps"][1]
    self.assertEqual(1, dumps.size)
    self.assertEqual("inc_v", dumps.dumped_tensor_data[0].node_name)

  def testRunsUnderDebugModeWithWatchFnFilteringTensorDTypes(self):
    wrapped_sess = LocalCLIDebuggerWrapperSessionForTest(
        [["run", "--op_type_filter", "Variable.*"],
         ["run", "--tensor_dtype_filter", "int32"],
         ["run"]],
        self.sess, dump_root=self._tmp_dir)

    # run under debug mode twice.
    wrapped_sess.run(self.w_int)
    wrapped_sess.run(self.w_int)

    # Verify that the dumps have been generated and picked up during run-end.
    self.assertEqual(2, len(wrapped_sess.observers["debug_dumps"]))

    dumps = wrapped_sess.observers["debug_dumps"][0]
    self.assertEqual(2, dumps.size)
    self.assertItemsEqual(
        ["v", "w"], [dumps.dumped_tensor_data[i].node_name for i in [0, 1]])

    dumps = wrapped_sess.observers["debug_dumps"][1]
    self.assertEqual(2, dumps.size)
    self.assertEqual(
        ["w_int_inner", "w_int_outer"],
        [dumps.dumped_tensor_data[i].node_name for i in [0, 1]])

  def testRunsUnderDebugModeWithWatchFnFilteringOpTypesAndTensorDTypes(self):
    wrapped_sess = LocalCLIDebuggerWrapperSessionForTest(
        [["run", "--op_type_filter", "Cast", "--tensor_dtype_filter", "int32"],
         ["run"]],
        self.sess, dump_root=self._tmp_dir)

    # run under debug mode twice.
    wrapped_sess.run(self.w_int)

    # Verify that the dumps have been generated and picked up during run-end.
    self.assertEqual(1, len(wrapped_sess.observers["debug_dumps"]))

    dumps = wrapped_sess.observers["debug_dumps"][0]
    self.assertEqual(1, dumps.size)
    self.assertEqual("w_int_inner", dumps.dumped_tensor_data[0].node_name)

  def testPrintFeedPrintsFeedValueForTensorFeedKey(self):
    wrapped_sess = LocalCLIDebuggerWrapperSessionForTest(
        [["print_feed", "ph:0"], ["run"], ["run"]], self.sess)

    self.assertAllClose(
        [[5.0], [-1.0]],
        wrapped_sess.run(self.y, feed_dict={self.ph: [[0.0, 1.0, 2.0]]}))
    print_feed_responses = wrapped_sess.observers["print_feed_responses"]
    self.assertEqual(1, len(print_feed_responses))
    self.assertEqual(
        ["Tensor \"ph:0 (feed)\":", "", "[[0.0, 1.0, 2.0]]"],
        print_feed_responses[0].lines)

  def testPrintFeedPrintsFeedValueForTensorNameFeedKey(self):
    wrapped_sess = LocalCLIDebuggerWrapperSessionForTest(
        [["print_feed", "ph:0"], ["run"], ["run"]], self.sess)

    self.assertAllClose(
        [[5.0], [-1.0]],
        wrapped_sess.run(self.y, feed_dict={"ph:0": [[0.0, 1.0, 2.0]]}))
    print_feed_responses = wrapped_sess.observers["print_feed_responses"]
    self.assertEqual(1, len(print_feed_responses))
    self.assertEqual(
        ["Tensor \"ph:0 (feed)\":", "", "[[0.0, 1.0, 2.0]]"],
        print_feed_responses[0].lines)

  def testPrintFeedPrintsErrorForInvalidFeedKey(self):
    wrapped_sess = LocalCLIDebuggerWrapperSessionForTest(
        [["print_feed", "spam"], ["run"], ["run"]], self.sess)

    self.assertAllClose(
        [[5.0], [-1.0]],
        wrapped_sess.run(self.y, feed_dict={"ph:0": [[0.0, 1.0, 2.0]]}))
    print_feed_responses = wrapped_sess.observers["print_feed_responses"]
    self.assertEqual(1, len(print_feed_responses))
    self.assertEqual(
        ["ERROR: The feed_dict of the current run does not contain the key "
         "spam"], print_feed_responses[0].lines)

  def testPrintFeedPrintsErrorWhenFeedDictIsNone(self):
    wrapped_sess = LocalCLIDebuggerWrapperSessionForTest(
        [["print_feed", "spam"], ["run"], ["run"]], self.sess)

    wrapped_sess.run(self.w_int)
    print_feed_responses = wrapped_sess.observers["print_feed_responses"]
    self.assertEqual(1, len(print_feed_responses))
    self.assertEqual(
        ["ERROR: The feed_dict of the current run is None or empty."],
        print_feed_responses[0].lines)

  def testRunUnderProfilerModeWorks(self):
    wrapped_sess = LocalCLIDebuggerWrapperSessionForTest(
        [["run", "-p"], ["run"]], self.sess)

    wrapped_sess.run(self.w_int)

    self.assertEqual(1, len(wrapped_sess.observers["profiler_run_metadata"]))
    self.assertTrue(
        wrapped_sess.observers["profiler_run_metadata"][0].step_stats)
    self.assertEqual(1, len(wrapped_sess.observers["profiler_py_graphs"]))
    self.assertIsInstance(
        wrapped_sess.observers["profiler_py_graphs"][0], ops.Graph)

  def testCallingHookDelBeforeAnyRun(self):
    wrapped_sess = LocalCLIDebuggerWrapperSessionForTest(
        [["run"], ["run"]], self.sess)
    del wrapped_sess


if __name__ == "__main__":
  googletest.main()