rewriter_config = rewriter_config_pb2.RewriterConfig(

disable_model_pruning=True,

constant_folding=rewriter_config_pb2.RewriterConfig.OFF,

arithmetic_optimization=rewriter_config_pb2.RewriterConfig.OFF,

dependency_optimization=rewriter_config_pb2.RewriterConfig.OFF)

graph_options = config_pb2.GraphOptions(rewrite_options=rewriter_config)

"""Parse a line containing an op node followed by a node name.

op_type = line.strip().split(" ")[0].replace("[", "").replace("]", "")

# Not using [-1], to tolerate any other items that might be present behind

# the node name.

node_name = line.strip().split(" ")[1]

command,

reverse,

node_name_regex,

op_type_regex,

tensor_filter_name):

tst.assertFalse(reverse and "-r" in command)

tst.assertFalse(not(op_type_regex) and ("-t %s" % op_type_regex) in command)

tst.assertFalse(

not(node_name_regex) and ("-t %s" % node_name_regex) in command)

tst.assertFalse(

out,

expected_tensor_names,

expected_op_types,

node_name_regex=None,

op_type_regex=None,

tensor_filter_name=None,

sort_by="timestamp",

reverse=False):

"""Check RichTextLines output for list_tensors commands.

line_iter = iter(out.lines)

attr_segs = out.font_attr_segs

line_counter = 0

num_tensors = len(expected_tensor_names)

if tensor_filter_name is None:

tst.assertEqual("%d dumped tensor(s):" % num_tensors, next(line_iter))

else:

tst.assertEqual("%d dumped tensor(s) passing filter \"%s\":" %

(num_tensors, tensor_filter_name), next(line_iter))

line_counter += 1

if op_type_regex is not None:

tst.assertEqual("Op type regex filter: \"%s\"" % op_type_regex,

next(line_iter))

line_counter += 1

if node_name_regex is not None:

tst.assertEqual("Node name regex filter: \"%s\"" % node_name_regex,

next(line_iter))

line_counter += 1

tst.assertEqual("", next(line_iter))

line_counter += 1

# Verify the column heads "t (ms)", "Op type" and "Tensor name" are present.

line = next(line_iter)

tst.assertIn("t (ms)", line)

tst.assertIn("Op type", line)

tst.assertIn("Tensor name", line)

# Verify the command shortcuts in the top row.

attr_segs = out.font_attr_segs[line_counter]

attr_seg = attr_segs[0]

tst.assertEqual(0, attr_seg[0])

tst.assertEqual(len("t (ms)"), attr_seg[1])

command = attr_seg[2][0].content

tst.assertIn("-s timestamp", command)

assert_column_header_command_shortcut(

tst, command, reverse, node_name_regex, op_type_regex,

tensor_filter_name)

tst.assertEqual("bold", attr_seg[2][1])

idx0 = line.index("Size")

attr_seg = attr_segs[1]

tst.assertEqual(idx0, attr_seg[0])

tst.assertEqual(idx0 + len("Size (B)"), attr_seg[1])

command = attr_seg[2][0].content

tst.assertIn("-s dump_size", command)

assert_column_header_command_shortcut(tst, command, reverse, node_name_regex,

op_type_regex, tensor_filter_name)

tst.assertEqual("bold", attr_seg[2][1])

idx0 = line.index("Op type")

attr_seg = attr_segs[2]

tst.assertEqual(idx0, attr_seg[0])

tst.assertEqual(idx0 + len("Op type"), attr_seg[1])

command = attr_seg[2][0].content

tst.assertIn("-s op_type", command)

assert_column_header_command_shortcut(

tst, command, reverse, node_name_regex, op_type_regex,

tensor_filter_name)

tst.assertEqual("bold", attr_seg[2][1])

idx0 = line.index("Tensor name")

attr_seg = attr_segs[3]

tst.assertEqual(idx0, attr_seg[0])

tst.assertEqual(idx0 + len("Tensor name"), attr_seg[1])

command = attr_seg[2][0].content

tst.assertIn("-s tensor_name", command)

assert_column_header_command_shortcut(

tst, command, reverse, node_name_regex, op_type_regex,

tensor_filter_name)

tst.assertEqual("bold", attr_seg[2][1])

# Verify the listed tensors and their timestamps.

tensor_timestamps = []

dump_sizes_bytes = []

op_types = []

tensor_names = []

for line in line_iter:

items = line.split(" ")

items = [item for item in items if item]

rel_time = float(items[0][1:-1])

tst.assertGreaterEqual(rel_time, 0.0)

tensor_timestamps.append(rel_time)

dump_sizes_bytes.append(command_parser.parse_readable_size_str(items[1]))

op_types.append(items[2])

tensor_names.append(items[3])

# Verify that the tensors should be listed in ascending order of their

# timestamps.

if sort_by == "timestamp":

sorted_timestamps = sorted(tensor_timestamps)

if reverse:

sorted_timestamps.reverse()

tst.assertEqual(sorted_timestamps, tensor_timestamps)

elif sort_by == "dump_size":

sorted_dump_sizes_bytes = sorted(dump_sizes_bytes)

if reverse:

sorted_dump_sizes_bytes.reverse()

tst.assertEqual(sorted_dump_sizes_bytes, dump_sizes_bytes)

elif sort_by == "op_type":

sorted_op_types = sorted(op_types)

if reverse:

sorted_op_types.reverse()

tst.assertEqual(sorted_op_types, op_types)

elif sort_by == "tensor_name":

sorted_tensor_names = sorted(tensor_names)

if reverse:

sorted_tensor_names.reverse()

tst.assertEqual(sorted_tensor_names, tensor_names)

else:

tst.fail("Invalid value in sort_by: %s" % sort_by)

# Verify that the tensors are all listed.

for tensor_name, op_type in zip(expected_tensor_names, expected_op_types):

tst.assertIn(tensor_name, tensor_names)

index = tensor_names.index(tensor_name)

out,

node_name,

op_type,

device,

input_op_type_node_name_pairs,

ctrl_input_op_type_node_name_pairs,

recipient_op_type_node_name_pairs,

ctrl_recipient_op_type_node_name_pairs,

attr_key_val_pairs=None,

num_dumped_tensors=None,

show_stack_trace=False,

stack_trace_available=False):

"""Check RichTextLines output for node_info commands.

line_iter = iter(out.lines)

tst.assertEqual("Node %s" % node_name, next(line_iter))

tst.assertEqual("", next(line_iter))

tst.assertEqual(" Op: %s" % op_type, next(line_iter))

tst.assertEqual(" Device: %s" % device, next(line_iter))

tst.assertEqual("", next(line_iter))

tst.assertEqual(" %d input(s) + %d control input(s):" %

(len(input_op_type_node_name_pairs),

len(ctrl_input_op_type_node_name_pairs)), next(line_iter))

# Check inputs.

tst.assertEqual(" %d input(s):" % len(input_op_type_node_name_pairs),

next(line_iter))

for op_type, node_name in input_op_type_node_name_pairs:

tst.assertEqual(" [%s] %s" % (op_type, node_name), next(line_iter))

tst.assertEqual("", next(line_iter))

# Check control inputs.

if ctrl_input_op_type_node_name_pairs:

tst.assertEqual(" %d control input(s):" %

len(ctrl_input_op_type_node_name_pairs), next(line_iter))

for op_type, node_name in ctrl_input_op_type_node_name_pairs:

tst.assertEqual(" [%s] %s" % (op_type, node_name), next(line_iter))

tst.assertEqual("", next(line_iter))

tst.assertEqual(" %d recipient(s) + %d control recipient(s):" %

(len(recipient_op_type_node_name_pairs),

len(ctrl_recipient_op_type_node_name_pairs)),

next(line_iter))

# Check recipients, the order of which is not deterministic.

tst.assertEqual(" %d recipient(s):" %

len(recipient_op_type_node_name_pairs), next(line_iter))

t_recs = []

for _ in recipient_op_type_node_name_pairs:

line = next(line_iter)

op_type, node_name = parse_op_and_node(line)

t_recs.append((op_type, node_name))

tst.assertItemsEqual(recipient_op_type_node_name_pairs, t_recs)

# Check control recipients, the order of which is not deterministic.

if ctrl_recipient_op_type_node_name_pairs:

tst.assertEqual("", next(line_iter))

tst.assertEqual(" %d control recipient(s):" %

len(ctrl_recipient_op_type_node_name_pairs),

next(line_iter))

t_ctrl_recs = []

for _ in ctrl_recipient_op_type_node_name_pairs:

line = next(line_iter)

op_type, node_name = parse_op_and_node(line)

t_ctrl_recs.append((op_type, node_name))

tst.assertItemsEqual(ctrl_recipient_op_type_node_name_pairs, t_ctrl_recs)

# The order of multiple attributes can be non-deterministic.

if attr_key_val_pairs:

tst.assertEqual("", next(line_iter))

tst.assertEqual("Node attributes:", next(line_iter))

kv_pairs = []

for key, val in attr_key_val_pairs:

key = next(line_iter).strip().replace(":", "")

val = next(line_iter).strip()

kv_pairs.append((key, val))

tst.assertEqual("", next(line_iter))

tst.assertItemsEqual(attr_key_val_pairs, kv_pairs)

if num_dumped_tensors is not None:

tst.assertEqual("%d dumped tensor(s):" % num_dumped_tensors,

next(line_iter))

tst.assertEqual("", next(line_iter))

dump_timestamps_ms = []

for _ in xrange(num_dumped_tensors):

line = next(line_iter)

tst.assertStartsWith(line.strip(), "Slot 0 @ DebugIdentity @")

tst.assertTrue(line.strip().endswith(" ms"))

dump_timestamp_ms = float(line.strip().split(" @ ")[-1].replace("ms", ""))

tst.assertGreaterEqual(dump_timestamp_ms, 0.0)

dump_timestamps_ms.append(dump_timestamp_ms)

tst.assertEqual(sorted(dump_timestamps_ms), dump_timestamps_ms)

if show_stack_trace:

tst.assertEqual("", next(line_iter))

tst.assertEqual("", next(line_iter))

tst.assertEqual("Traceback of node construction:", next(line_iter))

if stack_trace_available:

try:

depth_counter = 0

while True:

for i in range(5):

line = next(line_iter)

if i == 0:

tst.assertEqual(depth_counter, int(line.split(":")[0]))

elif i == 1:

tst.assertStartsWith(line, " Line:")

elif i == 2:

tst.assertStartsWith(line, " Function:")

elif i == 3:

tst.assertStartsWith(line, " Text:")

elif i == 4:

tst.assertEqual("", line)

depth_counter += 1

except StopIteration:

tst.assertEqual(0, i)

else:

tst.assertEqual("(Unavailable because no Python graph has been loaded)",

"""Check RichTextLines output for valid command prefix but invalid syntax."""

tst.assertEqual([

"Syntax error for command: %s" % command_prefix,

"For help, do \"help %s\"" % command_prefix

"""Check RichTextLines output from invalid/erroneous commands.

tst.assertGreater(len(out.lines), 2)

tst.assertStartsWith(out.lines[0],

"Error occurred during handling of command: %s %s" %

out,

list_tensors_enabled=False,

node_info_node_name=None,

print_tensor_node_name=None,

list_inputs_node_name=None,

list_outputs_node_name=None):

"""Check the main menu annotation of an output."""

tst.assertIn(debugger_cli_common.MAIN_MENU_KEY, out.annotations)

menu = out.annotations[debugger_cli_common.MAIN_MENU_KEY]

tst.assertEqual(list_tensors_enabled,

menu.caption_to_item("list_tensors").is_enabled())

menu_item = menu.caption_to_item("node_info")

if node_info_node_name:

tst.assertTrue(menu_item.is_enabled())

tst.assertTrue(menu_item.content.endswith(node_info_node_name))

else:

tst.assertFalse(menu_item.is_enabled())

menu_item = menu.caption_to_item("print_tensor")

if print_tensor_node_name:

tst.assertTrue(menu_item.is_enabled())

tst.assertTrue(menu_item.content.endswith(print_tensor_node_name))

else:

tst.assertFalse(menu_item.is_enabled())

menu_item = menu.caption_to_item("list_inputs")

if list_inputs_node_name:

tst.assertTrue(menu_item.is_enabled())

tst.assertTrue(menu_item.content.endswith(list_inputs_node_name))

else:

tst.assertFalse(menu_item.is_enabled())

menu_item = menu.caption_to_item("list_outputs")

if list_outputs_node_name:

tst.assertTrue(menu_item.is_enabled())

tst.assertTrue(menu_item.content.endswith(list_outputs_node_name))

else:

tst.assertFalse(menu_item.is_enabled())

tst.assertTrue(menu.caption_to_item("run_info").is_enabled())

expected_command):

attr_segs = out.font_attr_segs[line_index]

found_menu_item = False

for begin, end, attribute in attr_segs:

attributes = [attribute] if not isinstance(attribute, list) else attribute

menu_item = [attribute for attribute in attributes if

isinstance(attribute, debugger_cli_common.MenuItem)]

if menu_item:

tst.assertEqual(expected_begin, begin)

tst.assertEqual(expected_end, end)

tst.assertEqual(expected_command, menu_item[0].content)

found_menu_item = True

break

"""Create an analyzer CLI.

# Construct the analyzer.

analyzer = analyzer_cli.DebugAnalyzer(dump, _cli_config_from_temp_file())

# Construct the handler registry.

registry = debugger_cli_common.CommandHandlerRegistry()

# Register command handlers.

registry.register_command_handler(

"list_tensors",

analyzer.list_tensors,

analyzer.get_help("list_tensors"),

prefix_aliases=["lt"])

registry.register_command_handler(

"node_info",

analyzer.node_info,

analyzer.get_help("node_info"),

prefix_aliases=["ni"])

registry.register_command_handler(

"list_inputs",

analyzer.list_inputs,

analyzer.get_help("list_inputs"),

prefix_aliases=["li"])

registry.register_command_handler(

"list_outputs",

analyzer.list_outputs,

analyzer.get_help("list_outputs"),

prefix_aliases=["lo"])

registry.register_command_handler(

"print_tensor",

analyzer.print_tensor,

analyzer.get_help("print_tensor"),

prefix_aliases=["pt"])

registry.register_command_handler(

"print_source",

analyzer.print_source,

analyzer.get_help("print_source"),

prefix_aliases=["ps"])

registry.register_command_handler(

"list_source",

analyzer.list_source,

analyzer.get_help("list_source"),

prefix_aliases=["ls"])

registry.register_command_handler(

"eval",

analyzer.evaluate_expression,

analyzer.get_help("eval"),

prefix_aliases=["ev"])

@classmethod

def setUpClass(cls):

cls._dump_root = tempfile.mkdtemp()

cls._dump_root_for_unique = tempfile.mkdtemp()

cls._is_gpu_available = test.is_gpu_available()

if cls._is_gpu_available:

gpu_name = test_util.gpu_device_name()

cls._main_device = "/job:localhost/replica:0/task:0" + gpu_name

else:

cls._main_device = "/job:localhost/replica:0/task:0/device:CPU:0"

cls._curr_file_path = os.path.abspath(

tf_inspect.getfile(tf_inspect.currentframe()))

cls._sess = session.Session(config=no_rewrite_session_config())

with cls._sess as sess:

u_init_val = np.array([[5.0, 3.0], [-1.0, 0.0]])

v_init_val = np.array([[2.0], [-1.0]])

u_name = "simple_mul_add/u"

v_name = "simple_mul_add/v"

u_init = constant_op.constant(u_init_val, shape=[2, 2], name="u_init")

u = variables.Variable(u_init, name=u_name)

cls._u_line_number = line_number_above()

v_init = constant_op.constant(v_init_val, shape=[2, 1], name="v_init")

v = variables.Variable(v_init, name=v_name)

cls._v_line_number = line_number_above()

w = math_ops.matmul(u, v, name="simple_mul_add/matmul")

cls._w_line_number = line_number_above()

x = math_ops.add(w, w, name="simple_mul_add/add")

cls._x_line_number = line_number_above()

a = variables.Variable([1, 3, 3, 7], name="a")

u.initializer.run()

v.initializer.run()

a.initializer.run()

run_options = config_pb2.RunOptions(output_partition_graphs=True)

debug_utils.watch_graph(

run_options,

sess.graph,

debug_ops=["DebugIdentity"],

debug_urls="file://%s" % cls._dump_root)

# Invoke Session.run().

run_metadata = config_pb2.RunMetadata()

sess.run([x], options=run_options, run_metadata=run_metadata)

cls._debug_dump = debug_data.DebugDumpDir(

cls._dump_root, partition_graphs=run_metadata.partition_graphs)

cls._analyzer, cls._registry = create_analyzer_cli(cls._debug_dump)

@classmethod

def tearDownClass(cls):

# Tear down temporary dump directory.

shutil.rmtree(cls._dump_root)

shutil.rmtree(cls._dump_root_for_unique)

def testMeasureTensorListColumnWidthsGivesRightAnswerForEmptyData(self):

timestamp_col_width, dump_size_col_width, op_type_col_width = (

self._analyzer._measure_tensor_list_column_widths([]))

self.assertEqual(len("t (ms)") + 1, timestamp_col_width)

self.assertEqual(len("Size (B)") + 1, dump_size_col_width)

self.assertEqual(len("Op type") + 1, op_type_col_width)

def testMeasureTensorListColumnWidthsGivesRightAnswerForData(self):

dump = self._debug_dump.dumped_tensor_data[0]

self.assertLess(dump.dump_size_bytes, 1000)

self.assertEqual(

"VariableV2", self._debug_dump.node_op_type(dump.node_name))

_, dump_size_col_width, op_type_col_width = (

self._analyzer._measure_tensor_list_column_widths([dump]))

# The length of str(dump.dump_size_bytes) is less than the length of

# "Size (B)" (8). So the column width should be determined by the length of

# "Size (B)".

self.assertEqual(len("Size (B)") + 1, dump_size_col_width)

# The length of "VariableV2" is greater than the length of "Op type". So the

# column should be determined by the length of "VariableV2".

self.assertEqual(len("VariableV2") + 1, op_type_col_width)

def testListTensors(self):

# Use shorthand alias for the command prefix.

out = self._registry.dispatch_command("lt", [])

assert_listed_tensors(self, out, [

"simple_mul_add/u:0", "simple_mul_add/v:0", "simple_mul_add/u/read:0",

"simple_mul_add/v/read:0", "simple_mul_add/matmul:0",

"simple_mul_add/add:0"

], ["VariableV2", "VariableV2", "Identity", "Identity", "MatMul", "Add"])

# Check the main menu.

check_main_menu(self, out, list_tensors_enabled=False)

def testListTensorsInReverseTimeOrderWorks(self):

# Use shorthand alias for the command prefix.

out = self._registry.dispatch_command("lt", ["-s", "timestamp", "-r"])

assert_listed_tensors(

self,

out, [

"simple_mul_add/u:0", "simple_mul_add/v:0",

"simple_mul_add/u/read:0", "simple_mul_add/v/read:0",

"simple_mul_add/matmul:0", "simple_mul_add/add:0"

],

["VariableV2", "VariableV2", "Identity", "Identity", "MatMul", "Add"],

sort_by="timestamp",

reverse=True)

check_main_menu(self, out, list_tensors_enabled=False)

def testListTensorsInDumpSizeOrderWorks(self):

out = self._registry.dispatch_command("lt", ["-s", "dump_size"])

assert_listed_tensors(

self,

out, [

"simple_mul_add/u:0", "simple_mul_add/v:0",

"simple_mul_add/u/read:0", "simple_mul_add/v/read:0",

"simple_mul_add/matmul:0", "simple_mul_add/add:0"

],

["VariableV2", "VariableV2", "Identity", "Identity", "MatMul", "Add"],

sort_by="dump_size")

check_main_menu(self, out, list_tensors_enabled=False)

def testListTensorsInReverseDumpSizeOrderWorks(self):

out = self._registry.dispatch_command("lt", ["-s", "dump_size", "-r"])

assert_listed_tensors(

self,

out, [

"simple_mul_add/u:0", "simple_mul_add/v:0",

"simple_mul_add/u/read:0", "simple_mul_add/v/read:0",

"simple_mul_add/matmul:0", "simple_mul_add/add:0"

],

["VariableV2", "VariableV2", "Identity", "Identity", "MatMul", "Add"],

sort_by="dump_size",

reverse=True)

check_main_menu(self, out, list_tensors_enabled=False)

def testListTensorsWithInvalidSortByFieldGivesError(self):

out = self._registry.dispatch_command("lt", ["-s", "foobar"])

self.assertIn("ValueError: Unsupported key to sort tensors by: foobar",

out.lines)

def testListTensorsInOpTypeOrderWorks(self):

# Use shorthand alias for the command prefix.

out = self._registry.dispatch_command("lt", ["-s", "op_type"])

assert_listed_tensors(

self,

out, [

"simple_mul_add/u:0", "simple_mul_add/v:0",

"simple_mul_add/u/read:0", "simple_mul_add/v/read:0",

"simple_mul_add/matmul:0", "simple_mul_add/add:0"

],

["VariableV2", "VariableV2", "Identity", "Identity", "MatMul", "Add"],

sort_by="op_type",

reverse=False)

check_main_menu(self, out, list_tensors_enabled=False)

def testListTensorsInReverseOpTypeOrderWorks(self):

# Use shorthand alias for the command prefix.

out = self._registry.dispatch_command("lt", ["-s", "op_type", "-r"])

assert_listed_tensors(

self,

out, [

"simple_mul_add/u:0", "simple_mul_add/v:0",

"simple_mul_add/u/read:0", "simple_mul_add/v/read:0",

"simple_mul_add/matmul:0", "simple_mul_add/add:0"

],

["VariableV2", "VariableV2", "Identity", "Identity", "MatMul", "Add"],

sort_by="op_type",

reverse=True)

check_main_menu(self, out, list_tensors_enabled=False)

def testListTensorsInTensorNameOrderWorks(self):

# Use shorthand alias for the command prefix.

out = self._registry.dispatch_command("lt", ["-s", "tensor_name"])

assert_listed_tensors(

self,

out, [

"simple_mul_add/u:0", "simple_mul_add/v:0",

"simple_mul_add/u/read:0", "simple_mul_add/v/read:0",

"simple_mul_add/matmul:0", "simple_mul_add/add:0"

],

["VariableV2", "VariableV2", "Identity", "Identity", "MatMul", "Add"],

sort_by="tensor_name",

reverse=False)

check_main_menu(self, out, list_tensors_enabled=False)

def testListTensorsInReverseTensorNameOrderWorks(self):

# Use shorthand alias for the command prefix.

out = self._registry.dispatch_command("lt", ["-s", "tensor_name", "-r"])

assert_listed_tensors(

self,

out, [

"simple_mul_add/u:0", "simple_mul_add/v:0",

"simple_mul_add/u/read:0", "simple_mul_add/v/read:0",

"simple_mul_add/matmul:0", "simple_mul_add/add:0"

],

["VariableV2", "VariableV2", "Identity", "Identity", "MatMul", "Add"],

sort_by="tensor_name",

reverse=True)

check_main_menu(self, out, list_tensors_enabled=False)

def testListTensorsFilterByNodeNameRegex(self):

out = self._registry.dispatch_command("list_tensors",

["--node_name_filter", ".*read.*"])

assert_listed_tensors(

self,

out, ["simple_mul_add/u/read:0", "simple_mul_add/v/read:0"],

["Identity", "Identity"],

node_name_regex=".*read.*")

out = self._registry.dispatch_command("list_tensors", ["-n", "^read"])

assert_listed_tensors(self, out, [], [], node_name_regex="^read")

check_main_menu(self, out, list_tensors_enabled=False)

def testListTensorFilterByOpTypeRegex(self):

out = self._registry.dispatch_command("list_tensors",

["--op_type_filter", "Identity"])

assert_listed_tensors(

self,

out, ["simple_mul_add/u/read:0", "simple_mul_add/v/read:0"],

["Identity", "Identity"],

op_type_regex="Identity")

out = self._registry.dispatch_command("list_tensors",

["-t", "(Add|MatMul)"])

assert_listed_tensors(

self,

out, ["simple_mul_add/add:0", "simple_mul_add/matmul:0"],

["Add", "MatMul"],

op_type_regex="(Add|MatMul)")

check_main_menu(self, out, list_tensors_enabled=False)

def testListTensorFilterByNodeNameRegexAndOpTypeRegex(self):

out = self._registry.dispatch_command(

"list_tensors", ["-t", "(Add|MatMul)", "-n", ".*add$"])

assert_listed_tensors(

self,

out, ["simple_mul_add/add:0"], ["Add"],

node_name_regex=".*add$",

op_type_regex="(Add|MatMul)")

check_main_menu(self, out, list_tensors_enabled=False)

def testListTensorsFilterNanOrInf(self):

"""Test register and invoke a tensor filter."""

# First, register the filter.

self._analyzer.add_tensor_filter("has_inf_or_nan",

debug_data.has_inf_or_nan)

# Use shorthand alias for the command prefix.

out = self._registry.dispatch_command("lt", ["-f", "has_inf_or_nan"])

# This TF graph run did not generate any bad numerical values.

assert_listed_tensors(

self, out, [], [], tensor_filter_name="has_inf_or_nan")

# TODO(cais): A test with some actual bad numerical values.

check_main_menu(self, out, list_tensors_enabled=False)

def testListTensorNonexistentFilter(self):

"""Test attempt to use a nonexistent tensor filter."""

out = self._registry.dispatch_command("lt", ["-f", "foo_filter"])

self.assertEqual(["ERROR: There is no tensor filter named \"foo_filter\"."],

out.lines)

check_main_menu(self, out, list_tensors_enabled=False)

def testListTensorsInvalidOptions(self):

out = self._registry.dispatch_command("list_tensors", ["--bar"])

check_syntax_error_output(self, out, "list_tensors")

def testNodeInfoByNodeName(self):

node_name = "simple_mul_add/matmul"

out = self._registry.dispatch_command("node_info", [node_name])

recipients = [("Add", "simple_mul_add/add"), ("Add", "simple_mul_add/add")]

assert_node_attribute_lines(self, out, node_name, "MatMul",

self._main_device,

[("Identity", "simple_mul_add/u/read"),

("Identity", "simple_mul_add/v/read")], [],

recipients, [])

check_main_menu(

self,

out,

list_tensors_enabled=True,

list_inputs_node_name=node_name,

print_tensor_node_name=node_name,

list_outputs_node_name=node_name)

# Verify that the node name is bold in the first line.

self.assertEqual(

[(len(out.lines[0]) - len(node_name), len(out.lines[0]), "bold")],

out.font_attr_segs[0])

def testNodeInfoShowAttributes(self):

node_name = "simple_mul_add/matmul"

out = self._registry.dispatch_command("node_info", ["-a", node_name])

assert_node_attribute_lines(

self,

out,

node_name,

"MatMul",

self._main_device, [("Identity", "simple_mul_add/u/read"),

("Identity", "simple_mul_add/v/read")], [],

[("Add", "simple_mul_add/add"), ("Add", "simple_mul_add/add")], [],

attr_key_val_pairs=[("transpose_a", "b: false"),

("transpose_b", "b: false"),

("T", "type: DT_DOUBLE")])

check_main_menu(

self,

out,

list_tensors_enabled=True,

list_inputs_node_name=node_name,

print_tensor_node_name=node_name,

list_outputs_node_name=node_name)

def testNodeInfoShowDumps(self):

node_name = "simple_mul_add/matmul"

out = self._registry.dispatch_command("node_info", ["-d", node_name])

assert_node_attribute_lines(

self,

out,

node_name,

"MatMul",

self._main_device, [("Identity", "simple_mul_add/u/read"),

("Identity", "simple_mul_add/v/read")], [],

[("Add", "simple_mul_add/add"), ("Add", "simple_mul_add/add")], [],

num_dumped_tensors=1)

check_main_menu(

self,

out,

list_tensors_enabled=True,

list_inputs_node_name=node_name,

print_tensor_node_name=node_name,

list_outputs_node_name=node_name)

check_menu_item(self, out, 16,

len(out.lines[16]) - len(out.lines[16].strip()),

len(out.lines[16]), "pt %s:0 -n 0" % node_name)

def testNodeInfoShowStackTraceUnavailableIsIndicated(self):

self._debug_dump.set_python_graph(None)

node_name = "simple_mul_add/matmul"

out = self._registry.dispatch_command("node_info", ["-t", node_name])

assert_node_attribute_lines(

self,

out,

node_name,

"MatMul",

self._main_device, [("Identity", "simple_mul_add/u/read"),

("Identity", "simple_mul_add/v/read")], [],

[("Add", "simple_mul_add/add"), ("Add", "simple_mul_add/add")], [],

show_stack_trace=True, stack_trace_available=False)

check_main_menu(

self,

out,

list_tensors_enabled=True,

list_inputs_node_name=node_name,

print_tensor_node_name=node_name,

list_outputs_node_name=node_name)

def testNodeInfoShowStackTraceAvailableWorks(self):

self._debug_dump.set_python_graph(self._sess.graph)

node_name = "simple_mul_add/matmul"

out = self._registry.dispatch_command("node_info", ["-t", node_name])

assert_node_attribute_lines(

self,

out,

node_name,

"MatMul",

self._main_device, [("Identity", "simple_mul_add/u/read"),

("Identity", "simple_mul_add/v/read")], [],

[("Add", "simple_mul_add/add"), ("Add", "simple_mul_add/add")], [],

show_stack_trace=True, stack_trace_available=True)

check_main_menu(

self,

out,

list_tensors_enabled=True,

list_inputs_node_name=node_name,

print_tensor_node_name=node_name,

list_outputs_node_name=node_name)

def testNodeInfoByTensorName(self):

node_name = "simple_mul_add/u/read"

tensor_name = node_name + ":0"

out = self._registry.dispatch_command("node_info", [tensor_name])

assert_node_attribute_lines(self, out, node_name, "Identity",

self._main_device,

[("VariableV2", "simple_mul_add/u")], [],

[("MatMul", "simple_mul_add/matmul")], [])

check_main_menu(

self,

out,

list_tensors_enabled=True,

list_inputs_node_name=node_name,

print_tensor_node_name=node_name,

list_outputs_node_name=node_name)

def testNodeInfoNonexistentNodeName(self):

out = self._registry.dispatch_command("node_info", ["bar"])

self.assertEqual(

["ERROR: There is no node named \"bar\" in the partition graphs"],

out.lines)

# Check color indicating error.

self.assertEqual({0: [(0, 59, cli_shared.COLOR_RED)]}, out.font_attr_segs)

check_main_menu(self, out, list_tensors_enabled=True)

def testPrintTensor(self):

node_name = "simple_mul_add/matmul"

tensor_name = node_name + ":0"

out = self._registry.dispatch_command(

"print_tensor", [tensor_name], screen_info={"cols": 80})